make Source.subscriber().to(Sink.publisher()) work

[rkuhn]

Currently this is not supported by the materialization scheme, it will simply not do anything since both ends of this connection are “lazy”. The materializer should realize this and insert a dummy-Processor.

[drewhk]

Note: the test case should also include the variants:

Sink.subscriber().withAttributes(...).to(...)

.. and the same with Sink, since withAttributes introduces wrapping in the internal layout.

[mehmeteking]

I plan to tackle this issue. If OK, on which branch should I work?

[ktoso]

Cool, thanks! If you'd like to give it a shot then branch off release-2.3-dev and check contributing.md for general development hints.

[viktorklang]

What's the status here?

[rkuhn]

Do you have a plausible use-case?

[viktorklang]

File copying?

[drewhk]

WDYM file copy? How does it relate to Source.publisher() and Sink.subscriber?

[viktorklang]

imagine: def copy(p: Publisher[ByteString], s: Subscriber[ByteString]) = ??? What if `p` is provided by an end-user, and `s` is provided by a library writer, the risk is that a certain combination won't work at all, because of the situation above. Highly surprising, and hard to debug, don't you think?

[drewhk]

I think there is a misunderstanding here. What you want works:

def copy(p: Publisher[ByteString], s: Subscriber[ByteString]) = Source(p).to(Sink(s)).run()

The ticket here is about the Sink and Source that materializes to a Publisher and Subscriber. Source.pubisher.to(Sink.subscriber) should translate to an identity Processor since it does nothing with the elements.

[viktorklang]

No, no misunderstanding, I am talking about a method defined outside of
Akka.

On Sat, Jun 6, 2015 at 9:31 AM, drewhk notifications@github.com wrote:

I think there is a misunderstanding here. What you want works:

def copy(p: Publisher[ByteString], s: Subscriber[ByteString]) = Source(p).to(Sink(s)).run()

The ticket here is about the Sink and Source that materializes to a
Publisher and Subscriber. Source.pubisher.to(Sink.subscriber) should
translate to an identity Processor since it does nothing with the elements.

—
Reply to this email directly or view it on GitHub
#16923 (comment).

Cheers,
√

[drewhk]

But the code example you pasted is very easy to implement and has nothing to do with the bug discussed in this ticket. What is the real world example you think is problematic?

[viktorklang]

My point is that any generic code that hooks up Publishers to Subscribers
could be "vulnerable".
And hooking up a Publisher to a Subscriber is the entire point of RS, so it
should work, or do you disagree?

On Sat, Jun 6, 2015 at 10:18 AM, drewhk notifications@github.com wrote:

But the code example you pasted is very easy to implement and has nothing
to do with the bug discussed in this ticket. What is the real world example
you think is problematic?

—
Reply to this email directly or view it on GitHub
#16923 (comment).

Cheers,
√

[drewhk]

But in this bug there are no hooking up of Publishers to Subscribers. As I showed in my example snippet above, that scenario works perfectly fine.

Source(publisher) is not the same as Source.subscriber(). The latter is in reality a symbolic notation to express that you want the Subscriber interface of the stage to the right, while Sink.publisher is a symbolic notation that you want to access the Publisher of the stage to the left. Neither of these modues (the sink and the source) introduce a new Processor, they always expose the Processors to the left or right. For example

Source.subscriber.via(someProcessor).toMat(Sink.publisher)(Keep.both)

Will materialize into a pair of Subsriber and Publisher, directly giving access to someProcessor (or if it is a chain, then to the leftmost Subscriber and the rightmost Publisher). When you say

Source.subscriber.toMat(Sink.publisher)(Keep.both)

Then you basically instructed the framework to take the subscriber of the stage on the right, but that stage is instructed to take the publisher of the stage to the left. I.e. this does not work right now. Of course

Source.subsciber.via(Flow[T]).toMat(Sink.publisher)(Keep.both)

does work, since the empty Flow will materialize to an identity processor, and the materialized result will give you its Subscriber and Publisher interfaces.

This ticket is a real bug of course, but it is not trivial to fix, and is a really weird corner case.

[drewhk]

I try to fix it now, I might have a good solution...

[jroper]

Could I check, does this fix also fix the NPE thrown when you run this against RC3:

Source(1 to 3).runWith(
  Sink.publisher[Int].mapMaterializedValue(p => Source(p).runFold(0)(_ + _))
)

[drewhk]

Ok, file a ticket. I think it is because of the mapping functions on materialized values are evaluated during materialization, which is not safe in this case.

Btw, why do you need this? Why not

Source(1 to 3).runFold(...)

or

Source(1 to 3).to(Sink.fold(...)).run()

You only need Sink.publisher if you need access to the raw Publisher, which you usually need at the end of the stream and only if you are working with other RS implementations that accept a Publisher. Even then, if they don't need a Publisher passed, but they can provide a Subscriber instead you are better off with

Source(1 to 3).to(Sink(3rdPartySubscriber)).run()

[jroper]

My use case is actually to flatten Future[Sink[E, Future[A]]] to Sink[E, Future[A]], the example code above was just trying to simplify it into a reproducable test. Here's my actual code:

val future: Future[Sink[T, Future[M]]] = ...
Sink.publisher[T].mapMaterialized { publisher =>
  future.recover {
    case error => Sink.cancelled[T].mapMaterialized(_ => Future.failed(error))
  } flatMap { sink =>
    Source(publisher).toMat(sink)(Keep.right).run()
  }
}

If you can see another way to do this (especially if it doesn't require a materialiser), I'm very interested to know.

[drewhk]

Hmm, why do you need a Future[Sink] in the first place? There is no flattening for Sinks, but It is possible to flatten Sources like

Source(sourceFuture).mapAsync(identity).flatten(FlattenStrategy.concat)

But the issue here is that it is not idiomatic, since Source/Sink are lazy by default, i.e. it only executes when the stream runs, not before, so it is lazy already. Can you give me the use case for Future[Sink]? I think it might be better just to create a new Sink that solves your problem.

Btw, in your case, if you want to still do the hacky way, just use fanoutPublisher:

Sink.fanoutPublisher(4, 4).mapMaterialized ...

The above will work, because it creates an actual actor. The problem with Sink.publisher is that it is just a marker that "I want to access the Publisher to my left", which is problematic during materialization since "to the left" might not have been started up yet. In other words, Sink.publisher does not create anything runnable, it always piggybacks on the existing stages.

[jroper]

Play Framework actions return a Sink to consume the body - they essentially look like this:

RequestHeader => Sink[ByteString, Future[(ResponseHeader, Source[ByteString])]

Now, let's say that body is being uploaded to S3 directly, so the Sink will be the HTTP client. But before we even make the request, we need to authenticate the user, and that authentication is asynchronous. So we'll have a Future of a Sink to consume the body, which means we need to flatten it to a Sink before we can consume it.

Of course, we could change the API to return a Future of a Sink, instead of just a Sink, but conceptually a Future of a Sink is just a Sink that won't do anything in response to its onSubscribe method until sometime in the future, so that feels like it would be making the API needlessly complex.

[rkuhn]

Thanks for this example, @jroper, makes perfect sense to me. So it’s Reactive Iteratees all over :-)

[viktorklang]

Reacteratees, folks! You heard it here first!

Cheers,
√
On 17 Jun 2015 08:33, "Roland Kuhn" notifications@github.com wrote:

Thanks for this example, @jroper https://github.com/jroper, makes
perfect sense to me. So it’s Reactive Iteratees all over :-)

—
Reply to this email directly or view it on GitHub
#16923 (comment).

[drewhk]

@jroper ok, that makes sense. Currently just use the fanoutPublisher workaround which I posted above, and hopefully we will fix the original problem in the meantime.