Тестирование потока Akka, содержащего поток cachedHostConnectionPool Akka Http

Question

Любые идеи о том, как лучше всего протестировать Akka Stream, содержащий Akka Http Flow? Я борюсь со следующим методом, в частности:

def akkaHttpFlow(server: String)(implicit actorSystem: ActorSystem, actorMaterializer: ActorMaterializer) = {

  val uri = new java.net.URI(server)
  val port: Int = if( uri.getPort != -1) { uri.getPort } else { 80 }

  Http().cachedHostConnectionPool[Seq[String]](uri.getHost, port)
  .withAttributes(ActorAttributes.supervisionStrategy(decider))

}

Это тестовый код

val emails = Set("[e-mail]")
val source: Source[String, NotUsed] = Source(emails)

val f = source
  .grouped(10)
  .via(requestBuilderFlow)
  .via(akkaHttpFlow)
  .map(responseHandler)
  .runForeach(println)

f.futureValue.shouldBe(Done)

Сбой со следующей ошибкой (не неожиданно) >>>

 The future returned an exception of type: akka.stream.StreamTcpException, with message: Tcp command [Connect(localhost:9001,None,List(),Some(10 seconds),true)] failed because of Connection refused.

Можно ли встраивать http-сервер akka в тест? Или как лучше всего структурировать код, чтобы можно было его смоделировать?

Поддерживающий код

object MyOperations extends StrictLogging {

  val requestBuilderFunc : Seq[String] => (HttpRequest, Seq[String]) = { emails : Seq[String] =>
    HttpRequest(method = HttpMethods.POST, uri = "/subscribers").withEntity(ContentTypes.`application/json`, ByteString(Json.toJson(emails).toString())) -> emails.toVector
  }

  val requestBuilderFlow : Flow[Seq[String],(HttpRequest, Seq[String]),NotUsed] = Flow[Seq[String]] map requestBuilderFunc

  val responseHandler: ((Try[HttpResponse], Seq[String])) => (HttpResponse, Seq[String]) = {
   case (responseTry, context) =>
     logger.debug(s"Response: $responseTry")
      (responseTry.get, context.asInstanceOf[Seq[String]])
  }
}

Я должен признать, что я борюсь с тем, как организовать свои приложения Scala в объекты, черты, классы, функции высшего порядка и т. Д. И протестировать их.

Answer 1

То, что вы хотите сделать, это использовать что-то вроде внедрения зависимостей, чтобы внедрить Flow[(HttpRequest, Seq[String]), (Try[HttpResponse], Seq[String]), Any],

В производственном процессе этот поток будет поступать из akka http, но в тестовом режиме вы можете сами поиздеваться, чтобы вернуть все, что вам нужно.