問題タブ [rx-java2]

メソッドを呼び出してもどこにObervable.subscribe(Action1)もスローされないことを達成したいのですが、を呼び出すと、通常どおりエラーが発生したときに2番目のアクションが呼び出されます。私は2つの方法を試しました：OnErrorNotImplementedExceptionObervable.subscribe(Action1, Action1)

この方法OnErrorNotImplementedExceptionはスローされませんが、2 番目のアクションも渡すと、アクションも呼び出されません。

2番：

これの問題observeOn()は、後で呼び出された場合、これは非同期になり、ここでの例外処理は明らかに機能しません。

これを達成する方法はありますか。をsubscribe()投げないメソッドがあればいいのにと思います。OnErrorNotImplementedExceptiononError

RxJava 1 には CompositeSubscription がありましたが、RxJava2 にはそれがなく、rxJava2 には CompositeDisposable というものがあります。RxJava2 で CompositeDisposable または Disposable を使用するにはどうすればよいですか?

Observable ストリームがあり、それを Completable に変換したいのですが、どうすればそれを行うことができますか?

ドクター曰く

概念的には、これは Single と Completable の結合であり、0 または 1 のアイテム、または何らかのリアクティブ ソースによって通知されたエラーが存在する可能性のある放出パターンをキャプチャする手段を提供します。

しかし、それが本当に何を意味するのかはわかりません。java8 のようOptionalです。

Maybe次の 2 つのコードは結果は同じですが、何ができて何がOptionalできない (または面倒な)のかわかりません。

結果は同じです:

rxJava1 で、自分の API が返しObservable<Optional<T>>ていたのは、 悪臭ですか? に変更する必要がありObservable<Maybe<T>>ますか?

私のコードは次のとおりです。

そのコードは正しいですか。PS: Android Studio 2 を使用していますが、ラムダ式を使用するにはどうすれば設定できますか?

私の RestClient コンストラクター:

build.gradle:

jackOptions と compileOptions を追加することで、Lambda 構文が機能するようになりました。

新しい Location を送信できるを返す LocationHandler クラスを作成しようとしていますobservable<Location>。サブスクライバーは最後に追加されたものと後続の値を取得します。

私はこのクラスを書きましたが、それは機能しますが、コールバックを追加したので、それが正しい方法であるかどうかはわかりません。

助けてくれてありがとう。

@Blackbelt のアドバイスに従って、ReplaySubject で変更しました。

Observable<String>いくつかの作業を実行する があります。完了後、接続を閉じます (経由で) .setDisposable(Disposables.fromAction(logic*);。問題は、このクローズ アクションを実際のワークロードと同じスケジューラで実行する必要があることです。

出来ますか？どのように？

Observable を特定のスケジューラで実行するように強制したく myObservable.subscribeOn(x).unsubscribeOn(x);ありません。

org.reactivestreams.ProcessorRxJava 2.0 で使用したい があります。ただし、org.reactivestreams.Publisherを RxJavaと統合するための変換はありますが(または)io.reactivex.Flowable#fromPublisherを最適に統合する方法は明確ではありません。誰でもこれに光を当てることができますか？org.reactivestreams.Processororg.reactivestreams.Subscriber