Почему Akka HTTP закрывает пользовательское соединение, когда генерируется несколько сообщений?

Question

У меня есть простое приложение 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.fromSinkAndSource(incomingMessages, outgoingMessages)

Когда я выдаю клиенту более 30 сообщений в секунду, Akka закрывает соединение.

Не могу понять, где находится настройка, которая настраивает это поведение.Я знаю о OverflowStrategy , но я не настраиваю его явно.

Кажется, у меня OverflowStrategy.fail(), или моя проблема выглядит так.

Answer 1

Вы можете настроить Внутренние буферы .

Есть два способа, как это сделать:

1) application.conf :

akka.stream.materializer.max-input-buffer-size = 1024

2) Вы можете настроить его явно для своего потока:

Flow.fromSinkAndSource(incomingMessages, outgoingMessages)
  .addAttributes(Attributes.inputBuffer(initial = 1, max = 1024))