OperationMulticast { public static <T, R> ConnectableObservable<R> multicast(Observable<? extends T> source, final Subject<? super T, ? extends R> subject) { return new MulticastConnectableObservable<T, R>(source, subject); }  static ConnectableObservable<R> multicast(Observable<? extends T> source, final Subject<? super T, ? extends R> subject); static Observable<TResult> multicast(
            final Observable<? extends TInput> source,
            final Func0<? extends Subject<? super TInput, ? extends TIntermediate>> subjectFactory,
            final Func1<? super Observable<TIntermediate>, ? extends Observable<TResult>> selector);  }
@Test public void testMulticast() { Subject<String, String> source = PublishSubject.create(); ConnectableObservable<String> multicasted = OperationMulticast.multicast(source, PublishSubject.<String> create()); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); multicasted.subscribe(observer); source.onNext("one"); source.onNext("two"); multicasted.connect(); source.onNext("three"); source.onNext("four"); source.onCompleted(); verify(observer, never()).onNext("one"); verify(observer, never()).onNext("two"); verify(observer, times(1)).onNext("three"); verify(observer, times(1)).onNext("four"); verify(observer, times(1)).onCompleted(); }
@Test public void testMulticastConnectTwice() { Subject<String, String> source = PublishSubject.create(); ConnectableObservable<String> multicasted = OperationMulticast.multicast(source, PublishSubject.<String> create()); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); multicasted.subscribe(observer); source.onNext("one"); multicasted.connect(); multicasted.connect(); source.onNext("two"); source.onCompleted(); verify(observer, never()).onNext("one"); verify(observer, times(1)).onNext("two"); verify(observer, times(1)).onCompleted(); }
@Test public void testMulticastDisconnect() { Subject<String, String> source = PublishSubject.create(); ConnectableObservable<String> multicasted = OperationMulticast.multicast(source, PublishSubject.<String> create()); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); multicasted.subscribe(observer); source.onNext("one"); Subscription connection = multicasted.connect(); source.onNext("two"); connection.unsubscribe(); source.onNext("three"); multicasted.connect(); source.onNext("four"); source.onCompleted(); verify(observer, never()).onNext("one"); verify(observer, times(1)).onNext("two"); verify(observer, never()).onNext("three"); verify(observer, times(1)).onNext("four"); verify(observer, times(1)).onCompleted(); }