akka-stream

0熱度

1回答

我試圖運行一個簡單的Akka Stream File Sink示例，但沒有成功。我可以創建一個Source，運行Flow，然後創建一個文件，但ByteString不會寫入文件。而如果我嘗試將流量輸出打印到控制檯，我可以這樣做。我在這裏錯過了什麼嗎？ import akka.stream._ import akka.stream.scaladsl._ import akka.{ NotUsed,

1熱度

2回答

如何從FileIO.fromPath以慣用的方式獲取字節串

什麼是從Scala中的FileIO.fromPath獲取akka Bytestring的習慣用法？

3熱度

2回答

Akka流重試重複結果

我正在實現一個HTTP資源的迭代器，我可以恢復分頁元素列表，我試圖用一個普通的Iterator做到這一點，但它是一個阻塞實現，並且因爲我是使用akka它使我的調度員有點瘋狂。我會用akka-stream來實現相同的迭代器。問題是我需要不同的重試策略。該服務返回一個由id標識的元素列表，有時當我查詢下一頁時，該服務返回當前頁面上的相同產品。我現在的算法是這樣的。 var seenIds = S

2熱度

1回答

阿卡流：無參數GraphDSL.create（）VS GraphDSL.create（匯）

的doc具有下面的例子（只什麼有關我的問題是所示）： val resultSink = Sink.head[Int] val g = RunnableGraph.fromGraph(GraphDSL.create(resultSink) { implicit b => sink => import GraphDSL.Implicits._ // importing the

1熱度

1回答

如何批量使用阿卡流Flow.batch

我使用下面的代碼來測試阿卡流Flow.batch的行爲從光源發出的多種元素，但我不能以某種方式弄清楚爲什麼結果不是我所期望： Source(1 to 20) .map(x => { println(s"received: ${x}") x }) .batch(max=3, first => first.toString) {(batch, elem

15熱度

2回答

從阿卡流中的演員創建流

可以分別使用Source.actorPublisher()和Sink.actorSubscriber()方法從演員創建源和匯。但是有可能從演員創建一個Flow？從概念上講，似乎沒有一個很好的理由不考慮它，因爲它實現了ActorPublisher和ActorSubscriber特徵，但不幸的是，Flow對象沒有任何方法來做到這一點。在this優秀的博客文章中，它是在Akka Streams的早期版

4熱度

1回答

如何使用akka的http客戶端設置最大內容長度？

如果我嘗試下載一個非常大的文件（超過100 MB）我得到這個錯誤： You can configure this by setting `akka.http.[server|client].parsing.max-content-length` or calling `HttpEntity.withSizeLimit` before materializing the dataBytes stre

3熱度

2回答

Akka流 - 連接水槽到源？

我有這種情況，當一個sink（或中間flow）實際上可以產生一些副作用數據，必須推回（或附加）到Source。有沒有一種方法可以使用流式DSL來實現這一點？我可以使用一些阻塞隊列或排序來創建source，然後將數據直接推送到該隊列，但是這是打破流的抽象。也許有一個我不知道的更好的解決方案？

0熱度

1回答

無法使用Akka-Http發送具有相同websocket連接的多個消息

我正在使用akka-http websocket客戶端與websocket服務器連接併發送和檢索消息。 object WebSocketClientFlow { def main(args: Array[String]) = { implicit val system = ActorSystem() implicit val materializer = ActorMaterializer(

6熱度

1回答

鏈中的Akka-http客戶端請求

我想使用akka-http-client將鏈接http請求作爲Stream。鏈中的每個http請求都取決於先前請求的成功/響應，並使用它來構建新的請求。如果請求不成功，Stream應返回不成功請求的響應。我該如何在akka-http中構建這樣的流？我應該使用哪個akka-http客戶端級API？