Add a way to create a `Source` from a `Sink` to Akka Streams DSL

This has probably been top of the list when it comes to dealing with commonplace limitations when using pekko-streams, which is the fact that its not easy to create a Source when a Sink completes/materializes that also respects backpressure/cancellation/errors. While it may be true that that a Sink should always be the final step in your stream logic there is some impedance mismatch here as there are cases where you need to create a Source from a Sink that props up in practice.

As an example, imagine if you are using the pekko-http Flow API where you represent your entire route as a pekko stream. Lets say your route does a simple thing where it just pushes a message to Kafka and you only want to return a response when sending the message is successful (as is customary).

Understandably the act of sending a message to a Kafka is a Sink which is where the problem lies, once you send the message to Kafka thats the end of your stream business logic and in this specific case of using the pekko-http flow API, we have to get from a Sink[ProduceRecord[_,_],_] to a Source[HttpResponse, _] in order to return a response to the Http request.

It is currently possible to get around this by using techniques such as materializing the sink and then creating a Source from that using Sink.fromSubscriber/Source.fromPublisher but this creates its own issues as only cancellations are propagated, not errors so its ideal to just create a proper API for this, i.e. Source.fromSink(...) or something along these lines.

Existing Akka/Lightbend issues/discussions on this topic. Do note that all actual code contributions on this topic have been done by external contributors (i.e. no one from Lightbend/Akka) and the code never ended up being merged.

@He-Pin @jxnu-liguobin Not sure if you want to look into this?

apache / pekko

Add a way to create a `Source` from a `Sink` to Akka Streams DSL #865