Как превратить горячую наблюдаемую и одиночную в холодную наблюдаемую - PullRequest
Новая
       36

Как превратить горячую наблюдаемую и одиночную в холодную наблюдаемую

0 голосов
/ 25 февраля 2019

Пытаюсь сделать обертку rx для NuProcess, библиотеки для асинхронного выполнения внешних процессов.

Здесь основной класс - «связь» с процессом.Вот читаю stdout: 

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 - то же самое с обратным вызовом stdout
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 ???
    }

}

1 Ответ

0 голосов
/ 26 февраля 2019

Вы можете переставить существующий набор команд и превратить их в 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()));
}
...