問題タブ [rx-kotlin]

次のようなデータフローがあります。

ビジネス ロジック (実際には s を返すFuture) をに置き換えましたCompletableFuture.supplyAsync。はい、これは Kotlin ですが、意図は理解できたと思います。

「デッド」値 ( 57005, 0xDEAD) をコメントすると、出力は次のようになります。

しかし、その「死んだ」値がストリームに現れると、失敗します:

私はRXの初心者なので、すぐに解決策をグーグルで検索しました: onExceptionResumeNext: Observable.fromFuture(it)--> Observable.fromFuture(it).onExceptionResumeNext { Observable.empty<Int>() }. しかし、私のアプリケーションは永遠にハングアップします (私が期待する出力を生成した後)。ストリームは決して終わらないようです。

Observableどういうわけかそれを「シャットダウン」する必要がありますか？それとも、より一般的には、RX を使用する場合の適切なアプローチですか? 別の方法で考え直す必要がありますか？

これまで解決できなかった問題があります。RxKotlin は初めてなので、簡単かもしれません。コードを見てください：

ストリームは自作のコレクションです。Map は、そのコレクション内のすべてのアイテムを反復処理できるようにするメソッドです。

ここでの問題は、

理想的ではない Single を返す http 呼び出しです。

Single ではなく Item を返し、それを ClientInfo に渡す非同期呼び出しをマップ内に作成したいと思います。私がこれまでに試したことは、マップ内でサブスクライブを使用し、blockingGet() メソッドを使用することでしたが、これは別のスレッドで観察してサブスクライブしてもメインスレッドをブロックします

したがって、コレクション内のすべてのものに対して非同期呼び出しを行う必要があります。

手伝ってくれてありがとう

Spring-context 5.0.0.RELEASE で Kotlin 1.1.4-3 を使用しています。

プロジェクトを開始すると、次のエラーが発生します。

春はクラスを探しているようですrx.Schedulerが、クラスパスにはありません。の対応する依存関係がrx.Scheduler追加されていないことは明らかです。しかし、インターネット上で関連するものを見つけることができませんでした。

RxJava2( io.reactivex.rxjava2) をプロジェクトに追加しましたが、うまくいきませんでした。私に何ができる？

最終的な回答 - Rafal G. の回答に基づく:

このエラーは、konrad-kaminski/spring-kotlin-coroutineライブラリを使用しているためです。ただし、プロジェクトでこれらの依存関係を使用しました。

理解するのに苦労している奇妙な問題に遭遇しました。callable から observable を作成するコードをいくつか書きました。正常にコンパイルされますが、スケジューラを指定するとすぐに戻り値の型が変更され、コンパイルされません。

subscribeOn を使用しないコードは次のとおりです (コンパイルされます)。

subscribeOn を使用したコードは次のとおりです (コンパイルされません)。

コンパイル時のエラー メッセージは次のとおりです。

確かに、スケジューラーを指定しても、返されるタイプが変更されるべきではありませんか? 何か案は？