@Test public void testSetIfEmpty() { AtomicReference<Subscription> container = new AtomicReference<>(); Subscription sub1 = mock(Subscription.class); Subscription sub2 = mock(Subscription.class); assertThat(Subscriptions.setIfEmpty(container, sub1)).isTrue(); verify(sub1, never()).cancel(); assertThat(Subscriptions.setIfEmpty(container, sub2)).isFalse(); verify(sub1, never()).cancel(); verify(sub2, times(1)).cancel(); }
public static boolean setIfEmpty(AtomicReference<Subscription> container, Subscription subscription) { Objects.requireNonNull(subscription, "subscription is null"); if (!container.compareAndSet(null, subscription)) { subscription.cancel(); return false; } return true; }
Subscriptions { public static boolean setIfEmpty(AtomicReference<Subscription> container, Subscription subscription) { Objects.requireNonNull(subscription, "subscription is null"); if (!container.compareAndSet(null, subscription)) { subscription.cancel(); return false; } return true; } }
Subscriptions { public static boolean setIfEmpty(AtomicReference<Subscription> container, Subscription subscription) { Objects.requireNonNull(subscription, "subscription is null"); if (!container.compareAndSet(null, subscription)) { subscription.cancel(); return false; } return true; } private  Subscriptions(); }
Subscriptions { public static boolean setIfEmpty(AtomicReference<Subscription> container, Subscription subscription) { Objects.requireNonNull(subscription, "subscription is null"); if (!container.compareAndSet(null, subscription)) { subscription.cancel(); return false; } return true; } 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 setIfEmpty(AtomicReference<Subscription> container, Subscription subscription) { Objects.requireNonNull(subscription, "subscription is null"); if (!container.compareAndSet(null, subscription)) { subscription.cancel(); return false; } return true; } 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; }