akka-stream

1熱度

2回答

你能解釋一下如何在阿卡流中使用新的groupBy嗎？ Documentation似乎是相當無用的。 groupBy用於返回(T, Source)但不再。這裏是我的例子（我從模擬文件中模擬出一個）： Source(List( 1 -> "1a", 1 -> "1b", 1 -> "1c", 2 -> "2a", 2 -> "2b", 3 -> "3a", 3 -> "3b

0熱度

1回答

以Play 2.5命名的事件流和服務器發送的事件

服務器發送的事件中的默認事件名稱/類型爲「消息」。我正在嘗試更改事件名稱，但它不起作用。我使用Play 2.5和akka流。 (actorRef,sourcePublisher)= Source .actorRef[T](10, OverflowStrategy.fail) .toMat(Sink.asPublisher(true))(Keep.both) .r

0熱度

1回答

監控akka流的生命週期

我想要監控一個akka-stream的生命週期，它好像monitor會做我需要的，但是我的監控功能是異步的，返回Future，所以我需要監控器是異步。 monitor具有以下特徵： def monitor[Mat2]()(combine: (Mat, FlowMonitor[Out]) ⇒ Mat2): ReprMat[Out, Mat2] 但我需要的東西，如： def monitorAsyn

0熱度

1回答

使用OpenCV進行流實現時發生故障BackgroundSubtractorMOG2與Java中的Akka流

對於我的項目我想創建一個應用程序，在使用Akka流時使用Java中的OpenCV庫進行一些視頻分析。我試過在不使用Akka流的一個單獨的項目中使用BGsubtractorMOG2，並且一切正常，但是現在當我用包含MOG2函數的階段實現流時，我的程序崩潰了。我相信問題出在MOG2上，因爲如果我嘗試刪除它並在視頻中捕獲並顯示幀，則一切正常。這裏是一些代碼的阿卡演員 private final Ma

2熱度

2回答

Akka流 - 按流中元素的數量進行過濾

我正在Scala編寫應用程序，我正在使用Akka流。在某一點上，我需要過濾掉N個元素少於N個的流。因此，例如，與N=5： Source(List(1,2,3)).via(myFilter) // => List() Source(List(1,2,3,4)).via(myFilter) // => List() 將成爲空流， Source(List(1,2,3,4,5)).via(my

1熱度

2回答

Akka指令fileUpload將整個輸入實體文件複製到內存中？

Akka指令「fileUpload」是否將整個輸入實體文件複製到內存中？是否有一種方法可以將大文件分段上載爲1 GB而不消耗1 GB內存？ val uploadFile = File.createTempFile("uploadFile", ".txt") extractRequestContext { ctx => implicit val materializer =

0熱度

1回答

Haw在akka-stream-kafka中從配置（application.conf）創建ProducerSettings？

我正在嘗試學習和使用akka-stream-kafka並正在審閱其[文檔] [1]。在生產者設置部分，它告訴我們可以使用編程方式和從配置方式創建ProducerSettings。有一個程序化構建的例子，但沒有如何通過配置來創建它的例子。程序化結構很簡單，下面是一個例子。不過，我想使用配置基礎結構，並希望配置來自application.conf，因爲它會給我更多的控制權。我似乎無法在谷歌上找到它的一

0熱度

1回答

如何處理定製的阿卡水槽內的未來？

我想實現一個定製的Akka接收器，但是我找不到處理它內部未來的方法。 class EventSink(...) { val in: Inlet[EventEnvelope2] = Inlet("EventSink") override val shape: SinkShape[EventEnvelope2] = SinkShape(in) override def

1熱度

1回答

使用Akka Streams測試工具包時發生零星超時失敗

我有一些奇怪的行爲，每當我嘗試使用Akka的Streams測試工具包時，都會看到這種行爲。我有以下自定義定義： trait PauseFilter[T] { def shouldPause(message: T): Boolean } trait MessagePauser[T] { def pause(message: T): Unit } trait Me

0熱度

1回答

連接生產者流程圖

我是新的使用aka流kafka（和akka流一般）。我正在嘗試構建一個圖表，以便將消息發佈到不同的主題。如何將生產者作爲流連接以提交處理後的消息？我試着用Producer.flow但由於您使用的是GraphDSL我不能得到commitScaladsl object TestFoo { import akka.kafka.ProducerMessage.Message impl