用于处理JVM上的流数据的Akka实现
IllegalStateException - Sink.asPublisher 仅支持一个订阅 - 使用 WebSockets 时
在生产中,我们的应用程序生成了以下堆栈跟踪。这里没有我们自己的代码——没什么可继续的。 了解 akka-stream 库的人可以解释一下: 什么是
我有一个 akka http API,用户可以将 S3 URL 发送到服务器。然后,服务器从 AWS 服务器启动流并对源执行后续操作。不过我想验证
Akka Streams RestartSource.onFailuresWithBackoff 停止条件
如果出现异常,我正在使用 RestartSource.onFailuresWithBackoff 重新启动源,但如果收到某种异常类型,我想停止(取消)重新启动。例如:
Akka Streams flatMapConcat 在创建新源时停止以前的源
我有一个用例,我有一个提供服务“地址”的源。然后我想将该地址映射到从该服务传输数据的源。现在,万一我会得到......
如何捕获最大RestartSource后alpakka kafka源流失败
如何捕获最大重启次数后RestartSource的错误? 我想在源失败最大次数后做一些事情。 我可以看到源正在重新启动...
我正在尝试将传入的 Source[ByteString, Any] 广播到 2 个不同的流,然后扇入(zip)输出。但是我收到错误“重载方法 ~> 使用替代方案”。 瓦尔
我要解决的是以下情况: 给定一个无限运行的 Akka Stream,我希望能够监视流的某些点。我能想到的最好的办法就是把这些乱七八糟的东西送到哪里......
以下是问题的背景: 有一个 Source,它不断地滴答作响,无法保证滴答的频率 我们想要限制源的最大滴答率(例如,我们
从流中提取 Source[ByteString, Any] 到 2 个接收器中
我正在尝试将传入的 Source[ByteString, Any] 放入 2 个接收器中并尝试在 akka 流图中复制传入流。我想要“is”作为输入流,但不是 t...
嗨,我正在尝试使用 scala 中的 akka-stream-alpakka-amqp 库在兔子消息中设置标头,但我无法使其工作并读取存储库,我看不到任何参考或我...
Flow.batch 有逆向吗 我希望该方法的签名为 Flow[Vector[T],T,NotUsed] 它应该解包一个 Vector 并将 Vector 的每个元素作为单个元素发送到下游。
Scala-3.3.0:dotty.tools.FatalError:无法解析对
我在 scala-3.3.0 和 akka(或 pekko)中看到此错误: dotty.tools.FatalError:无法解析对 akka.stream.scaladsl.type.Source 类型的引用 dotty.tools.FatalError:无法解析引用...
当我使用 playframework 响应分块的 akka 源时,为什么会发生 java.lang.IllegalStateException?
当我使用 playframework 响应 mp3 音频时。我想使用分块收发器。然后我在 akka 流中创建一个 akka 源来响应。并使用播放 Ok.chunked 结果。但它有一个例外...
我们希望在 akka 流关闭时完成以下任务 源头停了, 流中的元素在超时点之前已完成 然后流/演员系统关闭 我们有...
我正在使用 akka 流从 TCP 端点消耗 ByteString 流。数据块应该形成一个或多个 XML 文档。我希望我的流输出带有完整 Xml 的 ByteString。 例如...
Akka Streams:向 QueueSource 提供 StreamDetachedException
我们有一个 DownloadFileFlow 类,它使用 Akka Streams 和 Akka Http 从不同域下载文件;有时(~50%)在向我们遇到的 QueueSource 提供服务时
我必须通过 supervision.decider 但在 akka 2.6.0+ Materializer 中没有类似的设置。 任何人都可以分享如何在更高版本中编写以下代码: 验证垫:ActorMaterializer =
Android Studio 中的 Akka-Http-client [关闭]
如何在 android kotlin 中使用 Akka HTTP 库...? 我正在 Kotlin 的 Android Studio 上开发 Akka HTTP 库...但我不知道如何在 android studio 上使用 Akka HTTP 库。我哈...
如果 postStop() 失败,我们如何重试/重启 graphStage? 我对 akka 比较陌生,非常感谢您的帮助。 这是我正在尝试做的事情: 来源 .via(音频...
我有一个返回Source流的请求,我想测试一下,像这样。* class * import sttp.client._ class HttpClient() (implicit sttpBackend: SttpBackend[Future, Source[...