@Test public void testMultiThreadedBasic() { TestMultiThreadedObservable onSubscribe = new TestMultiThreadedObservable("one", "two", "three"); Flowable<String> w = Flowable.unsafeCreate(onSubscribe); BusySubscriber busySubscriber = new BusySubscriber(); Subscriber<String> aw = serializedSubscriber(busySubscriber); w.subscribe(aw); onSubscribe.waitToFinish(); assertEquals(3, busySubscriber.onNextCount.get()); assertFalse(busySubscriber.onError); assertTrue(busySubscriber.onComplete); assertTrue(onSubscribe.maxConcurrentThreads.get() > 1); assertEquals(1, busySubscriber.maxConcurrentThreads.get()); }
public SerializedSubscriber(Subscriber<? super T> actual) { this(actual, false); }
SerializedSubscriber implements FlowableSubscriber<T>, Subscription { public SerializedSubscriber(Subscriber<? super T> actual) { this(actual, false); } }
SerializedSubscriber implements FlowableSubscriber<T>, Subscription { public SerializedSubscriber(Subscriber<? super T> actual) { this(actual, false); } SerializedSubscriber(Subscriber<? super T> actual);  SerializedSubscriber(Subscriber<? super T> actual, boolean delayError); }
SerializedSubscriber implements FlowableSubscriber<T>, Subscription { public SerializedSubscriber(Subscriber<? super T> actual) { this(actual, false); } SerializedSubscriber(Subscriber<? super T> actual);  SerializedSubscriber(Subscriber<? super T> actual, boolean delayError); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void request(long n); @Override void cancel(); }
SerializedSubscriber implements FlowableSubscriber<T>, Subscription { public SerializedSubscriber(Subscriber<? super T> actual) { this(actual, false); } SerializedSubscriber(Subscriber<? super T> actual);  SerializedSubscriber(Subscriber<? super T> actual, boolean delayError); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); @Override void request(long n); @Override void cancel();  }