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