如何在PlayFramework中使用Akka Streams SourceQueue

Question

我想使用SourceQueue将元素动态推送到Akka流源中。Play controller需要一个源，以便能够使用chuncked方法对结果进行流式传输。
由于Play使用自己的Akka流接收器，所以我无法使用接收器来实现源队列，因为源在chunced方法使用之前就会被消耗（除非我使用下面的hack）。

如果我使用reactive-streams发布器预物化源队列，我就可以使它正常工作，但这是一种“肮脏的攻击”：

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

有没有更简单的方法来使用Akka Streams SourceQueue和PlayFramework？

谢谢

Answer 1

解决方案是在源上使用MapMaterializedValue来获得其队列物化的未来：

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