Create Source from Sink #25150

cedretaber · 2018-05-27T07:55:08Z

Refs #24853

Add connect to Source.
It easily connects a source to a sink, based on #24853 (comment).

lightbend-cla-validator · 2018-05-27T07:55:09Z

Hi @cedretaber,

Thank you for your contribution! We really value the time you've taken to put this together.

Before we proceed with reviewing this pull request, please sign the Lightbend Contributors License Agreement:

http://www.lightbend.com/contribute/cla

akka-ci · 2018-05-27T07:56:04Z

Thank you for your pull request! After a quick sanity check one of the team will reply with 'OK TO TEST' to kick off our automated validation on Jenkins. This compiles the project, runs the tests, and checks for things like binary compatibility and source code formatting. When two team members have also manually reviewed and (perhaps after asking for some amendments) accepted your contribution, it should be good to be merged.

For more details about our contributing process, check out CONTRIBUTING.md - and feel free to ask!

johanandren · 2018-05-29T10:30:16Z

OK TO TEST

akka-ci · 2018-05-29T10:43:37Z

Test FAILed.

johanandren · 2018-05-29T11:00:34Z

Seems the docs are misplaced, but let's discuss the design first:

Is it really a good idea to run the graph, I don't think any other factories takes a materializer and runs anything?
If we should run the graph, could we raise the abstraction level even further and take one side as a parameter to make it easier to use/understand?
I don't think connect or connection as in Heikos original version is a good name, to me it does not convey at all what it does. It would be better with something along the lines of fromSink, toSink, asSink, sinkToSource, sinkAndSource?
Additional design consideration does it really belongs on Source or if it would fit better on Sink or maybe even Flow given that it is sort of a flow?

cedretaber · 2018-05-29T13:51:54Z

Thank you for your review!

Yes, it seems to make sense that there is no need to run the graph.
I have no idea what it's name is. However, I could feel that sinkAndSource is better.
As Flow has some methods which creates a flow from a sink and a source, it might be in Flow.

johanandren · 2018-05-30T09:56:25Z

With regards to Flow that makes somewhat sense, but it isn't returning a Flow. Maybe it wouldn't be the first place you'd look if the reverse-adapter-thingie is what you are looking for.

I think I like Souce.sinkToSource best, for discoverability and saying what it does.

akka-ci · 2018-05-30T15:48:38Z

Test FAILed.

akka-ci · 2018-05-30T16:03:32Z

Test FAILed.

ktoso · 2018-05-30T16:11:01Z

akka-docs/src/main/paradox/stream/operators/Source/sinkToSource.md

+
+@@@div { .callout }
+
+@@@


we'd need examples as well, from nowadays we should always right away include examples when we merge new operators. See here how: https://raw.githubusercontent.com/akka/akka/master/akka-docs/src/main/paradox/stream/operators/Sink/actorRefWithAck.md

Should I create SourceDocSpec.scala , or write the tests in IntegrationDocSpec.scala ?

ktoso · 2018-05-30T16:11:21Z

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala

+      .toMat(Sink.asPublisher[M](fanout = false))(Keep.both)
+      .mapMaterializedValue {
+        case (sub, pub) ⇒ (Sink.fromSubscriber(sub), Source.fromPublisher(pub))
+      }


needs to get a javadsl as well, on javadsl.Source

ktoso · 2018-05-30T16:11:39Z

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala

+   * >   Sink   |------>|  Source  >
+   * +----------+       +----------+
+   *
+   * Should be provided by Akka Streams, see https://github.com/akka/akka/issues/24853.


No need for such comment, this is IN akka streams then after all ;-)

ktoso · 2018-05-30T16:11:59Z

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala

+   * +----------+       +----------+
+   * >   Sink   |------>|  Source  >
+   * +----------+       +----------+
+   *


This needs to be put into a

{{{ ... }}}

for formatting reasons

ktoso · 2018-05-30T16:13:20Z

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala

+      .toMat(Sink.asPublisher[M](fanout = false))(Keep.both)
+      .mapMaterializedValue {
+        case (sub, pub) ⇒ (Sink.fromSubscriber(sub), Source.fromPublisher(pub))
+      }


code looks fine though :)

ktoso · 2018-05-30T16:14:21Z

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala

+   *
+   * Should be provided by Akka Streams, see https://github.com/akka/akka/issues/24853.
+   */
+  def sinkToSource[M]: RunnableGraph[(Sink[M, NotUsed], Source[M, NotUsed])] =


We need to ponder the name some more...
I have no good name today in mind though...

This is the best I can come up with

Regarding the name: This seems like the opposite of Flow#fromSinkAndSource. It seems like this could live on Flow (the class not the object), the idea would be like breaking an existing flow into a sink and source.

So something like:

class Flow[-In, +Out, +Mat] { . . . def toSinkAndSource: (Sink[In],Source[Out]) = Source .asSubscriber[In] .via(this) .toMat(Sink.asPublisher[Out](fanout = false))(Keep.both) .mapMaterializedValue { case (sub, pub) ⇒ (Sink.fromSubscriber(sub), Source.fromPublisher(pub)) }.run()

Of course, you have to run it in order to actually get the sink and the source, so that might not be desirable.

akka-ci · 2018-09-18T12:37:09Z

Test FAILed.

hseeberger · 2018-11-29T10:56:48Z

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala

+   *
+   * Should be provided by Akka Streams, see https://github.com/akka/akka/issues/24853.
+   */
+  def sinkToSource[M]: RunnableGraph[(Sink[M, NotUsed], Source[M, NotUsed])] =


Why add this method to Sink? It creates both a sink and source on materialization. I'd rather put it into RunnableGraph.

See my comment
#25150 (comment)

RunnableGraph would perhaps make sense but it would be completely undiscoverable as there are no other factories (fromGraph is the only one) there.

hseeberger · 2018-11-29T10:57:20Z

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala

+   *
+   * Should be provided by Akka Streams, see https://github.com/akka/akka/issues/24853.
+   */
+  def sinkToSource[M]: RunnableGraph[(Sink[M, NotUsed], Source[M, NotUsed])] =


M is typically used for the type of the materialized value. Please change to A or T.

cedretaber · 2018-12-02T07:42:26Z

@hseeberger
Thank you for reviewing!
I am sorry it is so busy now, please wait for a couple of weeks to respond.

akka-ci · 2019-01-02T15:38:54Z

Test FAILed.

akka-ci · 2019-03-28T17:03:28Z

Test FAILed.

cedretaber · 2019-03-29T13:27:11Z

I'm sorry, but in the present situation I do not have enough energy to discuss and promote this feature.
I apologize for the lack of my power and skills.

May I close this pull request?

I think that who have strong motivation for it should create the pull request again.

patriknw · 2019-04-15T07:40:30Z

no problem, thanks for your efforts

He-Pin · 2022-04-29T22:37:58Z

@cedretaber Hi @cedretaber would you like to continue this PR?

cedretaber · 2022-04-29T23:32:29Z

@hepin1989
At this moment, I have no intention to continue.

Add connect to Source

7feb987

akka-ci added validating PR is currently being validated by Jenkins needs-attention Indicates a PR validation failure (set by CI infrastructure) and removed validating PR is currently being validated by Jenkins labels May 29, 2018

Fix the method up following a comment

c599f92

akka-ci added validating PR is currently being validated by Jenkins and removed needs-attention Indicates a PR validation failure (set by CI infrastructure) labels May 30, 2018

Add a doc for Source.sinkToSource

781d9d8

cedretaber force-pushed the issue-24853-source-from-sink branch from df4b894 to 781d9d8 Compare May 30, 2018 15:48

akka-ci added needs-attention Indicates a PR validation failure (set by CI infrastructure) and removed validating PR is currently being validated by Jenkins labels May 30, 2018

ktoso reviewed May 30, 2018

View reviewed changes

hseeberger suggested changes Nov 29, 2018

View reviewed changes

patriknw closed this Apr 15, 2019

raboof mentioned this pull request Sep 1, 2021

Source from sink an vice versa #24853

Open

mdedetrich mentioned this pull request Dec 23, 2023

Add a way to create a Source from a Sink to Akka Streams DSL apache/pekko#865

Open


		@@@div { .callout }

		@@@

Create Source from Sink #25150

Create Source from Sink #25150

Uh oh!

Conversation

cedretaber commented May 27, 2018

Uh oh!

lightbend-cla-validator commented May 27, 2018

Uh oh!

akka-ci commented May 27, 2018

Uh oh!

johanandren commented May 29, 2018

Uh oh!

akka-ci commented May 29, 2018

Uh oh!

johanandren commented May 29, 2018

Uh oh!

cedretaber commented May 29, 2018

Uh oh!

johanandren commented May 30, 2018

Uh oh!

akka-ci commented May 30, 2018

Uh oh!

akka-ci commented May 30, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

akka-ci commented Sep 18, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cedretaber commented Dec 2, 2018

Uh oh!

akka-ci commented Jan 2, 2019

Uh oh!

akka-ci commented Mar 28, 2019

Uh oh!

cedretaber commented Mar 29, 2019

Uh oh!

patriknw commented Apr 15, 2019

Uh oh!

He-Pin commented Apr 29, 2022

Uh oh!

cedretaber commented Apr 29, 2022

Uh oh!

Uh oh!