問題タブ [reactive-streams]

以下のコード スニペットが実行している理由を理解しようとしています。シンクはソースがコンテンツを生成するよりも速く要求を生成できないため、一部のオファー (オーバーフロー戦略が [ドロップ バッファー] に設定されている) に応答してメッセージがドロップされ、エラーとキューが閉じられたというメッセージが表示されると考えていました。自爆ピースの後。

スニペット:

出力：

私の誤解は、QueueSource クラスでのgetAsyncCallbackの使用に関するものだと思います。QueueSource のオファー呼び出しが正しいオファーの詳細で stageLogic を呼び出しても、ステージ ロジック内のこのコードの実際のハンドラーは、前の要素の処理が完了するまで呼び出されないため、バッファー サイズをチェックしたり、オーバーフローを適用したりするためのロジックはありません。戦略が適用されています... :-/

反応ストリーム サブスクライバーを akka ソースにアタッチしようとしています。

私のソースは単純なシンク (foreach など) で問題なく動作するようですが、サブスクライバーから作成された実際のシンクを挿入すると、何も得られません。

私のコンテキストは次のとおりです。

私のテストケースは次のとおりです。

出力が得られます：

xxx、yyy、zzz が表示されないのはなぜですか?

私は Java 9 の新機能のいくつかを試していました。そこで、特定のレートで数値を発行するパブリッシャーを用意するためのテストをまとめました。また、これらのパブリケーションをリッスンしてコンソールに出力するためのサブスクライバーも実装しました。

onNext()メソッドは何も出力せず、getLastItem()0 しか返さないため、この Api の使用方法を完全には理解していない可能性があります。

機能しているように見える唯一の部分は、変数onSubscribe()を正しく初期化する部分です。lastItem

誰かが私のテストで何が間違っているのか教えてもらえますか? テストがこれらの数字を出力し、最後の項目として 5 を返すことを期待します。

Observables と Subjects は Angular2 でのみ使用していると言わざるを得ませんが、理解しやすいようです。