問題タブ [akka-stream]

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

したがって、このコードでは、派生型 T1 または T2 の場合、基本型 A のメッセージを受け取ります。ただし、タイプが T3 の場合は、それを捨てて、lastStage に渡さずに、次のメッセージに進みたいだけです。どうすればいいですか？

このライブラリを使用して、ウィキペディアの IRC チャネルを継続的に読み取ろうとしています: https://github.com/implydata/wikiticker

カスタムの Akka Publisher を作成しました。これは、システムでSource.

ここに私のクラスのいくつかがあります：

私はこのすべてのオブジェクトを次のように作成しています:

だから私が直面している問題はこのSourceオブジェクトです。この実装は他のソース (ローカル ファイルなど) とうまく連携しますが、ActorPublisher はRequest()メッセージを受信しません。

マークされた 2 行にコメントすると、アクターがフローから Request(count) メッセージを受信したことがわかります。そうしないと、すべてのメッセージがキューにプッシュされますが、フローにはプッシュされません (したがって、MSG メッセージが出力されていることがわかります)。

ここではマルチスレッド/同期を使用していると思います。

Akka Streams ライブラリには、すでに非常に豊富なドキュメントが付属しています。しかし、私にとっての主な問題は、それがあまりにも多くの資料を提供することです.私は、学ばなければならない概念の数に圧倒されています. そこに示されている多くの例は非常に重く感じられ、実際のユースケースに簡単に変換することができないため、非常に難解です。すべてのビルディング ブロックを一緒に構築する方法や、特定の問題の解決にどのように役立つかを説明せずに、詳細を提供しすぎていると思います。

ソース、シンク、フロー、グラフ ステージ、部分グラフ、マテリアライゼーション、グラフ DSL などがありますが、どこから始めればよいかわかりません。クイック スタート ガイドは出発点となることを意図していますが、理解できません。上記の概念を説明せずに投げ込むだけです。さらに、コード例を実行することはできません - テキストをたどることが多かれ少なかれ不可能になる欠落部分があります。

ソース、シンク、フロー、グラフ ステージ、部分グラフ、マテリアライゼーション、および私が見逃した他のいくつかの概念について、簡単な言葉と、すべての詳細を説明していない簡単な例を使用して説明できますか (いずれにせよ、これらはおそらく必要ありません)。始まり）？

より焦点を絞って明確にするために、以前の質問をやり直しています。次のような状況があります。

したがって、ストリームの開始時に Hello(ユーザー名) を取得し、ストリームの終了時に Goodbye を受信します。さよなら(ユーザー名)で返信したいです。

ユーザー名 (u) をキャッシュして、ストリームの存続期間中に使用できるようにするにはどうすればよいですか? 活用できるコンテキストはありますか？それとも、フレームワークの外でそれを行う必要がありますか?

akka の Reactive Streams を理解しようとしています。このブログhttp://bryangilbert.com/blog/2015/02/04/akka-reactive-streams/を読んだことがありますが、それがどのように機能するかについての基本的なアイデアが得られたと思います。しかし、私が理解していないのは、この概念におけるプロセッサの目的です。それはなんのためですか？サブスクライバーが N オブジェクトを要求し、パブリッシャーが onNext() を使用してそれらを送信するだけで十分ではないでしょうか?

私の akka-streams Learn-o-thon は続きます。akka-streams アプリケーションをakka-cluster および DistributedPubSubMediator と統合したいと考えています。

パブリッシュのサポートを追加するのはかなり簡単ですが、サブスクライブの部分に問題があります。

参考までに、サブスクライバーはTypesafe サンプルで次のように指定されています。

私の質問は、このアクターを自分のフローに統合するにはどうすればよいですか? また、ストリーム バックプレッシャーがない状態でパブリッシュ メッセージを確実に取得するにはどうすればよいですか?

あるストリームがメッセージを発行し、別のストリームがそれを消費する (購読している場合) pubsub モデルを実現しようとしています。

彼のような情報源がある場合:

特定の区切りパターンで要素を結合して再分割することは可能ですか?たとえば、「\ n」文字で、結果としてこのようなものになりますか? :

ありがとう！

入力要素が出力に渡され、(フィルターを使用して) すべてをドロップするフィードバック ループに送り返される小さなサンプル グラフを作成しました。

フィードバック ブランチはすべてをドロップするため、アイデンティティ Flow[T] と同じ動作が得られることを期待しています。

代わりに、入力要素は期待どおりに発行されますが、実体化は​​完了しません。

私は何か間違ったことをしていますか？これは起こるはずですか？入力ストリームが完了すると、ブロードキャストのフィードバック出力が完了してはいけませんか?

問題は、ここで説明されている鶏と卵のシナリオに似ていると思いますか?

akka-stream-experimental 2.0.3 を使用しています

乾杯

次のケースでは、処理をアクターに委任しようとしています。私が望んでいるのは、フローがメッセージを処理するたびにそれをアクターに送信し、アクターがそれを大文字にして、応答としてストリームに書き込むことです。

したがって、ポート 8000 に接続し、「hello」と入力して、フローにアクターに送信させ、アクターにそれをストリームに発行させて、大文字でエコー バックされるようにする必要があります。ドキュメントの ActorPublisher の例から、アクター自体はかなり基本的なものです。

このコードが機能しないことはわかっています。実験をクリーンアップしてコンパイルできるようにしました。現在、それは 2 つの別個のストリームにすぎません。ソースまたはシンクをマージして実験しようとしましたが、役に立ちませんでした。

私はAkka Streamsを試しています。ここに私が持っている短いスニペットがあります:

このスニペットを実行すると、次のようになります。

fileSource が空ではないように思えますか? どうしてこれなの？何か案は？FoodFacts.csv のサイズが 40 MB で、1 MB のデータ ストリームを作成するだけです。

8192 の defaultChunkSize を使用しても機能しませんでした!