Subscriber implements Observer<T>, Subscription { protected final void request(long n) { if (n < 0) { throw new IllegalArgumentException("number requested cannot be negative: " + n); } Producer shouldRequest = null; synchronized (this) { if (p != null) { shouldRequest = p; } else if (requested == Long.MIN_VALUE) { requested = n; } else { final long total = requested + n; if (total < 0) { requested = Long.MAX_VALUE; } else { requested = total; } } } if (shouldRequest != null) { shouldRequest.request(n); } } protected  Subscriber(); protected  Subscriber(Subscriber<?> op); protected  Subscriber(Subscriber<?> op, boolean shareSubscriptions); final void add(Subscription s); @Override final void unsubscribe(); @Override final boolean isUnsubscribed(); void onStart(); void setProducer(Producer producer);  }
@Test public void testOnStartCalledOnceViaSubscribe() { final AtomicInteger c = new AtomicInteger(); Observable.just(1, 2, 3, 4).take(2).subscribe(new Subscriber<Integer>() { @Override public void onStart() { c.incrementAndGet(); request(1); } @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Integer t) { request(1); } }); assertEquals(1, c.get()); }
@Test public void testOnStartCalledOnceViaUnsafeSubscribe() { final AtomicInteger c = new AtomicInteger(); Observable.just(1, 2, 3, 4).take(2).unsafeSubscribe(new Subscriber<Integer>() { @Override public void onStart() { c.incrementAndGet(); request(1); } @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Integer t) { request(1); } }); assertEquals(1, c.get()); }
@Test public void testOnStartCalledOnceViaLift() { final AtomicInteger c = new AtomicInteger(); Observable.just(1, 2, 3, 4).lift(new Operator<Integer, Integer>() { @Override public Subscriber<? super Integer> call(final Subscriber<? super Integer> child) { return new Subscriber<Integer>() { @Override public void onStart() { c.incrementAndGet(); request(1); } @Override public void onCompleted() { child.onCompleted(); } @Override public void onError(Throwable e) { child.onError(e); } @Override public void onNext(Integer t) { child.onNext(t); request(1); } }; } }).subscribe(); assertEquals(1, c.get()); }
@Test public void testNegativeRequestThrowsIllegalArgumentException() throws InterruptedException { final CountDownLatch latch = new CountDownLatch(1); final AtomicReference<Throwable> exception = new AtomicReference<Throwable>(); Observable.just(1,2,3,4).subscribe(new Subscriber<Integer>() { @Override public void onStart() { request(1); } @Override public void onCompleted() { } @Override public void onError(Throwable e) { exception.set(e); latch.countDown(); } @Override public void onNext(Integer t) { request(-1); request(1); }}); assertTrue(latch.await(10, TimeUnit.SECONDS)); assertTrue(exception.get() instanceof IllegalArgumentException); }