@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); }
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); } }
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); } } }
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); }
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); }
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);  }