Тестирование Akka Реактивных Streams

Question

Я тестирую код, потоки сообщений через исходящее соединение с потоком TCP, полученные с помощью:

(IO(StreamTcp) ? StreamTcp.Connect(settings, address)) 
.mapTo[StreamTcp.OutgoingTcpConnection] 
.map(_.outputStream) 

В моих тестах, я заменяю в результате Subscriber[ByteString] с фиктивным абонентом, вызвать некоторые исходящие сообщения, и утверждают, что прибыли, как ожидалось. Я использую метод ниже для создания фиктивного подписчика и будущего потока. (До сих пор, так хорошо)

def testSubscriber[T](settings: FlowMaterializer)(implicit ec: ExecutionContext): (Subscriber[T], Future[Seq[T]]) = { 
    var sent = Seq.empty[T] 
    val (subscriber, streamComplete) = 
    Duct[T].foreach(bs => sent = sent :+ bs)(settings) 
    (subscriber, streamComplete.map(_ => sent)) 
} 

Мой вопрос заключается в следующем: есть ли какой-то канонический метод тестирования, что потоки вывода ожидаемых значений, что-то похожее на Акка-х TestActorRef? А если нет, есть ли какая-то библиотечная функция, подобная указанной выше функции?

Answer 1

Тестирование потоков возможно с помощью akka-streams-testkit.
Подробнее об этом здесь: http://doc.akka.io/docs/akka/current/scala/stream/stream-testkit.html