Akka Fluxo continuamente consumir websocket

Que código é bastante Akka-ish: scaladsl é tão Akka como o GraphDSL ou a implementação de um personalizado GraphStage. A única razão, IMO/E, para ir para o GraphDSL é se a forma do gráfico não estão expressas no scaladsl.

Eu poderia ir pessoalmente a rota de definição de uma CoinPrice classe para fazer o modelo explícito

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

E, em seguida, ter uma Flow[Message, CoinPrice, NotUsed] que analisa 1 mensagem recebida em zero ou mais CoinPrices. Algo (usando Jogar JSON aqui) como:

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
            }
          }
        }
    }

Você pode, dependendo do que o tamanho da JSONs na mensagem são, querem quebrar que em diferentes sequência de fases para permitir um assíncrono fronteira entre o JSON analisar e a extração em CoinPrices. Por exemplo,

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
      }
    }
  }

No exemplo acima, as fases em ambos os lados da async limite será executado em separado atores e, assim, possivelmente, simultaneamente (se houver um número suficiente de núcleos de CPU disponível, etc.), o custo de sobrecarga extra para os atores, para coordenar e trocar mensagens. Extra coordenação/sobrecarga de comunicação (cf. Gunther Universal Escalabilidade da Lei), só vai valer a pena se a objetos JSON são suficientemente grandes e chegando de forma suficientemente rápida (consistentemente vindo antes que a anterior tenha acabado de processamento).

Se a sua intenção é consumir o websocket até que o programa pára, você pode encontrá-lo mais claro para só usar Source.never[Message].