問題タブ [akka-stream]

これに本当に似た何かをする必要があります https://github.com/typesafehub/activator-akka-stream-scala/blob/master/src/main/scala/sample/stream/GroupLogFile.scala

私の問題は、不明な数のグループがあり、mapAsync の並列処理の数が取得したグループの数より少なく、最後のシンクでエラーが発生することです。

アップストリーム エラー (akka.stream.impl.StreamSubscriptionTimeoutSupport$$anon$2) による SynchronousFileSink(/Users/sam/dev/projects/akka-streams/target/log-ERROR.txt) の破棄

akkaストリームのパターンガイドhttp://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-cookbook.htmlで提案されているように、途中にバッファを配置しようとしました

しかし、同じ結果で

次のように、ReactiveStreams Publisher に基づいて Akka Stream の Source を作成しました。

Flickr に検索要求を行い、結果をJsValues としてソースします。さまざまなフローとシンクに接続しようとしましたが、これが最も基本的なセットアップになります。

がonNext数回呼び出され、次にonComplete. ただし、シンクは何も受け取りません。これはソースを作成する有効な方法ではありませんか?

この例を実行することで、Akka HTTP の経験を積んでいます。学習のために、Maven プロジェクトに変換しました。ただし、Akka v2.3.12 と Akka Stream v1.0 を使用すると、次のようなコンパイル エラーが発生します。POST DSL は、簡潔にするために投稿していない同様のエラーで失敗します。サンプルを実行するにはどうすればよいですか?

私の質問はどこかに関連しています: Source.actorRef によって作成された akka ストリーム Source の基になる ActorRef にアクセスする いくつかの違いがあります:

1. akka-stream 実験的 1.0 を使用しています
2. 私はactorPublisherモデルを使用しています
3. 並列処理によるストリーム定義に FlowGraph DSL を使用しています

Source が保持する Actor Publisher インスタンスにメッセージを送信するために、actorRef を取得する方法が見つかりません。

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

オリバー

いくつかのブロッキング呼び出しを でラップしようとしていFutureます。戻り値の型はSeq[User]です。以下は、さまざまなオーバーロードされたバージョンが存在するという苦情でコンパイルされません。助言がありますか？私はほとんどすべてのバリエーションを試しましたが、運がありません。Usercase classSource.apply

Akka Streams を使用する場合、リソースのクリーンアップに不要になったストリームを閉じる/シャットダウンする方法はありますか?

EDIT : ソースが無限ストリームで構成されている場合、ソースが完成しない可能性があり、ソースが完成する前に停止したいと思います。

使用例:

ストリームをシャットダウンする方法はありますか?