@Test public void autoDispose_withMaybe_normal() { TestSubscriber<Integer> firstSubscriber = new TestSubscriber<>(); TestSubscriber<Integer> secondSubscriber = new TestSubscriber<>(); PublishProcessor<Integer> source = PublishProcessor.create(); CompletableSubject scope = CompletableSubject.create(); Subscriber<Integer>[] subscribers = new Subscriber[] {firstSubscriber, secondSubscriber}; source.parallel(DEFAULT_PARALLELISM).to(autoDisposable(scope)).subscribe(subscribers); assertThat(firstSubscriber.hasSubscription()).isTrue(); assertThat(secondSubscriber.hasSubscription()).isTrue(); assertThat(source.hasSubscribers()).isTrue(); assertThat(scope.hasObservers()).isTrue(); source.onNext(1); source.onNext(2); firstSubscriber.assertValue(1); secondSubscriber.assertValue(2); source.onNext(3); source.onNext(4); source.onComplete(); firstSubscriber.assertValues(1, 3); firstSubscriber.assertComplete(); secondSubscriber.assertValues(2, 4); secondSubscriber.assertComplete(); assertThat(source.hasSubscribers()).isFalse(); assertThat(scope.hasObservers()).isFalse(); }
@Override public void subscribe(Subscriber<? super T>[] subscribers) { if (!validate(subscribers)) { return; } @SuppressWarnings("unchecked") Subscriber<? super T>[] newSubscribers = new Subscriber[subscribers.length]; for (int i = 0; i < subscribers.length; i++) { AutoDisposingSubscriberImpl<? super T> subscriber = new AutoDisposingSubscriberImpl<>(scope, subscribers[i]); newSubscribers[i] = subscriber; } source.subscribe(newSubscribers); }
AutoDisposeParallelFlowable extends ParallelFlowable<T> implements ParallelFlowableSubscribeProxy<T> { @Override public void subscribe(Subscriber<? super T>[] subscribers) { if (!validate(subscribers)) { return; } @SuppressWarnings("unchecked") Subscriber<? super T>[] newSubscribers = new Subscriber[subscribers.length]; for (int i = 0; i < subscribers.length; i++) { AutoDisposingSubscriberImpl<? super T> subscriber = new AutoDisposingSubscriberImpl<>(scope, subscribers[i]); newSubscribers[i] = subscriber; } source.subscribe(newSubscribers); } }
AutoDisposeParallelFlowable extends ParallelFlowable<T> implements ParallelFlowableSubscribeProxy<T> { @Override public void subscribe(Subscriber<? super T>[] subscribers) { if (!validate(subscribers)) { return; } @SuppressWarnings("unchecked") Subscriber<? super T>[] newSubscribers = new Subscriber[subscribers.length]; for (int i = 0; i < subscribers.length; i++) { AutoDisposingSubscriberImpl<? super T> subscriber = new AutoDisposingSubscriberImpl<>(scope, subscribers[i]); newSubscribers[i] = subscriber; } source.subscribe(newSubscribers); } AutoDisposeParallelFlowable(ParallelFlowable<T> source, CompletableSource scope); }
AutoDisposeParallelFlowable extends ParallelFlowable<T> implements ParallelFlowableSubscribeProxy<T> { @Override public void subscribe(Subscriber<? super T>[] subscribers) { if (!validate(subscribers)) { return; } @SuppressWarnings("unchecked") Subscriber<? super T>[] newSubscribers = new Subscriber[subscribers.length]; for (int i = 0; i < subscribers.length; i++) { AutoDisposingSubscriberImpl<? super T> subscriber = new AutoDisposingSubscriberImpl<>(scope, subscribers[i]); newSubscribers[i] = subscriber; } source.subscribe(newSubscribers); } AutoDisposeParallelFlowable(ParallelFlowable<T> source, CompletableSource scope); @Override void subscribe(Subscriber<? super T>[] subscribers); @Override int parallelism(); }
AutoDisposeParallelFlowable extends ParallelFlowable<T> implements ParallelFlowableSubscribeProxy<T> { @Override public void subscribe(Subscriber<? super T>[] subscribers) { if (!validate(subscribers)) { return; } @SuppressWarnings("unchecked") Subscriber<? super T>[] newSubscribers = new Subscriber[subscribers.length]; for (int i = 0; i < subscribers.length; i++) { AutoDisposingSubscriberImpl<? super T> subscriber = new AutoDisposingSubscriberImpl<>(scope, subscribers[i]); newSubscribers[i] = subscriber; } source.subscribe(newSubscribers); } AutoDisposeParallelFlowable(ParallelFlowable<T> source, CompletableSource scope); @Override void subscribe(Subscriber<? super T>[] subscribers); @Override int parallelism();  }