LastSeenObserver does not check downstream disposal

[kxfang]

While investigating https://github.com/JakeWharton/RxReplayingShare/issues/30, I noticed that LastSeenObserver does the following:

        @Override
        public void onSubscribe(Disposable d) {
            downstream.onSubscribe(d);

            T value = lastSeen.value;
            if (value != null) {
                downstream.onNext(value);
            }
        }

Is there a reason why we don't check d.isDisposed() before calling downstream.onNext(v)? A downstream operator might call d.dispose() in its onSubscribe, so in that case we probably shouldn't emit another value right?

[JakeWharton]

Just that it seems unlikely as this operator is meant to be used for long-lived streams and isn't a protocol violation.

[kxfang]

Yeah fair, it's not a common situation, I ran into it while I was debugging the other issue. Checking isDisposed() feels like the more correct thing to do though. I had something like this:

        replayingShareStream.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                d.dispose();
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println("Observer 1:" + integer);
            }

and it totally surprised me that onNext got called. Given that replay(1) doesn't behave this way and this operator is meant to be a combination of replay(1), publish(), and refCount() operators, it'd be nice to stay consistent in behavior so that we don't surprise people.

Let me know what you think -- happy to contribute a PR if you think it's a reasonable change.

[JakeWharton]

Sure. Sounds good.