Using dynamic sink destination in Akka Streams

Viewed 1205

I'm rather new to Akka and am trying to learn the basics. My use case is to continually read messages from a JMS queue and output each message to a new file. I have the basic setup working with:

Source<String, NotUsed> jmsSource =
  JmsSource
    .textSource(JmsSourceSettings
    .create(connectionFactory)
    .withQueue("myQueue")
    .withBufferSize(10));

Sink<ByteString, CompletionStage<IOResult>> fileSink =
  FileIO.toFile(new File("random.txt"));

final Flow<String, ByteString, NotUsed> flow = Flow.fromFunction((String n) -> ByteString.fromString(n));

final RunnableGraph<NotUsed> runnable = jmsSource.via(flow).to(fileSink);

runnable.run(materializer);

However, I want the file name to be dynamic (and not hard-coded to "random.txt"): it should be changed depending on the content of each message on the queue. I could, of course, pick up the file name in the flow, but how do I set that name in fileSink? How do I best set this up?

2 Answers
Related