問題タブ [akka-stream]

akka-streams ベースの Tcp サーバーでの close() の予想される動作は、クライアントがすぐに SocketException を確認し、例外の後に発行されたメッセージがサーバーに配信されないことです。ただし、例外の前に送信された最後のメッセージを配信する必要があります。新しいクライアントはすべて ConnectException に遭遇するはずです。

Tcp サーバーに対して次のように実装 close() を使用しているakka-streams 1.0 ドキュメントの「接続を閉じる」に関する段落を解読できません。

これはサーバーを閉じる良い方法ですか? (展開用など)

上記の close() の実装の有効性について、誰かがレビューしてコメントできれば幸いです。このコードがすべての要件を満たしているかどうかを証明しようとしていますが、最初にフローのソースを成功させるべきか、最初に serverBinding をバインド解除するべきか疑問に思っています。レビュー/批判は大歓迎です。前もって感謝します ！

バックプレッシャー機能でデータを処理するためのフレームワークであるAkka Streamsの学習を開始しました。ライブラリはAkkaの一部であり、次のように記述されています。

Akka は、JVM 上で高度な同時実行性、分散性、回復力を備えたメッセージ駆動型アプリケーションを構築するためのツールキットおよびランタイムです。

これらの機能は、Akka アクターの性質に由来します。しかし、私の見解では、ストリーム処理とアクターは互いに無関係な概念です。

質問: Akka Streams は Akka アクターのこれらの機能を利用しますか? はいの場合、俳優がストリームをどのように支援しているか説明していただけますか?

フロー A が生成する場合

このアウトレットを、ソースにマッピングされているフロー B のインレットに接続するにはどうすればよいですか。たとえば、フロー B のアウトレットは次のようになります。

私が達成しようとしているのは、akka ストリームを使用した同期フィードバック ループのようなものを実装することです。

を持っているとしましょうFlow[Int].filter(_ % 5 == 0)。のストリームIntをこのフローにブロードキャストし、タプルをその直後に圧縮すると、次のような結果が得られます

Option[Int]次の要素をプッシュした後にフローが要素を放出したかどうかを示すを放出する方法はありますか?

DetachedStageフローが前Flowの段階で引っ張られるたびに、私は彼が次の要素を必要とすることを知っていました。背後のステージが要素を受け取らなかった場合、それは None でした。

残念ながら、結果は良くなく、多くの順位がずれています。

補足事項

フィルタ フローは単なる例です。これは非常に長いフローになる可能性があり、そのフローのOptionすべての段階で を発行する機能を提供することはできません。そのため、フローが次をプッシュしたかどうかを実際に知る必要があります。代わりにダウンストリームから次をリクエストしました

と も試してみましたがconflate、expandこれらは結果の位置オフセットでさらに悪化しています

構成で変更したことの 1 つは、フローのinitialandmaxバッファーでした。これにより、示された要求が実際にプッシュした要素の後にあることを確認できます。

この問題を解決する方法についていくつかの提案をいただければ幸いです。

akka-httpを使用して、受信データのみをリッスンし、何も書き戻さないFlow[Message, Message, _] 渡されるを構築するにはどうすればよいですか? handleWebsocketMessageとにかくシンクを使用することは可能ですか? シンクは私が必要としているもののように聞こえるからです。

私のアプリケーション構造は次のようになります

EmailService.scala

MonitorStatusService.scala

サーバー.scala

何が必要ですか？ のように複数のルートをバインドしたい

しかし、私がそうすると、エラーが次のように表示されます

何をすればよいでしょうか？