Add a way to map element to send only to a specific flow

One of the things that I miss is the ability to map an element to send to a specific flow preserving the old element at the same time.

Let me build my case.

For instance, there are a lot of scenarios that we consume a message from a queue, process it and then acknowledge it back to the broker, so we just:

def parseToDomainObject(m: Message): DomainObject = ???
def insertIntoDatabase(domainObject: DomainObject): Future[Unit]

QueueSource("my.queue")
  .map { message =>
    (message, parseToDomainObject(m))
  }
  .mapAsync(10) { case (message, object) =>
    insertIntoDatabase(object).map(_ => message)
  }
  .runWith(QueueAck.sink)

This is usually fine, since you don't use another flow.

If, for instance, insertIntoDatabase wasn't an asynchronous call but another flow, things gets a little uglier:

def parseToDomainObject(m: Message): DomainObject = ???

val databaseInsertionFlow: Flow[DomainObject, Done, Notused] = ???

QueueSource("my.queue")
  .map { message =>
    (message, parseToDomainObject(m))
  }
  //won't work unless we provide only DomainObject
  //but we can't do that since we want to preserve Message for the acknowledgement
  .via(databaseInsertionFlow)
  .runWith(QueueAck.sink)

So, we have two alternatives. First one is simpler and usually is what I see most people doing, which is materializing another stream inside the main one, like:

def parseToDomainObject(m: Message): DomainObject = ???

val databaseInsertionFlow: Flow[DomainObject, Done, Notused] = ???

QueueSource("my.queue")
  .map { message =>
    (message, parseToDomainObject(m))
  }
  .mapAsync(10) { case (message, object) =>
    Source
      .single(object)
      .via(databaseInsertionFlow)
      .map(_ => message)
      .runWith(Sink.head)
  }
  .runWith(QueueAck.sink)

While it solves most cases, we have three problems here:

• Materialization is very expensive, doing it for each element certainly is not recommended
• Although mapAsync(10) backpressures the stream, we don't know for sure that this is the recommended settings, neither if databaseInsertionFlow can handle that throughput, since materializing it for each element makes any backpressure mechanism useless.
• It's very verbose and not that nice to look at for new programmers

SourceWithContext and FlowWithContext seem to have the features to help us, but, both require to send the context downstream if you're not using map, mapAsync, filter and other stages like that, so, it's useless in our case.

If we want to make things right we, we gotta rely on the GraphDSL:

val databaseInsertionFlow: Flow[DomainObject, Done, Notused] = ???

def mapInputFlow: Flow[Message, DomainObject, NotUsed] = ???

val databaseInsertionFlowOnSteroids[Message, (Message, Done), NotUsed] = 
  Flow.fromGraph {
    GraphDSL.create() { implicit b =>
      import GraphDSL.Implicits._

      val bcast = b.add(Broadcast[In](2))
      val mapper = b.add(mapInputFlow)
      val zip = b.add(Zip[In, OutF]())
      val flow = b.add(databaseInsertionFlow)

      bcast ~> Flow[In].async ~> zip.in0
      bcast ~> mapper ~> flow ~> zip.in1

      FlowShape(bcast.in, zip.out)
    }
  }

//and now we can:
QueueSource("my.queue")
  .via(databaseInsertionFlowOnSteroids)
  .map { case (message, _) => message }
  .runWith(QueueAck.sink)

Now we solve most of our problems, but, there's one that's a huge blocker: Although GraphDSL is pretty amazing and incredibly powerful, it's not easy for newcomers and I don't think we should design basic stuff with it.

Sorry for all the speech and all the code but I wanted to explain a recurrent scenario for me at work and I cannot stop to think that it happens elsewhere too.

Things would have been simpler if we could just:

def parseToDomainObject(m: Message): DomainObject = ???

val databaseInsertionFlow: Flow[DomainObject, Done, Notused] = ???

QueueSource("my.queue")
  .viaMap(databaseInsertionFlow) { message => parseToDomainObject(message)) }
  .map { case (message, databaseInsertionResult) => message }
  .runWith(QueueAck.sink)

Also, viaMap implementation is quite simple I think:

def viaMap[In, InF, OutF, Mat](f: Flow[InF, OutF, Mat])(mapF: In => InF): Flow[In, (In, OutF), NotUsed] =
  Flow.fromGraph {
    GraphDSL.create() { implicit b =>
      import GraphDSL.Implicits._

      val bcast = b.add(Broadcast[In](2))
      val flow = b.add(f)
      val zip = b.add(Zip[In, OutF]())

      bcast ~> Flow[In].async         ~>      zip.in0
      bcast ~> Flow[In].map(mapF)  ~> flow ~> zip.in1

      FlowShape(bcast.in, zip.out)
    }
  }

So, I wonder if it makes sense to add this to the Akka Streams DSL or if it doesn't.

Waiting for feedbacks, thanks if you got until here. :)

[]'s