Multicast Requirement

[benjchristensen]

Currently the spec states "A Publisher can serve multiple subscribers subscribed dynamically at various points in time. In the case of multiple subscribers the Publisher should respect the processing rates of all of its subscribers (possibly allowing for a bounded drift between them)."

I think this is a mistake to complicate the spec and require implementations to support multicasting and therefore management of subscriptions over time. In fact, I think it should expressly stick to unicast (new lifecycle per Subscription).

Multicasting techniques should be layered on top by libraries, not required of the Publisher instances themselves.

For example, each Subscriber could result in a new network connection, open file, etc. This would be a basic implementation.

In Rx this greatly simplifies things and is a good separation of concern. Multicasting can be added and done with different behaviors such as replaying all of history, a portion of history, the last value only, or ignoring history and just starting from now onwards, etc.

In other words, someone providing me a Publisher should not concern themselves with multiple subscriptions, how I want to multicast or other such things. If I subscribe it should start a new lifecycle, emit the data as I request it and stop when I unsubscribe.

This keeps the mental model clear, the Publisher implementations simple and allows libraries a simple contract to interface with.

[normanmaurer]

I agree with @benjchristensen here :)

[rkuhn]

I fully agree with the part about not mandating multicast: it is a complication that does not pull its own weight (due to there being so many different useful variants that picking a default is too limiting).

The remaining question is whether or not it should be allowed to call subscribe twice on the same Publisher. The simplification in implementation only fully materializes if no multicast is required at all, meaning that a Publisher will only ever publish to at most one Subscriber. We have wording in the spec that defines the shut-down state as that which is entered after having been disposed by its downstream (currently called cancelation, but that is another discussion).

What you have in mind is the Rx way of the Subscription being the “active” entity in the sense of holding the state relevant between Observable and Observer, but I would prefer this spec to be more general and allowing the Publisher to be the active (asynchronous) entity.

Thinking about the user-visible contract I prefer very clear semantics:

• if you get a Publisher, then you can subscribe to it exactly once
• if you pass someone a Publisher, only do so for a “fresh” one (because the recipient can reasonably expect being able to subscribe to it)
• if you need to attach multiple Subscribers, make that explicit with a suitable fan-out combinator

This behavior matches Rx practices as far as I can see, the difference is that in Rx the Subject and any other Observable are not distinguished in the type system. In this spec I would propose to reserve Publisher for single-output streams, have an API element that explicitly supports fan-out (à la Rx Subject) and keep the high-level combinators DSL outside the scope as we currently have (i.e. the transformation methods on Observable are library specific and this spec does not say anything about them or their semantics).

[viktorklang]

👍 @rkuhn!

[benjchristensen]

I think a Publisher should be capable of being subscribed to as many times as wanted, otherwise a library must have a "Publisher Factory" and the user can't pass a Publisher around safely as once consumed it couldn't be used again.

If a Publisher can be subscribed to many times, it is the only type needed by a library and can be passed around and subscribed to whenever and however often they want.

I would not add anything to this spec for fan-out, that's for libraries to do.

[rkuhn]

There is a relationship between subscribing to a Publisher more than once and fan-out, which divides Publishers into two categories:

• If the Publisher just replays the same stream for all Subscribers independently, then there is no fan-out, and in effect you have one Publisher per Subscriber (i.e. the behavior is not distinguishable from such a design).
• If the Publisher only has one (live) stream which is sent to its Subscriber, then having no fan-out logic prohibits having a second Subscriber—that would be fan-out by definition!

As I said above, the semantics must be clear: if I get a Publisher (e.g. as a method argument) then I must know what that means. Saying “it can be used multiple times unless it cannot” violates this constraint. In the second case above there cannot be reuse since the point of this discussion is that we agree that having a fan-out logic in every Publisher is a bad idea. This is a difference to Rx Observable because for Publisher we would have to define how to merge the back pressure streams when splitting the data streams (which we cannot do in a one size fits all fashion).

Therefore the only consistent choice is to disallow a Publisher from being reused.

Having a source of data which can support multiple sinks is perfectly reasonable, but I don’t think it should be covered by this specification—at least not in the first version. This means that e.g. Observable should have a “toPublisher” method which you can use to get a back pressure aware outlet that can be connected to any Reactive Streams Subscriber, and you would implement it such that you get a new one every time. Would that not solve this issue in a quite natural fashion?

[benjchristensen]

Therefore the only consistent choice is to disallow a Publisher from being reused.

This just means we're forcing people to use a factory pattern for handing out Publisher instances rather than being able to use the Publisher directly. The Publisher should be the factory and each time subscribe is invoked it starts a new lifecycle. Multi-subscribe on a single Publisher is a far nicer API and contract.

Rx Observables work this way very elegantly. If I have an Observable anyone can subscribe to it as many times as they want and they'll each get whatever stream the Observable is going to send to it. I do not need to ship around an "Observable Factory" for someone to generate an Observable and then subscribe to it.

A Publisher should ALWAYS be capable of being subscribed to as many times as it wants. Each one though is independent. It is not multicast, it is unicast per subscription.

The ability to subscribe multiple times to a Publisher is not to be confused with multicast.

Multicast means it is sharing a stream. Ability to subscribe multiple times to a Publisher does not.

Multicast immediately means that a Publisher is required to maintain state and always share the stream, which is an incorrect thing to force or make every Publisher do.

Here are examples ...

1) Stock Stream - hot, always changing

Publisher: StockStream stream = new StockStream()
Consumer 1: stream.subscribe(subscriberA)
Consumer 2: stream.subscribe(subscriberB)

Two independent streams are flowing to subscriberA and subscriberB.
Each consumer starts receiving data from whenever it starts.

The Publisher is free to choose whether it has 2 separate network connections or 1.
Thus, it can do multicast internally if it wants, but it is not a requirement, nor is it expected to. Hot streams can always choose to do so though to conserve resources if they want.

2) Catalog Stream - cold, starts from beginning each time

Publisher: CatalogStream stream = new CatalogStream()
Consumer 1: stream.subscribe(subscriberA)
Consumer 2: stream.subscribe(subscriberB)

In this case the Publisher will establish a new network connection each time and start fresh.
Each consumer will receive all titles in the catalog from the beginning regardless of when they subscribe.

3) File Stream - cold, loads each time

Publisher: FileStream stream = new FileStream()
Consumer 1: stream.subscribe(subscriberA)
Consumer 2: stream.subscribe(subscriberB)

In this case the Publisher will most likely open and read the file each time and start fresh.
Each consumer will receive all bytes of the file regardless of when they subscribe.

Multicast vs Multiple Subscribe

A Publisher can be subscribed to multiple times and let each subscription be independent of each other (unicast). This trivial example demonstrates this:

public class AStreamExample implements Publisher<String> {

    public static void main(String[] args) {
        AStreamExample stream = new AStreamExample();

        MySubscriber s1 = new MySubscriber();
        MySubscriber s2 = new MySubscriber();
        stream.subscribe(s1);
        stream.subscribe(s2);
    }

    @Override
    public void subscribe(final Subscriber<String> subscriber) {
        // you wouldn't really implement it like this ... but you get the point
        new Thread(new Runnable() {

            @Override
            public void run() {
                System.out.println("Subscribed on thread: " + Thread.currentThread());
                BooleanSubscription s = new BooleanSubscription();
                subscriber.onSubscribe(s);
                // ignoring unsubscribe and requestMore behavior for this simple example
                subscriber.onNext("1");
                subscriber.onNext("2");
                subscriber.onComplete();
            }

        }).run();
    }

}

A Publisher is the factory. Each time subscribe is invoked it kicks off the work.

It is not required to multicast. In fact, it is assumed it will unicast. If a backing stream naturally multicasts (a hot stream of events that never stops) then it could choose as an implementation decision to do multicasting. This ties back to the 3 example use cases above with "hot" and "cold" sources.

If we didn't allow multi-subscribe, then the code would need to look like this:

    public static void main(String[] args) {
        MySubscriber s1 = new MySubscriber();
        MySubscriber s2 = new MySubscriber();
        new AStreamExample().subscribe(s1);
        new AStreamExample().subscribe(s2);
    }

This seems trivial in this case, but put it behind an API and it becomes more clear:

    public static void main(String[] args) {
        MySubscriber s1 = new MySubscriber();
        MySubscriber s2 = new MySubscriber();
        getStream().subscribe(s1);
        getStream().subscribe(s2);
    }

Do I have to call getStream() each time I subscribe? If so, why? The subscribe methods says I can subscribe so why not multiple times?

It gets worse once someone starts passing the stream around.

    public void doStuff(AStreamExample stream) {
        stream.subscribe();
        doMoreStuff(stream);
    }

    public void doMoreStuff(AStreamExample stream) {
        // can I subscribe? what if someone else already did?
        stream.subscribe();
    }

Perhaps my original statement was unclear, but I'm not saying a Publisher can't do multicast on a live stream, but that a Publisher should not be required to do multicast as it does not cater to all use cases. There are also many multicast use cases that are better suited to libraries (such as replay with its many different variants).

I suggest we remove the multicast requirement for a Publisher but require that subscribe be capable of being invoked multiple times. It is then left to the Publisher to decide whether it does multicast or unicast based on its use cases, resource usage, etc.

[rkuhn]

@benjchristensen, thanks for your elaboration, now it is perfectly clear what we are talking about: you are equating Publisher with Observable, perhaps because of similarities in method names and also due to the existence of Subscription as a name shared between Rx and Reactive Streams.

To be very blunt and concise: these names have misled you, and I am very sorry about that.

The Publisher/Subscriber pair is designed only for one purpose: to transport exactly one stream of elements across an asynchronous boundary (without unbounded buffering etc.). Therefore a Publisher delivers exactly one stream, which is why multiple subscriptions and multicast are identical in this case.

Thinking about it some more, it would probably have been better to leave out the Subscription altogether—we had it since we started out from the position of building in multicast as a convenience feature that we have come to realize is detrimental—which would lead us to equate Publisher with an Rx Subscription, leaving the details of what you want to publish out of the scope of this specification.

What you want is to have an easy path towards using Reactive Streams as a way to express remote Observables, and what we want is to use it as a way to push data to an Actor without blowing up message queues, and as far as I know everyone agrees on the underlying basic problem set. I’ll try out what the above proposal would look like, and if it makes sense (i.e. passes my plausibility test) then we can continue the discussion on the resulting pull request.

Again, I’m sorry that it took so long for me to realize what the root cause of the misunderstanding was.

[benjchristensen]

Roland, this is not about Observables (though that heavily influences my perspective as I've spent almost 2 years building "reactive stream" applications in production).

Let me equate it with Iterable/Iterator instead so there is no chance of conflict with Observable.

In code, I can pass an Iterable around to anyone and anything can "subscribe" to it via a new Iterator.

Since it's pull based, it creates the Iterator for me on a getIterator() method or something along those lines.

The getIterator() method can be called n times and every time will get a clean "subscription" to iterating the Iterable.

Now we flip to a Publisher/Subscriber. Since it's push, I provide the Subscriber instead of calling getSubscriber(). It is via the Publisher.subscribe(Subscriber s) that I start the push-based iteration (or streaming).

The point I'm making is two-fold:

1. The Publisher should be usable like an Iterable so that developers can pass it around and allow anyone to subscribe to it whenever and however many times they want. Otherwise there ends up needing to be a 3rd component, a PublisherFactory that would be used like PublisherFactory.getPublisher().subscribe(Subscriber s).

2. Multicast is an option, not a requirement that is up to the Publisher to decide upon based on what they're streaming.

Therefore, please put aside comparisons to or use cases of Actors and Observables and consider this being an interface in the JDK for decades to come that everyone will implement. In that case, it will no longer be used purely for interop (as it will be now since it's a 3rd-party dependency). If it becomes part of the JDK it will be implemented directly by classes and be the common interface, like Iterable, and should therefore be capable of being used like an Iterable without a factory being required.

Code such as the following should be safe to write:

doStuff() {
  Publisher p = new PublisherImpl();
  doMoreStuff(p)
  p.subscribe(subscriber);
}

doMoreStuff(Publisher p) {
  p.subscribe(subscriber);
}

If we don't allow multi-subscribe behavior this will instead need to be written with a 3rd level such as:

doStuff() {
  PublisherFactory p = new PublisherFactoryImpl();
  doMoreStuff(p)
  p.getPublisher().subscribe(subscriber);
}

doMoreStuff(PublisherFactory p) {
  p.getPublisher().subscribe(subscriber);
}

[jrudolph]

@benjchristensen What about streams that can naturally only be used once? Like a stream of network data for an already opened network connection? Given your suggestion would it be possible to get a Publisher for a stream like that?

IMO either, you would have to allow that this Producer would instantly call onError for all subsequent subscriptions after the first, or you need some buffering logic / fan-out logic that allows consumers to join the game later.

[benjchristensen]

It's up to the Publisher whether it wants to reuse the underlying network connection or not.

See the 3 use cases I listed above at #19 (comment)

[jrudolph]

Yes, I've seen them. That's why I ask. I don't think an open connection fits into any of those three categories. An already open network connection (similar to an already generated Iterator) has very different backpressure needs than something that publishes events. "Reuse" is the thing that doesn't work easily because the backpressure demand of all possible consumers have to be coordinated somehow.

So, that's the question: can an open network connection (an iterator, an InputStream) be a Publisher in your view (in Rx)?

[benjchristensen]

There are two cases where an open network stream applies.

1. Hot

Assume a "hot" stream where back pressure is not relevant, it's stock prices, mouse events, system metrics or some other thing that continues to fire at you. You can't say "stop", all you can say is "I can't take any more". In that case it's up to the developer to choose whether they want to drop, sample, debounce, buffer (rarely useful in this case). The "requestMore" behavior from a Subscriber can be translated into one of these mechanisms, but that's up to each implementation and the developer to choose.

In a "hot" case like this, a Publisher can safely reuse the network connection because it's going to stream the exact same data whether a single connection or multiple connections are open. There is no buffering, no state per Subscriber, just a flow of data. It is transparent to the Subscriber whether there are network connections being shared or not, it's just a hot stream of data that they can throttle on if they wish. Each Subscription will manage the requestMore behavior independently of each other, and each one will just be doing throttling of some flavor, the Publisher just keeps emitting the data.

Summary:

• single Publisher
• multiple Subscriber instances each with their own Subscription
• single network connection (probably only open if there is >0 subscribers)

2. Cold

A "cold" source is very different. In this case the Publisher would need to create a new network connection for each Subscription because "requestMore" is now meaningful.

Each Subscriber will request however much it can handle and the state of data flowing is one-to-one between the underlying resource. The single Publisher can be subscribed to multiple times, but each time a new Subscription is created it will represent a separate network connection that only serves that Subscriber/Subscription pair.

Summary:

• single Publisher
• multiple Subscriber instances each with their own Subscription
• network connection per subscription

In short, the Publisher is the factory that returns a Subscription to the Subscriber that manages the correct state and behavior (via the "requestMore" invocations) depending on what fits the use case.