code: /*
   * Copyright (c) 2016-present, RxJava Contributors.
   *
   * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
   * compliance with the License. You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software distributed under the License is
   * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
   * the License for the specific language governing permissions and limitations under the License.
   */
  
  package io.reactivex.rxjava3.internal.subscriptions;
  
  import org.reactivestreams.Subscriber;
  
  import io.reactivex.rxjava3.annotations.*;
  
  /**
   * A subscription that signals a single value eventually.
   * <p>
   * Note that the class leaks all methods of {@link java.util.concurrent.atomic.AtomicLong}.
   * Use {@link #complete(Object)} to signal the single value.
   * <p>
   * The this atomic integer stores a bit field:<br>
   * bit 0: indicates that there is a value available<br>
   * bit 1: indicates that there was a request made<br>
   * bit 2: indicates there was a cancellation, exclusively set<br>
   * bit 3: indicates in fusion mode but no value yet, exclusively set<br>
   * bit 4: indicates in fusion mode and value is available, exclusively set<br>
   * bit 5: indicates in fusion mode and value has been consumed, exclusively set<br>
   * Where exclusively set means any other bits are 0 when that bit is set.
   * @param <T> the value type
   */
  public class DeferredScalarSubscription<@NonNull T> extends BasicIntQueueSubscription<T> {
  
      private static final long serialVersionUID = -2151279923272604993L;
  
      /** The Subscriber to emit the value to. */
      protected final Subscriber<? super T> downstream;
  
      /** The value is stored here if there is no request yet or in fusion mode. */
      protected T value;
  
      /** Indicates this Subscription has no value and not requested yet. */
      static final int NO_REQUEST_NO_VALUE = 0;
      /** Indicates this Subscription has a value but not requested yet. */
      static final int NO_REQUEST_HAS_VALUE = 1;
      /** Indicates this Subscription has been requested but there is no value yet. */
      static final int HAS_REQUEST_NO_VALUE = 2;
      /** Indicates this Subscription has both request and value. */
      static final int HAS_REQUEST_HAS_VALUE = 3;
  
      /** Indicates the Subscription has been cancelled. */
      static final int CANCELLED = 4;
  
      /** Indicates this Subscription is in fusion mode and is currently empty. */
      static final int FUSED_EMPTY = 8;
      /** Indicates this Subscription is in fusion mode and has a value. */
      static final int FUSED_READY = 16;
      /** Indicates this Subscription is in fusion mode and its value has been consumed. */
      static final int FUSED_CONSUMED = 32;
  
      /**
       * Creates a DeferredScalarSubscription by wrapping the given Subscriber.
       * @param downstream the Subscriber to wrap, not null (not verified)
       */
      public DeferredScalarSubscription(Subscriber<? super T> downstream) {
          this.downstream = downstream;
      }
  
      @Override
      public final void request(long n) {
          if (SubscriptionHelper.validate(n)) {
              for (;;) {
                  int state = get();
                  // if the any bits 1-31 are set, we are either in fusion mode (FUSED_*)
                  // or request has been called (HAS_REQUEST_*)
                  if ((state & ~NO_REQUEST_HAS_VALUE) != 0) {
                      return;
                  }
                  if (state == NO_REQUEST_HAS_VALUE) {
                      if (compareAndSet(NO_REQUEST_HAS_VALUE, HAS_REQUEST_HAS_VALUE)) {
                          T v = value;
                          if (v != null) {
                              value = null;
                              Subscriber<? super T> a = downstream;
                              a.onNext(v);
                              if (get() != CANCELLED) {
                                  a.onComplete();
                              }
                          }
                      }
                      return;
                  }
                  if (compareAndSet(NO_REQUEST_NO_VALUE, HAS_REQUEST_NO_VALUE)) {
                      return;
                  }
              }
          }
      }
  
      /**
       * Completes this subscription by indicating the given value should
       * be emitted when the first request arrives.
       * <p>Make sure this is called exactly once.
       * @param v the value to signal, not null (not validated)
       */
      public final void complete(T v) {
          int state = get();
          for (;;) {
              if (state == FUSED_EMPTY) {
                  value = v;
                  lazySet(FUSED_READY);
  
                  Subscriber<? super T> a = downstream;
                  a.onNext(null);
                  if (get() != CANCELLED) {
                      a.onComplete();
                  }
                  return;
              }
  
              // if state is >= CANCELLED or bit zero is set (*_HAS_VALUE) case, return
              if ((state & ~HAS_REQUEST_NO_VALUE) != 0) {
                  return;
              }
  
              if (state == HAS_REQUEST_NO_VALUE) {
                  lazySet(HAS_REQUEST_HAS_VALUE);
                  Subscriber<? super T> a = downstream;
                  a.onNext(v);
                  if (get() != CANCELLED) {
                      a.onComplete();
                  }
                  return;
              }
              value = v;
              if (compareAndSet(NO_REQUEST_NO_VALUE, NO_REQUEST_HAS_VALUE)) {
                  return;
              }
              state = get();
              if (state == CANCELLED) {
                  value = null;
                  return;
              }
          }
      }
  
      @Override
      public final int requestFusion(int mode) {
          if ((mode & ASYNC) != 0) {
              lazySet(FUSED_EMPTY);
              return ASYNC;
          }
          return NONE;
      }
  
      @Nullable
      @Override
      public final T poll() {
          if (get() == FUSED_READY) {
              lazySet(FUSED_CONSUMED);
              T v = value;
              value = null;
              return v;
          }
          return null;
      }
  
      @Override
      public final boolean isEmpty() {
          return get() != FUSED_READY;
      }
  
      @Override
      public final void clear() {
          lazySet(FUSED_CONSUMED);
          value = null;
      }
  
      @Override
      public void cancel() {
          set(CANCELLED);
          value = null;
      }
  
      /**
       * Returns true if this Subscription has been cancelled.
       * @return true if this Subscription has been cancelled
       */
      public final boolean isCancelled() {
          return get() == CANCELLED;
      }
  
      /**
       * Atomically sets a cancelled state and returns true if
       * the current thread did it successfully.
       * @return true if the current thread cancelled
       */
      public final boolean tryCancel() {
          return getAndSet(CANCELLED) != CANCELLED;
      }
  }
  