@Test public void calculateSourceRequestedWithLargerValuesDoesNotOverflow() { emitted = Long.MAX_VALUE - 4; requestN = Long.MAX_VALUE; sourceRequested = Long.MAX_VALUE - 3; assertEquals(3, calculateSourceRequested(requestNUpdater, sourceRequestedUpdater, emittedUpdater, Integer.MAX_VALUE, this)); assertEquals(Long.MAX_VALUE - 4, emitted); assertEquals(Long.MAX_VALUE, requestN); assertEquals(Long.MAX_VALUE, sourceRequested); }
public static <T> int calculateSourceRequested(final AtomicLongFieldUpdater<T> requestNUpdater, final AtomicLongFieldUpdater<T> sourceRequestedUpdater, final AtomicLongFieldUpdater<T> emittedUpdater, final int limit, final T owner) { for (;;) { final long sourceRequested = sourceRequestedUpdater.get(owner); final long requested = requestNUpdater.get(owner); if (requested == sourceRequested) { return 0; } final long emitted = emittedUpdater.get(owner); if (emitted > sourceRequested) { final long delta = requested - emitted; final int toRequest = (int) min(limit, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, emitted + toRequest)) { return toRequest; } } else { final long outstanding = sourceRequested - emitted; final long delta = requested - sourceRequested; final int toRequest = (int) min(limit - outstanding, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, sourceRequested + toRequest)) { return toRequest; } } } }
SubscriberUtils { public static <T> int calculateSourceRequested(final AtomicLongFieldUpdater<T> requestNUpdater, final AtomicLongFieldUpdater<T> sourceRequestedUpdater, final AtomicLongFieldUpdater<T> emittedUpdater, final int limit, final T owner) { for (;;) { final long sourceRequested = sourceRequestedUpdater.get(owner); final long requested = requestNUpdater.get(owner); if (requested == sourceRequested) { return 0; } final long emitted = emittedUpdater.get(owner); if (emitted > sourceRequested) { final long delta = requested - emitted; final int toRequest = (int) min(limit, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, emitted + toRequest)) { return toRequest; } } else { final long outstanding = sourceRequested - emitted; final long delta = requested - sourceRequested; final int toRequest = (int) min(limit - outstanding, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, sourceRequested + toRequest)) { return toRequest; } } } } }
SubscriberUtils { public static <T> int calculateSourceRequested(final AtomicLongFieldUpdater<T> requestNUpdater, final AtomicLongFieldUpdater<T> sourceRequestedUpdater, final AtomicLongFieldUpdater<T> emittedUpdater, final int limit, final T owner) { for (;;) { final long sourceRequested = sourceRequestedUpdater.get(owner); final long requested = requestNUpdater.get(owner); if (requested == sourceRequested) { return 0; } final long emitted = emittedUpdater.get(owner); if (emitted > sourceRequested) { final long delta = requested - emitted; final int toRequest = (int) min(limit, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, emitted + toRequest)) { return toRequest; } } else { final long outstanding = sourceRequested - emitted; final long delta = requested - sourceRequested; final int toRequest = (int) min(limit - outstanding, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, sourceRequested + toRequest)) { return toRequest; } } } } private  SubscriberUtils(); }
SubscriberUtils { public static <T> int calculateSourceRequested(final AtomicLongFieldUpdater<T> requestNUpdater, final AtomicLongFieldUpdater<T> sourceRequestedUpdater, final AtomicLongFieldUpdater<T> emittedUpdater, final int limit, final T owner) { for (;;) { final long sourceRequested = sourceRequestedUpdater.get(owner); final long requested = requestNUpdater.get(owner); if (requested == sourceRequested) { return 0; } final long emitted = emittedUpdater.get(owner); if (emitted > sourceRequested) { final long delta = requested - emitted; final int toRequest = (int) min(limit, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, emitted + toRequest)) { return toRequest; } } else { final long outstanding = sourceRequested - emitted; final long delta = requested - sourceRequested; final int toRequest = (int) min(limit - outstanding, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, sourceRequested + toRequest)) { return toRequest; } } } } private  SubscriberUtils(); static boolean checkDuplicateSubscription(@Nullable Subscription existing, Subscription next); static boolean isRequestNValid(long n); static IllegalArgumentException newExceptionForInvalidRequestN(long n); static int calculateSourceRequested(final AtomicLongFieldUpdater<T> requestNUpdater,
                                                   final AtomicLongFieldUpdater<T> sourceRequestedUpdater,
                                                   final AtomicLongFieldUpdater<T> emittedUpdater,
                                                   final int limit,
                                                   final T owner); static boolean trySetTerminal(TerminalNotification toSet, boolean overrideComplete,
                     AtomicReferenceFieldUpdater<R, TerminalNotification> terminalNotificationUpdater, R flagOwner); static void deliverCompleteFromSource(Subscriber<T> subscriber); static void deliverSuccessFromSource(SingleSource.Subscriber<T> subscriber, @Nullable T value); static void deliverCompleteFromSource(CompletableSource.Subscriber subscriber); static void deliverErrorFromSource(Subscriber<T> subscriber, Throwable cause); static void deliverErrorFromSource(SingleSource.Subscriber<T> subscriber, Throwable cause); static void deliverErrorFromSource(CompletableSource.Subscriber subscriber, Throwable cause); static void handleExceptionFromOnSubscribe(Subscriber<T> subscriber, Throwable cause); static void handleExceptionFromOnSubscribe(SingleSource.Subscriber<T> subscriber, Throwable cause); static void handleExceptionFromOnSubscribe(CompletableSource.Subscriber subscriber, Throwable cause); static void safeOnError(CompletableSource.Subscriber subscriber, Throwable cause); static void safeOnError(SingleSource.Subscriber<T> subscriber, Throwable cause); static void safeOnError(PublisherSource.Subscriber<T> subscriber, Throwable cause); static void safeOnComplete(PublisherSource.Subscriber<T> subscriber); static void safeOnSuccess(SingleSource.Subscriber<T> subscriber, @Nullable T value); static void safeOnComplete(CompletableSource.Subscriber subscriber); }
SubscriberUtils { public static <T> int calculateSourceRequested(final AtomicLongFieldUpdater<T> requestNUpdater, final AtomicLongFieldUpdater<T> sourceRequestedUpdater, final AtomicLongFieldUpdater<T> emittedUpdater, final int limit, final T owner) { for (;;) { final long sourceRequested = sourceRequestedUpdater.get(owner); final long requested = requestNUpdater.get(owner); if (requested == sourceRequested) { return 0; } final long emitted = emittedUpdater.get(owner); if (emitted > sourceRequested) { final long delta = requested - emitted; final int toRequest = (int) min(limit, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, emitted + toRequest)) { return toRequest; } } else { final long outstanding = sourceRequested - emitted; final long delta = requested - sourceRequested; final int toRequest = (int) min(limit - outstanding, delta); if (sourceRequestedUpdater.compareAndSet(owner, sourceRequested, sourceRequested + toRequest)) { return toRequest; } } } } private  SubscriberUtils(); static boolean checkDuplicateSubscription(@Nullable Subscription existing, Subscription next); static boolean isRequestNValid(long n); static IllegalArgumentException newExceptionForInvalidRequestN(long n); static int calculateSourceRequested(final AtomicLongFieldUpdater<T> requestNUpdater,
                                                   final AtomicLongFieldUpdater<T> sourceRequestedUpdater,
                                                   final AtomicLongFieldUpdater<T> emittedUpdater,
                                                   final int limit,
                                                   final T owner); static boolean trySetTerminal(TerminalNotification toSet, boolean overrideComplete,
                     AtomicReferenceFieldUpdater<R, TerminalNotification> terminalNotificationUpdater, R flagOwner); static void deliverCompleteFromSource(Subscriber<T> subscriber); static void deliverSuccessFromSource(SingleSource.Subscriber<T> subscriber, @Nullable T value); static void deliverCompleteFromSource(CompletableSource.Subscriber subscriber); static void deliverErrorFromSource(Subscriber<T> subscriber, Throwable cause); static void deliverErrorFromSource(SingleSource.Subscriber<T> subscriber, Throwable cause); static void deliverErrorFromSource(CompletableSource.Subscriber subscriber, Throwable cause); static void handleExceptionFromOnSubscribe(Subscriber<T> subscriber, Throwable cause); static void handleExceptionFromOnSubscribe(SingleSource.Subscriber<T> subscriber, Throwable cause); static void handleExceptionFromOnSubscribe(CompletableSource.Subscriber subscriber, Throwable cause); static void safeOnError(CompletableSource.Subscriber subscriber, Throwable cause); static void safeOnError(SingleSource.Subscriber<T> subscriber, Throwable cause); static void safeOnError(PublisherSource.Subscriber<T> subscriber, Throwable cause); static void safeOnComplete(PublisherSource.Subscriber<T> subscriber); static void safeOnSuccess(SingleSource.Subscriber<T> subscriber, @Nullable T value); static void safeOnComplete(CompletableSource.Subscriber subscriber);  }