如何将hot observable和single变成cold observable答案

【问题标题】：How to turn hot observable and single into cold observable如何将hot observable和single变成cold observable
【发布时间】：2019-02-26 16:20:32
【问题描述】：

我尝试为 NuProcess 制作 rx 包装器，用于异步执行外部进程的库。

这里的主要类 - 与进程的“通信”。在这里我阅读了标准输出：

static class MyProcessHandler extends NuAbstractProcessHandler {

    final PublishSubject<String> stdout = PublishSubject.create();

    @Override
    public void onStdout(ByteBuffer buffer, boolean closed) {
        if (!closed) {
            byte[] bytes = new byte[buffer.remaining()];
            buffer.get(bytes);
            stdout.onNext(new String(bytes));
        }
    }

    @Override
    public void onExit(int statusCode) {
        if (statusCode != 0)
            stdout.onError(new RuntimeException("non zero code"));
        else
            stdout.onComplete();
    }
}

这是我如何开始的过程：

static class Streams {

    RxProcessHandler handler = new RxProcessHandler();

    Single<Integer> waitDone(long timeout, TimeUnit timeUnit) {
        return Single.create(emitter -> {
            NuProcessBuilder b = new NuProcessBuilder("some cmd");
            b.setProcessListener(handler);
            NuProcess process = b.start();
            emitter.setCancellable(() -> process.destroy(true));
            int code = process.waitFor(timeout, timeUnit);
            emitter.onSuccess(code);
        });
    }

    public PublishSubject<String> stdOut() {
        return handler.stdout;
    }

}

最后是我的 api。如您所见，这里有三个变体：

1 - 等待过程结束
2 - 与标准输出回调相同
3 - 读取标准输出直到进程结束。 onComplete 表示零退出代码，错误 - 非零退出代码。 subscribe() 应该开始进程。

我不知道如何实现 3d 变体。

static class PublicApi {

    //just wait process ends
    public Single<Integer> asWaitDone(long timeout, TimeUnit timeUnit) {
        return new Streams().waitDone(timeout, timeUnit);
    }

    //wait process ends and have stdout callback
    public Pair<Single<Integer>, Observable<String>> asWaitDoneWithStdout(long timeout, TimeUnit timeUnit) {
        Streams streams = new Streams();
        return new ImmutablePair(streams.waitDone(timeout, timeUnit), streams.stdOut());
    }

    //read stdout until process ends
    public Observable<String> asStdout(long timeout, TimeUnit timeUnit) {
        return ???
    }

}

【问题讨论】：

这毫无意义。为什么不简单地使用intervalRange 来获得 10 个号码，只要您有订阅者？
因为它是“假”实现，只是为了提问，所以有人可以重现它。有了它，我展示了我的合同“PublishSubject + Single =???=> Cold Observable”
您对问题的描述越详细，您得到正确答案的可能性就越大。 PublishSubject + Single -> Cold 没有任何意义。您可以通过按需创建它们来将热源变冷，但您的示例并不意味着您首先要这样做。请描述您的原始需求是什么，而不是您认为可以通过 PublishSubject + Single -> Cold 解决。
现在完全重写问题。

标签： rx-java reactive-programming rx-java2

【解决方案1】：

您可以重新排列现有的命令集并将它们变成一个 Observable

static final class MyProcessHandlerObservable extends NuAbstractProcessHandler {

    final ObservableEmitter<String> emitter;

    MyProcessHandler(ObservableEmitter<String> emitter) {
        this.emitter = emitter;
    }

    @Override
    public void onStdout(ByteBuffer buffer, boolean closed) {
        if (!closed) {
            byte[] bytes = new byte[buffer.remaining()];
            buffer.get(bytes);
            emitter.onNext(new String(bytes));
        }
    }

    @Override
    public void onExit(int statusCode) {
        if (statusCode != 0) {
            stdout.onError(new RuntimeException("non zero code: " + statusCode));
        } else {
            stdout.onComplete();
        }
    }
}

public Observable<String> asStdout(long timeout, TimeUnit timeUnit) {
    return Observable.create(emitter -> {
        MyProcessHandlerObservable handler = new MyProcessHandlerObservable(emitter);
        NuProcessBuilder b = new NuProcessBuilder("some cmd");
        b.setProcessListener(handler);
        NuProcess process = b.start();
        emitter.setCancellable(() -> process.destroy(true));
    })
    .takeUntil(Observable.timer(timeout, timeUnit).map(v -> throw new TimeoutException()));
}

【讨论】：