@Ignore @Test public void testThreadStarvation() throws InterruptedException { TestSubscriber<String> to = new TestSubscriber<String>(new Observer<String>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String t) { try { Thread.sleep(1); } catch (InterruptedException e) { } } }); Observer<String> o = serializedObserver(to); AtomicInteger p1 = new AtomicInteger(); AtomicInteger p2 = new AtomicInteger(); Subscription s1 = infinite(p1).subscribe(o); Subscription s2 = infinite(p2).subscribe(o); Thread.sleep(100); System.out.println("p1: " + p1.get() + " p2: " + p2.get() + " => should be close to each other unless we have thread starvation"); assertEquals(p1.get(), p2.get(), 10000); s1.unsubscribe(); s2.unsubscribe(); }
public SerializedObserver(Observer<? super T> s) { this.actual = s; }
SerializedObserver implements Observer<T> { public SerializedObserver(Observer<? super T> s) { this.actual = s; } }
SerializedObserver implements Observer<T> { public SerializedObserver(Observer<? super T> s) { this.actual = s; } SerializedObserver(Observer<? super T> s); }
SerializedObserver implements Observer<T> { public SerializedObserver(Observer<? super T> s) { this.actual = s; } SerializedObserver(Observer<? super T> s); @Override void onCompleted(); @Override void onError(final Throwable e); @Override void onNext(T t); }
SerializedObserver implements Observer<T> { public SerializedObserver(Observer<? super T> s) { this.actual = s; } SerializedObserver(Observer<? super T> s); @Override void onCompleted(); @Override void onError(final Throwable e); @Override void onNext(T t);  }