@Test public void testSetIfEmptyAndRequest() { AtomicReference<Subscription> container = new AtomicReference<>(); AtomicLong requests = new AtomicLong(); Subscription sub1 = mock(Subscription.class); Subscription sub2 = mock(Subscription.class); assertThat(Subscriptions.setIfEmptyAndRequest(container, requests, sub1)).isTrue(); verify(sub1, never()).cancel(); verify(sub1, never()).request(anyLong()); requests.set(2); assertThat(Subscriptions.setIfEmptyAndRequest(container, requests, sub2)).isFalse(); verify(sub1, never()).cancel(); verify(sub2, times(1)).cancel(); verify(sub1, never()).request(anyLong()); assertThat(requests).hasValue(2); container.set(null); requests.set(30); assertThat(Subscriptions.setIfEmptyAndRequest(container, requests, sub1)).isTrue(); verify(sub1, never()).cancel(); verify(sub1, times(1)).request(30); assertThat(requests).hasValue(0); }
public static boolean setIfEmptyAndRequest(AtomicReference<Subscription> container, AtomicLong requested, Subscription subscription) { if (Subscriptions.setIfEmpty(container, subscription)) { long r = requested.getAndSet(0L); if (r > 0L) { subscription.request(r); } else if (r < 0) { throw new IllegalArgumentException("Invalid amount of request"); } return true; } return false; }
Subscriptions { public static boolean setIfEmptyAndRequest(AtomicReference<Subscription> container, AtomicLong requested, Subscription subscription) { if (Subscriptions.setIfEmpty(container, subscription)) { long r = requested.getAndSet(0L); if (r > 0L) { subscription.request(r); } else if (r < 0) { throw new IllegalArgumentException("Invalid amount of request"); } return true; } return false; } }
Subscriptions { public static boolean setIfEmptyAndRequest(AtomicReference<Subscription> container, AtomicLong requested, Subscription subscription) { if (Subscriptions.setIfEmpty(container, subscription)) { long r = requested.getAndSet(0L); if (r > 0L) { subscription.request(r); } else if (r < 0) { throw new IllegalArgumentException("Invalid amount of request"); } return true; } return false; } private  Subscriptions(); }
Subscriptions { public static boolean setIfEmptyAndRequest(AtomicReference<Subscription> container, AtomicLong requested, Subscription subscription) { if (Subscriptions.setIfEmpty(container, subscription)) { long r = requested.getAndSet(0L); if (r > 0L) { subscription.request(r); } else if (r < 0) { throw new IllegalArgumentException("Invalid amount of request"); } return true; } return false; } private  Subscriptions(); static IllegalArgumentException getInvalidRequestException(); static Subscription empty(); static void complete(Subscriber<?> subscriber); static void fail(Subscriber<?> subscriber, Throwable failure); static void fail(Subscriber<?> subscriber, Throwable failure, Publisher<?> upstream); static long add(long a, long b); static long add(AtomicLong requested, long requests); static long subtract(AtomicLong requested, long emitted); static int unboundedOrLimit(int prefetch); static long unboundedOrRequests(int concurrency); static boolean addFailure(AtomicReference<Throwable> failures, Throwable failure); static void cancel(AtomicReference<Subscription> reference); static Throwable markFailureAsTerminated(AtomicReference<Throwable> failures); static void terminateAndPropagate(AtomicReference<Throwable> failures, Subscriber<?> subscriber); static long multiply(long n, long times); static void requestIfNotNullOrAccumulate(AtomicReference<Subscription> field, AtomicLong requested, long requests); static boolean setIfEmptyAndRequest(AtomicReference<Subscription> container, AtomicLong requested,
            Subscription subscription); static boolean setIfEmpty(AtomicReference<Subscription> container, Subscription subscription); static Throwable terminate(AtomicReference<Throwable> failure); static long produced(AtomicLong requested, long amount); static long subOrZero(long a, long b); static SingleItemSubscription<T> single(Subscriber<T> downstream, T item); static long producedAndHandleAlreadyCancelled(AtomicLong requested, long n); static long addAndHandledAlreadyCancelled(AtomicLong requested, long n); }
Subscriptions { public static boolean setIfEmptyAndRequest(AtomicReference<Subscription> container, AtomicLong requested, Subscription subscription) { if (Subscriptions.setIfEmpty(container, subscription)) { long r = requested.getAndSet(0L); if (r > 0L) { subscription.request(r); } else if (r < 0) { throw new IllegalArgumentException("Invalid amount of request"); } return true; } return false; } private  Subscriptions(); static IllegalArgumentException getInvalidRequestException(); static Subscription empty(); static void complete(Subscriber<?> subscriber); static void fail(Subscriber<?> subscriber, Throwable failure); static void fail(Subscriber<?> subscriber, Throwable failure, Publisher<?> upstream); static long add(long a, long b); static long add(AtomicLong requested, long requests); static long subtract(AtomicLong requested, long emitted); static int unboundedOrLimit(int prefetch); static long unboundedOrRequests(int concurrency); static boolean addFailure(AtomicReference<Throwable> failures, Throwable failure); static void cancel(AtomicReference<Subscription> reference); static Throwable markFailureAsTerminated(AtomicReference<Throwable> failures); static void terminateAndPropagate(AtomicReference<Throwable> failures, Subscriber<?> subscriber); static long multiply(long n, long times); static void requestIfNotNullOrAccumulate(AtomicReference<Subscription> field, AtomicLong requested, long requests); static boolean setIfEmptyAndRequest(AtomicReference<Subscription> container, AtomicLong requested,
            Subscription subscription); static boolean setIfEmpty(AtomicReference<Subscription> container, Subscription subscription); static Throwable terminate(AtomicReference<Throwable> failure); static long produced(AtomicLong requested, long amount); static long subOrZero(long a, long b); static SingleItemSubscription<T> single(Subscriber<T> downstream, T item); static long producedAndHandleAlreadyCancelled(AtomicLong requested, long n); static long addAndHandledAlreadyCancelled(AtomicLong requested, long n); static final Throwable TERMINATED; static final EmptySubscription CANCELLED; }