akka-stream

13熱

1答えて

PlayFrameworkでAkka Streams SourceQueueを使用する方法

SourceQueueを使用して、エレメントをAkkaストリームソースに動的にプッシュしたいとします。再生コントローラには、chunckedメソッドを使用して結果をストリーミングできるソースが必要です。 Playは独自のAkkaストリームシンクをボンネットで使用しているため、シンクを使用してソースキューを自分で実現することはできません。chunkedメソッドで使用する前にソースが消費されるためで

1熱

1答えて

グループストリーム要素

は重み関数 val weightFunction: Element => Int 考える val myFlow: Flow[Element] = ... //some flow.. を想像してみて、私は、各要素のこと val transformedFlow: Flow[List[Element]] などを取得したいと思います transformedFlowはList [Element]で

3熱

1答えて

Akka Streamsでは、Sink.fold（）処理がシリアル化されていますか？

私はAkkaストリームを開始しています。私は、Webサービスからデータを読み込んでS3でそれらを維持するストリームを作成しようとしています。永続性のためにSink.foldメソッド（永続化された要素に関する情報を収集するために）を使用してシンクを定義すると、シンクに送信される要素が順番に処理されるか、並列処理されるのでしょうか？基本的な質問ですが、私はドキュメントで決定的な答えを見つけることが

0熱

2答えて

AkkaのStrictとStreamed WebSocketメッセージの両方を消費する

私は、Akka HTTPを使用してWebソケットサービスを構築することを試しています。私は完全に到着する厳密なメッセージを処理する必要があり、m個のフレームに到着するストリームメッセージを処理する必要があります。私はhandleWebSocketMessages（）でWebソケットの処理をフローに渡すためにルートを使用しています。 val route: Route = get {

0熱

1答えて

アッカストリーム - Source.fromPublisher

Iはアッカ・ストリームクイックスタートガイドに基づいて、次のコードを実行しようとしている： implicit val system = ActorSystem("QuickStart") implicit val materializer = ActorMaterializer() val songs = Source.fromPublisher(SongsService.stream)

5熱

2答えて

バックプレッシャ付き行単位ファイルIOのシンク

現在、処理パイプラインを処理するために手動で管理されたバックプレッシャを持つakkaアクタを使用するファイル処理ジョブがありますが、入力時にバックプレッシャを正常に管理できませんでしたファイル読み取り段階。このジョブは、各行の先頭にあるID番号で入力ファイルとグループ行を取り出し、新しいID番号の行にヒットすると、グループ化された行をメッセージ経由で処理アクターにプッシュし、新しいID番号をファ

1熱

1答えて

Akka Streams入門

ここ数日間、Akka Streamsについて読んできました。私はここ数ヶ月間ScalaのRxライブラリで作業していました。私には、両方のライブラリが提供してくれるものがある程度重なっているようです。 RxScalaはちょっと使い始め、理解して使いやすくなっていました。たとえばScalaのRxライブラリを使用してカフカのトピックに接続し、Observableにラップしてサブスクライバにメッセージを受

2熱

1答えて

Akka-httpでHTTPSリクエストを処理する[java]

小さなプロキシを作成して、すべてのローカルhttpリクエストを傍受し、リモートプロキシで認証しました。コードスニペットは次のとおりです。 Source<IncomingConnection, CompletionStage<ServerBinding>> serverSource = Http.get(system).bind(ConnectWithHttps.toHost("lo

5熱

1答えて

akkaストリームの条件に基づいた代替フロー

カスタムフローのあるストリームがあり、特定の段階でストリームを分割し、後で再びマージする2つの代替データ処理が必要です。など。 -> F3 -> F6 Src -> F1 -> F2 > Merge -> Sink -> F4 -> F5 F2データはフォーマットAが含まれている場合、それはF4に行く他に、F3を流れるように行くべきと言っ条件を持っている必要

6熱

1答えて

クラスタノード上で実行されているAkkaストリーム

Akkaストリームのスケーリング部分を理解しようとしていますが、ここではさまざまな回答がありましたが、この機能がどのように達成されたかはまだ分かりません。私が理解しているところからストリームとやりとりすることを意図しているActorPublisher/Subscribersがありますが、同時にこれらは遠隔にすることはできません。ストリームと一緒にakkaクラスタを使用することはできますか？私は