Как реально проверить поток akka? - PullRequest
Новая
       15

Как реально проверить поток akka?

4 голосов
/ 13 мая 2019

Я использую WebsocketClient и хотел бы проверить полученное сообщение.Я выбрал фреймворк Scalatest и знаю, что тест должен выполняться асинхронно.

Клиент веб-сокета выглядит следующим образом: 

import akka.{Done}
import akka.http.scaladsl.Http
import akka.stream.scaladsl._
import akka.http.scaladsl.model.ws._
import io.circe.syntax._

import scala.concurrent.Future


object WsClient {

  import Trigger._

  private val convertJson: PreMsg => String = msg =>
    msg.asJson.noSpaces

  val send: PreMsg => (String => Unit) => RunnableGraph[Future[Done]] = msg => fn =>
    Source.single(convertJson(msg))
      .map(TextMessage(_))
      .via(Http().webSocketClientFlow(WebSocketRequest(s"ws://{$Config.host}:{$Config.port}/saprs")))
      .map(_.asTextMessage.getStrictText)
      .toMat(Sink.foreach(fn))(Keep.right)


}

и тест: 

  feature("Process incoming messages") {
    info("As a user, I want that incoming messages is going to process appropriately.")
    info("A message should contain the following properties: `sap_id`, `sap_event`, `payload`")

    scenario("Message is not intended for the server") {
      Given("A message with `sap_id:unknown`")
      val msg = PreMsg("unknown", "unvalid", "{}")
      When("the message gets validated")
      val ws = WsClient.send(msg)
      Then("it should has the `status: REJECT` in the response content")
      ws { msg =>
        //Would like test against the msg here
      }.run()
        .map(_ => assert(1 == 1))

    }

Я хотел бы проверить содержимое msg, но яне знаю, как это сделать.

1 Ответ

3 голосов
/ 13 мая 2019

Я последовал примеру play-scala-websocket

Они используют WebSocketClient в качестве помощника, см. WebSocketClient.java

Тогда тест выглядит так: 

Helpers.running(TestServer(port, app)) {
    val myPublicAddress = s"localhost:$port"
    val serverURL = s"ws://$myPublicAddress/ws"

    val asyncHttpClient: AsyncHttpClient = client.underlying[AsyncHttpClient]
    val webSocketClient = new WebSocketClient(asyncHttpClient)
    val queue = new ArrayBlockingQueue[String](10)
    val origin = serverURL
    val consumer: Consumer[String] = new Consumer[String] {
      override def accept(message: String): Unit = queue.put(message)
    }
    val listener = new WebSocketClient.LoggingListener(consumer)
    val completionStage = webSocketClient.call(serverURL, origin, listener)
    val f = FutureConverters.toScala(completionStage)

    // Test we can get good output from the websocket
    whenReady(f, timeout = Timeout(1.second)) { webSocket =>
      val condition: Callable[java.lang.Boolean] = new Callable[java.lang.Boolean] {
        override def call(): java.lang.Boolean = webSocket.isOpen && queue.peek() != null
      }
      await().until(condition)
      val input: String = queue.take()
      val json:JsValue = Json.parse(input)
      val symbol = (json \ "symbol").as[String]
      List(symbol) must contain oneOf("AAPL", "GOOG", "ORCL")
    }
  }
}

См. Здесь: FunctionalSpec.scala

...