Akka Flux continuu consuma websocket

Acest cod este o multime Akka-ish: scaladsl este doar ca Akka ca GraphDSL sau de punere în aplicare un obicei GraphStage. Singurul motiv pentru care, OMI/E, pentru a merge la GraphDSL dacă forma actuală a graficului nu este ușor exprimate în scaladsl.

Eu personal as merge pe traseul de definire a unui CoinPrice clasa pentru a face modelul explicit

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

Și apoi au o Flow[Message, CoinPrice, NotUsed] care interpreteaza 1 mesaj de intrare în zero sau mai multe CoinPrices. Ceva (folosind Juca JSON aici), cum ar fi:

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

S-ar putea, în funcție de ceea ce dimensiune de JSONs în mesaj sunt, să rupă în diferite stream etape, pentru a permite o asincron granița dintre JSON analiza și extragerea în CoinPrices. De exemplu,

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

În cele de mai sus, etapele pe fiecare parte a async limita va executa separat actori și, astfel, eventual, concomitent (dacă nu e suficient de nuclee CPU disponibile etc.), la costul de deasupra suplimentar pentru actori de a coordona și de a face schimb de mesaje. Acel plus de coordonare/comunicare aeriene (cf. Gunther Universal Scalabilitate Lege) este doar de gând să fie în valoare de ea, dacă obiecte JSON sunt suficient de mari și vine suficient de repede (în mod constant vin în fața anterioară a terminat de procesat).

Dacă intenția dumneavoastră este de a consuma websocket până când programul se oprește, s-ar putea găsi mai clară a folosi doar Source.never[Message].