我有一个简单的WebSocket应用程序,它基于Akka HTTP
/ Reactive streams
,就像这个https://github.com/calvinlfer/akka-http-streaming-response-examples/blob/master/src/main/scala/com/experiments/calvin/ws/WebSocketRoutes.scala#L82。
换句话说,我有Sink
,Source
(由Publisher
制作)和Flow:
Flow.fromSinkAndSource(incomingMessages, outgoingMessages)
当我每秒向客户端生成超过30条消息时,Akka
关闭了一个连接。
我无法理解,配置此行为的设置在哪里。我知道OverflowStrategy,但我没有明确配置它。
看来,我有OverflowStrategy.fail()
,或者我的问题看起来像它。
你可以调整Internal buffers。
有两种方法,如何做到:
1)application.conf:
akka.stream.materializer.max-input-buffer-size = 1024
2)您可以为您的Flow明确配置它:
Flow.fromSinkAndSource(incomingMessages, outgoingMessages)
.addAttributes(Attributes.inputBuffer(initial = 1, max = 1024))