SubjectSubscriptionManager should be public

[jjongsma]

Hiding SubjectSubscriptionManager (and making all Subject implementations final) makes it hard to create subjects with custom behavior. Please make this class public.

[akarnokd]

SubjectSubscriptionManager was not designed to be used outside the subjects package; changing it to support public usage would increase its overhead by a factor of 2.

What kind of Subject behavior do you need?

[benjchristensen]

Yes, SubjectSubscriptionManager is very much an internal implementation detail and non-trivial. It is not something we want to make public if for no other reason than the fact that it would then never be able to be changed.

Most new Subject implementations can be built on top of existing such as PublishSubject using composition instead of inheritance, so it would be good to know what use case can not be achieved through composition.

[jjongsma]

I have a use for a one-and-done subject, where each subscriber gets a
single item and then completes.

This single item is either the most recently received item (in the past),
or if no items have been received yet it waits for the next one. So it's
sort of like BehaviorSubject.subscribe(s).first(), except with no
initial/default item.

Can you think of a way to do this through composition? It would seem to
require access to the subscriber list, and I'm trying to avoid rewriting
what's already done inside SubjectSubscriptionManager.
On May 8, 2014 11:56 AM, "Ben Christensen" notifications@github.com wrote:

Yes, SubjectSubscriptionManager is very much an internal implementation
detail and non-trivial. It is not something we want to make public if for
no other reason than the fact that it would then never be able to be
changed.

Most new Subject implementations can be built on top of existing such as
PublishSubject using composition instead of inheritance, so it would be
good to know what use case can not be achieved through composition.

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

[DylanSale]

behaviourSubject.drop(1).first()?

[jjongsma]

I still want the first item, but only if it came from the the source
observable instead of BehaviorSubject's default value.
On May 8, 2014 5:49 PM, "Dylan Sale" notifications@github.com wrote:

behaviourSubject.drop(1).first()?

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

[DylanSale]

Make the default a sentinel then use a predicate with first to check it isn't that?

(On my phone so sorry for the formatting)

final Foo sentinel = new Foo();
BehaviourSubject b = BehaviourSubject.create(sentinel);
b.first(f -> f != sentinel)

[benjchristensen]

I've considered adding BehaviorSubject.create() without a default for this case as I've had to use the sentinel solution before and it's awkward.

[headinthebox]

Why not use asyncsubject?

[benjchristensen]

AsyncSubject is only the last value after completed, not last value and all subsequent.

[headinthebox]

have a use for a one-and-done subject, where each subscriber gets a
single item and then completes.

[benjchristensen]

Oh. Missed that part. Yup, that's AsyncSubject :-)

[jjongsma]

The difference is that I want the subscriber to complete after one item
even though the source observable is still active.
On May 8, 2014 6:55 PM, "Ben Christensen" notifications@github.com wrote:

Oh. Missed that part. Yup, that's AsyncSubject :-)

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

[benjchristensen]

Sounds like AsyncSubject.take(1).subscribe()

[benjchristensen]

Sorry, Publish or BehaviorSubject and take(1). Not AsyncSubject.

[akarnokd]

How about this:

public class TakeOne {
    static final class Client extends Subscriber<Integer> {
        final int id;
        public Client(int id) {
            this.id = id;
        }

        @Override
        public void onNext(Integer t) {
            System.out.printf("%d: %s%n", id, t);
        }

        @Override
        public void onError(Throwable e) {
            System.out.printf("%d: %s%n", id, e);
        }

        @Override
        public void onCompleted() {
            System.out.printf("%d: %s%n", id, "Done.");
        }

    }
    public static void main(String[] args) {
        PublishSubject<Integer> source = PublishSubject.create();

        ConnectableObservable<Integer> intermediate = source.replay(1);
        intermediate.connect();

        Observable<Integer> result = intermediate.replay(o -> o.take(1));

        Client c1 = new Client(1);
        result.subscribe(c1);

        source.onNext(1);
        source.onNext(2);

        Client c2 = new Client(2);
        result.subscribe(c2);

        source.onNext(3);

        Client c3 = new Client(3);
        result.subscribe(c3);

        Client c4 = new Client(4);
        result.subscribe(c4);

        source.onNext(4);

        Client c5 = new Client(5);
        result.subscribe(c5);
    }
}

prints:

1: 1
1: Done.
2: 2
2: Done.
3: 3
3: Done.
4: 3
4: Done.
5: 4
5: Done.

It may require PR #1175 to work correctly. Also, we seem to have found an use case for replay(Func1) :)

[jdreesen]

I've considered adding BehaviorSubject.create() without a default for this case as I've had to use the sentinel solution before and it's awkward.

+1 this would be great to have

[jjongsma]

@akarnokd That seems to work, thank you!

Though my unit tests for this implementation also pass when just using this:

intermediate.take(1).subscribe(c);

What is the reason for the replay(Func1) call?

[benjchristensen]

BehaviorSubject.create() is merged -> https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java#L75

[benjchristensen]

Anything else needed before closing this issue?

[jjongsma]

Looks like that should work.

On Tue, May 20, 2014 at 3:14 AM, Ben Christensen
notifications@github.comwrote:

Anything else needed before closing this issue?

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