Is it possible to get a reference to a `Source#actorRef` before it is materialized?

Akka Akka Streams & Alpakka
,
1

I’m building a stream that will schedule and dispatch work with some fixed number of worker actors. I consume work in a Flow once my Source actor receives a message of type CanDoWork. However, my downstream currently has no way to tell the source that it can do work because the actor reference isn’t available when the Flow needs to be constructed. In short, I have a “loop”.

My attempt is here:

val workTrigger: Source[CanDoWork.type, ActorRef] =
  Source.actorRef(???, OverflowStrategy.dropNew)

val stream =
  workTrigger
    .via(Flow[CanDoWork.type].map(_ => schedulingFlow.schedule()))
    .to(Sink.ignore)

lazy val dispatcher = system.actorOf(Dispatcher.props(ref))
lazy val schedulingFlow = new SchedulingFlow(dispatcher)
lazy val ref = stream.run()

As can be seen, the Dispatcher needs a reference to the stream, which isn’t available until the stream has been created and materialized (with the dispatcher!).

How can I get this reference ahead of time? I’ve looked at mapMaterializedValue, but it seems like that won’t do the job.

2

Source#preMaterialize() came to the rescue!

val workTrigger: Source[CanDoWork.type, ActorRef] =
  Source.actorRef(???, OverflowStrategy.dropNew)

val (sourceActor, source) = workTrigger.preMaterialize()
val dispatcher = system.actorOf(Dispatcher.props(sourceActor))
val schedulingFlow = new SchedulingFlow(dispatcher)

val stream = source
  .via(Flow[ActorFreeMessage.type].map(_ => schedulingFlow.schedule()))
  .to(Sink.ignore)

stream.run()

sourceActor ! CanDoWork
1 Like