使用 RxJava 2 的多订阅者和数据流策略

Question

我无法决定如何使用RxJava2 正确执行此任务。

问题如下。我正在使用AuidoRecord 录制音频。

目前我已经实现了这样的自定义Flowable 类

private class StreamAudioRecordRunnable extends Flowable<short[]> implements Runnable {
    private int mShortBufferSize;
    private List<Subscriber<? super short[]>> mSubscribers = new ArrayList<>();
    private short[] mAudioShortBuffer;

    private void removeAllNullableSubscribers() {
        mSubscribers.removeAll(Collections.singleton(null));
    }

    private void notifyAllSubscribers(short[] audioBuffer) {
        removeAllNullableSubscribers();
        for (Subscriber<? super short[]> subscriber : mSubscribers) {
            subscriber.onNext(audioBuffer);
        }
    }

    @Override
    protected void subscribeActual(Subscriber<? super short[]> newSubscriber) {
        mSubscribers.add(newSubscriber);
    }

    private void notifyAllSubscribersAboutError(Throwable error) {
        for (Subscriber<? super short[]> subscriber : mSubscribers) {
            subscriber.onError(error);
        }
    }

    @Override
    public void run() {
            // Init stuff
            while (mIsRecording.get()) {
                int ret;
                ret = mAudioRecord.read(mAudioShortBuffer, 0, mShortBufferSize);                              
                notifyAllSubscribers(mAudioShortBuffer);
             }
        mAudioRecord.release();
    }
}

如您所见，我正在手动将订阅者添加到列表中。然后，当我获得新缓冲区时，所有订阅者都会收到通知。

我猜这不是最高效的方法。

我需要什么

1. 至于这个在服务中运行的 flowable。即使没有订阅者，它也应该一直运行到服务处于活动状态。
2. 订阅者不是固定的，他们可以订阅然后取消订阅，但 Flowable/Observable 应该仍在运行。
3. 由于 Flowable 发出的数据是流，因此不应通知订阅者已发出的项目，他们应该只获取当前的流数据。开火即忘。
4. 即使所有订阅者都消失了，Flowable 也应该运行。

请提出正确的实施策略。 如果有任何帮助，我将不胜感激。

Answer 1

类似

public class StreamAudioRecordRunnable  {
    private int mShortBufferSize;
    private short[] mAudioShortBuffer;
    private ConnectedFlowable<short[]> audioFlowable();

    public StreamAudioRecordRunnable() {
        audioFlowable = Flowable.create(new ObservableOnSubscribe<short[]>() {
            @Override
            public void subscribe(FlowableEmitter<short[]> emitter) throws Exception {
                try {
                    while (mIsRecording.get()) {
                        int ret;
                        ret = mAudioRecord.read(mAudioShortBuffer, 0, mShortBufferSize);                              
                        emitter.onNext(mAudioShortBuffer);
                    }
                    emitter.onComplete();
                    mAudioRecord.release();
                } catch (Exception e) {
                    emitter.onError(e);
                    mAudioRecord.release();
                }
            }
        }).subscribeOn(Schedulers.io()).publish();  
    }


    public Flowable<short[]> getFlowable() {
            return audioFlowable.hide();
    }

    @Override
    public void start() {
        audioObservable.connect();      
    }
}

会是我的偏好。