akka-stream

0熱度

1回答

Akka Streams：使用KillSwitch關閉後創建另一個RunnableGraph

我嘗試關閉一個流，轉換一些數字並創建另一個同一個圖形藍圖的流。但是，第二個流實例不運行，或者至少不會將任何內容打印到控制檯。我錯過了什麼？ object KillSwitchSample extends App { implicit val actorSystem = ActorSystem() implicit val materializer = ActorMateria

3熱度

1回答

多個http請求的Akka流

在我的一個項目中，我有一個akka actor向我的谷歌fcm服務器發送請求。參與者需要一個ID列表，並且應該儘可能多地提出請求。我在runForeach(println(_))中打印出服務器的響應，但我只收到一個打印輸出的整個ID列表。爲什麼會發生？ class FCMActor(val key: String) extends Actor{ import fcm.FCMActor._

3熱度

2回答

Akka流 - 將一串ByteString分割爲多個文件

我試圖將一個傳入的Akka字節流（來自http請求的正文，但也可能來自一個文件）分成多個定義的文件尺寸。例如，如果我正在上傳10Gb文件，它會創建類似10G的1個文件。這些文件會有隨機生成的名稱。我的問題是，我不知道從哪裏開始，因爲我讀過的所有響應和示例都將整個塊存儲到內存中，或者使用基於字符串的分隔符。除非我真的不能擁有1Gb的「塊」，然後就把它們寫到磁盤上。有沒有容易方法來執行那種操作？我

1熱度

1回答

如何在多個來源之間切換？

假設我有兩個無限的相同類型的女巫可以連接到一個Graph。我想從外部已經物化的圖形之間切換它們，可能與使用KillSwitch關閉其中一個圖形的方式相同。 val source1: Source[ByteString, NotUsed] = ??? val source2: Source[ByteString, NotUsed] = ??? val (switcher: Switcher,

1熱度

1回答

如何從主要方法中終止Akka actor系統？

我有this應用程序使用Akka Streams和ReactiveMongo。沒有用戶定義的演員。該應用程序從main方法啓動。問題是在main方法完成後JVM將繼續運行。這是我現在在做什麼： val g = (file: String) => RunnableGraph.fromGraph(GraphDSL.create(Sink.ignore) { implicit builder

0熱度

1回答

Akka HTTP只響應4個請求

我寫了一個簡單的演員，下載一個網頁並將此頁面的主體發送給其發件人。我使用Akka HTTP來構建HTTP請求並處理HTTP響應。這裏是我的代碼： class Downloader(uri: String) extends Actor { import akka.pattern.pipe import context.dispatcher final implici

2熱度

1回答

如何在帶有超時的單獨調度程序上運行Akka Streams圖？

這個問題是基於我做的一個寵物項目和這個SO線程。在Akka HTTP路由定義中，我開始了一個長期運行的過程，當然我想這樣做而不會阻止用戶。我可以用下面的代碼片段來實現這一目標： blocking-io-dispatcher { type = Dispatcher executor = "thread-pool-executor" thread-pool-executo

3熱度

1回答

維護Akka Streams中涉及多個JVM的背壓的方法

我知道，從Akka 2.4.16開始，沒有「遠程」實現Reactive Streams。規範側重於在單個JVM上運行的流。但是，考慮用例涉及另一個JVM進行某些處理，同時保持背壓。這個想法是有一個主要的應用程序，提供一個運行流的用戶界面。例如，這個流有一個階段執行一些重要的計算，應該運行在不同的機器上。我感興趣的方式來運行在分佈式的方式流 - 我一些相關的文章來指出一些想法：通過TCP連接流使

0熱度

1回答

來自ftp的Akka流，逐行

我試圖從使用alpakka和scala流的ftp服務器讀取文件。我從Ftp.fromPath(...)得到的類型是Source[ByteString, Future[IOResult]]。我想逐行閱讀文件（這是一個CSV文件），但我不知道如何。我將不勝感激任何幫助。

0熱度

1回答

測試完成無限流的akka-http路由

我有一個akka-http路由，它返回一個包含無限實體流的Source。我如何測試使用路由測試工具包？我想檢查流的前n個元素，但是我已經看了一下testkit代碼，看起來好像沒有直接的方式來訪問響應中的Source。它總是被轉換爲ByteString的序列，在我的情況下，這只是導致TimeoutException，因爲流不會終止。作爲參考，這個問題可以用路線轉載看起來像這樣： case clas