akka-stream

0熱度

1回答

我正在嘗試追蹤自定義流程中存在額外請求向上遊的錯誤。我寫了一個測試案例，鑑於我看到的行爲和日誌記錄應該失敗，但事實並非如此。我然後做一個平凡的流量明顯帶有額外拉動，但測試仍然沒有失敗。下面是一個例子： class EagerFlow extends GraphStage[FlowShape[String, String]] { val in: Inlet[String] = Inl

5熱度

1回答

Source <Out,Mat>的類型參數是什麼意思？

我想了解Akka流的源類型，指定爲here。不幸的是，我找到的文檔和示例沒有解釋每個類型參數的實際含義。我猜測Out是物化時發出的類型。那是對的嗎？其他類型參數Mat是什麼？

1熱度

3回答

阿卡流TCP +阿卡流卡夫卡生產者未停止不發佈消息，而不是錯誤-ING出

我有以下流： Source(IndexedSeq(ByteString.empty)) .via( Tcp().outgoingConnection(bsAddress, bsPort) .via(Framing.delimiter(ByteString("\n"), 256, allowTruncation = true)) .map(_.utf8String)

1熱度

1回答

阿卡-HTTP如果失敗

我調用外部API，我想在不同的狀態代碼的情況下返回結果「AS IS」用戶比OK與傳出響應響應： val connectionFlow: Flow[HttpRequest, HttpResponse, Future[Http.OutgoingConnection]] = Http().outgoingConnection("akka.io") def responseFuture: Fu

1熱度

2回答

如何創建一個接收器來讀取源中的所有字節？

我有一個Scala的方法，它這個源按我的要求 def func(data : Source[ByteString,Any]) { ... } 所有我想做的事，現在，是從這個源中讀取所有字節並將其寫入文件。我已經閱讀了很多例子，它們可以做更復雜的事情，比如對所有值進行摺疊計算。我想要更簡單的東西，只是獲取字節，以便我可以寫出來。我將如何做到這一點？

3熱度

1回答

使用Akka從服務器到客戶端的流文件

基本上我想允許用戶從服務器下載csv文件。假定服務器上已經存在CSV文件。 API端點通過GET/export顯示。如何將文件從Akka HTTP服務器流式傳輸到客戶端？這是我迄今爲止... 服務： def export(): Future[IOResult] = { FileIO.fromPath(Paths.get("file.csv")) .to(Sink.ignore

0熱度

1回答

Akka Streams：KillSwitch自定義SourceShape從視頻文件中發射幀

繼documentation to implement a KillSwitch之後，我能夠編寫這個簡單的示例來停止Source發出無限數字。 object KillSwitchSample extends App { implicit val actorSystem = ActorSystem() implicit val materializer = ActorMateri

0熱度

1回答

收集水槽物化值作爲源

是否有更好的方法來實現下面的方法（即通過去除Materializer約束。）： import akka.NotUsed import akka.stream.Materializer import akka.stream.scaladsl.{Keep, Sink, Source} def assemble[A, B, C](source: Source[A, NotUsed])

2熱度

1回答

允許與Akka流的單個連接作爲tcp服務器

我正嘗試使用Akka流構建一個簡單的tcp服務器。 Tcp() .bind(props.host, props.port) .to(Sink.foreach(_.handleWith(handler))) .run() .onComplete { case Success(i) => logger.info(s"Server is bound at

0熱度

1回答

觸發信號流

在akka-stream cookbook documentation該代碼段示出了如何觸發元件的編程的流量： val graph = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder => import GraphDSL.Implicits._ val zip = builder.add(Zip[Mess