ReplaySubject emits items received after onError

[samuelgruetter]

This snippet

    val sub = ReplaySubject[Int]()

    sub.onNext(1)
    sub.onNext(2)
    sub.onError(new Exception("Kabloooy!")) // Gets moved to after onNext(3)
    sub.onNext(3)

    sub.subscribe(
      n => println("sub1: " + n),
      e => println("sub1: " + e.getMessage),
      () => println("sub1: completed")
    )

    println("done")

outputs

sub1: 1
sub1: 2
sub1: 3
sub1: Kabloooy!
done

but I would expect

sub1: 1
sub1: 2
sub1: Kabloooy!
done

since items received after onError should not be emitted by ReplaySubject.

(Thank you Dragisa Krsmanovic for pointing this out)

[headinthebox]

Ben and I hacked on subjects earlier this week, all subjects are/were in pretty bad shape, but we'll fix it.

[benjchristensen]

ReplaySubject is the one I didn’t touch, I refactored Publish/Behavior/Async. I’ll put Replay on my TODO.

[akarnokd]

I think this is a trivial fix

    @Override
    public void onNext(T args)
    {
        synchronized (subscriptions) {
            if (isDone) {
                return;
            }
            history.add(args);
            for (Observer<? super T> observer : new ArrayList<Observer<? super T>>(subscriptions.values())) {
                observer.onNext(args);
            }
        }
    }

[headinthebox]

We should also make sure no additional,on errors,are propagated.

Sent from my iPad

On Nov 30, 2013, at 4:52 AM, akarnokd [email protected] wrote:

I think this is a trivial fix

@Override
public void onNext(T args)
{
    synchronized (subscriptions) {
        if (isDone) {
            return;
        }
        history.add(args);
        for (Observer<? super T> observer : new ArrayList<Observer<? super T>>(subscriptions.values())) {
            observer.onNext(args);
        }
    }
}

—
Reply to this email directly or view it on GitHub.