There are two ways to do that.

The code is available under this commit . If you check out and execute this ref, you’ll find our game runs exactly the same, meaning we accomplished our goal.

Now all that’s left is to replace our Sink.ignore with the queue…​

Let’s add a storage var for this queue to our MainScreen , as well as a type alias for abstracting away implementation details:

However, the most convenient method for us here is Sink.queue . Materializing it generates a SinkQueue that can be polled for every separate element that arrives at the sink.

We turn to the nice list of available stages , again, from the docs . Happily, we find there’s a couple of sink stages that produce "dynamic" output, e.g. actorRef which is a dual to Source.actorRef .

However, materialization can provide an arbitrary number of values , as illustrated by this diagram from the docs :

We’ve already taken into advantage of materialization to obtain an input to our stream, by way of a materialized ActorRef .

Assigning a custom dispatcher

(This part is optional to the understanding of the text as a whole.)

On first glance, it seems it would be possible to instead define a Sink.foreach stage with a dispatcher that runs on the game’s UI thread.

Furthermore, it sounds like the Testkits’s CallingThreadDispatcher would be the way to go. Let’s try that out

BulletHell.scala val graph = tickSource .via(setUpLogic(List(generator, mover, worldUpdater, tickIncrementer))) (1) .to(Sink.foreach[Seq[Action]](actions => { actions.foreach(_()) (2) println(Thread.currentThread().getName) (3) }).withAttributes(Attributes(Dispatcher(CallingThreadDispatcher.Id)))) (4) tickActor = Some(graph.run()) (5)

1 Same code as in the SinkQueue variant. 2 Iterating and calling all actions. 3 Debug statement to see whether we’re on the current thread. 4 Dispatcher’s in Akka are created dynamically from the configuration and referenced by IDs. 5 Going back to the same code as in the previous part of this blog.

However, when we run it, we would see e.g.:

game-akka.actor.default-dispatcher-4 game-akka.actor.default-dispatcher-3 game-akka.actor.default-dispatcher-4 game-akka.actor.default-dispatcher-3 game-akka.actor.default-dispatcher-4

whereas the UI thread is named LWJGL Application . If we think for a minute, that makes sense - the calling thread here is a thread from the pool used by the previous stages (the ones generating the actions, or rather the ZipN substage).

So, what we would need to do is to setup a custom dispatcher that would enqueue Runnable s , and expose a method named e.g. popAndRun() that would take the first Runnable out of the queue and execute it in the current thread. This popAndRun() would then be invoked during MainScreen#render() .

However, since dispatchers cannot be created programmatically, we would need to employ some trickery in order to invoke such a method. I think it would work, specifically through the Dispatcher -lookup system and several casts, but would be a hacky way to implement what we already have achieved anyway.