@Test public void ifParallelism_and_subscribersCount_dontMatch_shouldFail() { TestSubscriber<Integer> subscriber = new TestSubscriber<>(); CompletableSubject scope = CompletableSubject.create(); Subscriber<Integer>[] subscribers = new Subscriber[] {subscriber}; Flowable.just(1, 2) .parallel(DEFAULT_PARALLELISM) .to(autoDisposable(scope)) .subscribe(subscribers); subscriber.assertError(IllegalArgumentException.class); }
@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();  }