play - scala for web development

Cómo usar una SourceQueue de Akka Streams con PlayFramework

Question

Me gustaría usar un SourceQueue para insertar elementos dinámicamente en una fuente Akka Stream. El controlador de juego necesita una Fuente para poder transmitir un resultado usando el método de chuncked .
Como Play usa su propio Akka Stream Sink debajo del capó, no puedo materializar la cola de la fuente utilizando un Sink porque la fuente se consumiría antes de que la utilice el método chunked (excepto si utilizo el siguiente hack).

Soy capaz de hacer que funcione si pre-materializo la cola de origen utilizando un editor de secuencias reactivas, pero es una especie de "truco sucio":

def sourceQueueAction = Action{
val (queue, pub) = Source.queue[String](10, OverflowStrategy.fail).toMat(Sink.asPublisher(false))(Keep.both).run()
//stupid example to push elements dynamically
val tick = Source.tick(0 second, 1 second, "tick")
tick.runForeach(t => queue.offer(t))
Ok.chunked(Source.fromPublisher(pub))
}

¿Hay una forma más sencilla de utilizar una fuente de Akka Streams con PlayFramework?

Gracias

Answer 1

La solución es usar mapMaterializedValue en la fuente para obtener un futuro de su materialización de la cola:

def sourceQueueAction = Action {
val (queueSource, futureQueue) = peekMatValue(Source.queue[String](10, OverflowStrategy.fail))
futureQueue.map { queue =>
Source.tick(0.second, 1.second, "tick")
.runForeach (t => queue.offer(t))
}
Ok.chunked(queueSource)
}
//T is the source type, here String
//M is the materialization type, here a SourceQueue[String]
def peekMatValue[T, M](src: Source[T, M]): (Source[T, M], Future[M]) = {
val p = Promise[M]
val s = src.mapMaterializedValue { m =>
p.trySuccess(m)
m
}
(s, p.future)
}

Answer 2

Me gustaría compartir la información que obtuve hoy, aunque puede que no sea apropiado para su caso con Play.

En lugar de pensar en una Source que desencadenar, a menudo se puede dar la vuelta al problema y proporcionar un Sink a la función que realiza la fuente.

En tal caso, el Sink sería la etapa de "receta" (no materializada) y ahora podemos usar Source.queue y materializarla de inmediato. Tengo cola Tiene el flujo que corre.