java - FutureCallback 到 Observable

Question

有没有办法从 FutureCallback 创建 Observable ？我发现 Observable.create 已被弃用，并且不是正确的方法(如下)。

我正在转换

Observable.create(new Observable.OnSubscribe<HttpResponse>() {
        @Override
        public void call(Subscriber<? super HttpResponse> subscriber) {
            getClient().execute(httpRequest, new FutureCallback<HttpResponse>() {
                @Override
                public void completed(HttpResponse response) {
                    subscriber.onNext(response);
                    subscriber.onCompleted();
                }
                @Override
                public void failed(Exception ex) {
                    subscriber.onError(ex);
                }
                @Override
                public void cancelled() {
                    subscriber.onError(new Exception());
                }
            });
        }
})

Answer 1

使用其他创建 overload :

Observable.<Event>create(emitter -> {
    Callback listener = new Callback() {
        @Override
        public void onEvent(Event e) {
            emitter.onNext(e);
            if (e.isLast()) {
                emitter.onCompleted();
            }
        }

        @Override
        public void onFailure(Exception e) {
            emitter.onError(e);
        }
    };

    AutoCloseable c = api.someMethod(listener);

    emitter.setCancellation(c::close);

}, BackpressureMode.BUFFER);