Поток Akka непрерывно потребляет websocket

Этот код очень похож на Акка-иш: scaladsl это так же Акка, как и GraphDSL или реализация пользовательского GraphStage. Единственная причина, ИМО/Е, пойти в GraphDSL является, если фактическая форма графика не легко выражается в scaladsl.

Я бы лично пошел по пути определения CoinPrice класс, чтобы сделать модель явной

case class CoinPrice(coin: String, price: BigDecimal)

А затем выпейте Flow[Message, CoinPrice, NotUsed] который анализирует 1 входящее сообщение на ноль или более CoinPrices. Что-то (используя здесь Play JSON), например:

val toCoinPrices =
  Flow[Message]
    .mapConcat { msg =>
      Json.parse(msg.toString)
        .asOpt[JsObject]
        .toList
        .flatMap { json =>
          json.underlying.flatMap { kv =>
            import scala.util.Try

            kv match {
              case (coin, JsString(priceStr)) =>
                Try(BigDecimal(priceStr)).toOption
                  .map(p => CoinPrice(coin, p))                

              case (coin, JsNumber(price)) => Some(CoinPrice(coin, price))
              case _ => None
            }
          }
        }
    }

Вы можете, в зависимости от размера JSON в сообщении, разбить его на различные этапы потока, чтобы обеспечить асинхронную границу между анализом JSON и извлечением в CoinPrices. Например,

Flow[Message]
  .mapConcat { msg =>
    Json.parse(msg.toString).asOpt[JsObject].toList
  }
  .async
  .mapConcat { json =>
    json.underlying.flatMap { kv =>
      import scala.util.Try

      kv match {
        case (coin, JsString(priceStr)) =>
          Try(BigDecimal(priceStr)).toOption
            .map(p => CoinPrice(coin, p))

        case (coin, JsNumber(price)) => Some(CoinPrice(coin, price))
        case _ => None
      }
    }
  }

В приведенном выше примере этапы по обе стороны от async граница будет выполняться отдельными участниками и, таким образом, возможно, одновременно (если доступно достаточно ядер процессора и т. Д.), За счет дополнительных накладных расходов для участников для координации и обмена сообщениями. Эти дополнительные накладные расходы на координацию/связь (см. Универсальный закон масштабируемости Гюнтера) будут оправданы только в том случае, если объекты JSON достаточно велики и поступают достаточно быстро (последовательно поступают до того, как предыдущий завершит обработку).

Если вы намерены использовать websocket до тех пор, пока программа не остановится, возможно, вам будет проще просто использовать Source.never[Message].