PublishSubject is broken?

[andrask]

Can I ask someone to explain me why this code terminates without writing out anything?

PublishSubject<String> s = PublishSubject.create();

Subscription subscribe = s.replay(1).doOnNext(new Action1<String>() {
    @Override
    public void call(String t1) {
        System.out.println(t1);
    }

}).subscribe();

s.onNext("hello");

While debugging, I see that the subject has no subscribers. True, there are no direct subscribers but the replay() creates a multicast connection.

[akarnokd]

replay() creates a ConnectableObservable which you need to call connect() on, otherwise it won't do anything.

ConnectableObservable co = s.replay(1);
co.doOnNext(v -> { System.out.println(v); }).subscribe();
co.connect();
s.onNext("hello");

[andrask]

Thank you, that explains the operation. And also shows that the connect is a must, so it is not possible to chain the calls as I did. Then what is the point of having replay() return anything that resembles an Observable? Or is there some autoconnect method too?

[akarnokd]

There isn't any simple operator for that, though there is an enhancement request (somewhere among the issues) to add those variants. The closest auto-connect version is this:

s.replay(o -> o, 1).doOnNext(...).subscribe();

[andrask]

Ugh, this doesn't look good. Especially in Java if that nop function has no static method creator. Thanks again.

[headinthebox]

Note that if you think you need subjects, there is a 99% chance you can find a more elegant solution that does not require subjects. Really.

[benjchristensen]

There isn't any simple operator for that

There is refCount() that does auto-connect.

[andrask]

Thank you for the support. I'll look at refCount().

At the same time, I see that the API of replay() is exactly the same as the .NetRx version's. So I guess, the semantics is the same in both implementations. But the documentation is very ambiguous in both places. (At least for me as a non-native English speaker.) It says: "can be connected and disconnected". It doesn't mention that nothing is going to happen until is is connected because it starts in disconnected state. Nor does replay() mention that it starts disconnected. The modifier "able" in the ConnectAbleObserver feels like it's just an addition and doesn't change semantics of the Observer interface. But this is not true and it leads to misuse of the API.

Probably an explicit note in the documentation about replay() returning a disconnected by default Observer will help solve this confusion.

@headinthebox I know, I usually try to avoid subjects. However, I have a two way causal chain in my case. In one way goes the calculation result, in the other way goes the cancellation. It's possible to implement without subjects but it will not differ much from what subjects already provide and it would break the scope of my signals. Currently every signal is used only inside a single object and it is pretty easy to follow it this way. Also, it makes debugging a lot easier. But thanks for the note.

[DavidMGross]

The wiki documentation is more explicit about the need to call connect():

https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators

On Fri, May 9, 2014 at 12:00 PM, andrask notifications@github.com wrote:

Thank you for the support. I'll look at refCount().

At the same time, I see that the API of replay() is exactly the same as
the .NetRx version's. So I guess, the semantics is the same in both
implementations. But the documentation is very ambiguous in both places.
(At least for me as a non-native English speaker.) It says: "can be
connected and disconnected". It doesn't mention that nothing is going to
happen until is is connected because it starts in disconnected state. Nor
does replay() mention that it starts disconnected. The modifier "able" in
the ConnectAbleObserver feels like it's just an addition and doesn't change
semantics of the Observer interface. But this is not true and it leads to
misuse of the API.

Probably an explicit note in the documentation about replay() returning a
disconnected by default Observer will help solve this confusion.

@headinthebox https://github.com/headinthebox I know, I usually try to
avoid subjects. However, I have a two way causal chain in my case. In one
way goes the calculation result, in the other way goes the cancellation.
It's possible to implement without subjects but it will not differ much
from what subjects already provide and it would break the scope of my
signals. Currently every signal is used only inside a single object and it
is pretty easy to follow it this way. Also, it makes debugging a lot
easier. But thanks for the note.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/1180#issuecomment-42701498
.

David M. Gross
PLP Consulting

[andrask]

@DavidMGross Adding that single line to the replay() documentation would be enough to spare the headache with the returned connectable observer.

[DavidMGross]

Your wish is my command:

50b618e

On Fri, May 9, 2014 at 1:08 PM, andrask notifications@github.com wrote:

@DavidMGross https://github.com/DavidMGross Adding that single line to
the replay() documentation would be enough to spare the headache with the
returned connectable observer.

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/1180#issuecomment-42708058
.

David M. Gross
PLP Consulting

[andrask]

@DavidMGross You're awesome! Instead of whining, I should have done it. Fortunately, there is always next time in open source and I love it!