content: /**
   * 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.flowable.processors;
  
  import java.lang.reflect.Array;
  import java.util.concurrent.atomic.*;
  import java.util.concurrent.locks.*;
  
  import org.reactivestreams.*;
  
  import io.reactivex.common.RxJavaCommonPlugins;
  import io.reactivex.common.annotations.*;
  import io.reactivex.common.exceptions.MissingBackpressureException;
  import io.reactivex.common.internal.functions.ObjectHelper;
  import io.reactivex.common.internal.utils.AbstractAppendOnlyLinkedArrayList.NonThrowingPredicate;
  import io.reactivex.common.internal.utils.ExceptionHelper;
  import io.reactivex.flowable.internal.subscriptions.SubscriptionHelper;
  import io.reactivex.flowable.internal.utils.*;
  
  /**
   * Processor that emits the most recent item it has observed and all subsequent observed items to each subscribed
   * {@link Subscriber}.
   * <p>
   * <img width="640" height="460" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/S.BehaviorProcessor.png" alt="">
   * <p>
   * Example usage:
   * <pre> {@code
  
    // observer will receive all events.
    BehaviorProcessor<Object> processor = BehaviorProcessor.create("default");
    processor.subscribe(observer);
    processor.onNext("one");
    processor.onNext("two");
    processor.onNext("three");
  
    // observer will receive the "one", "two" and "three" events, but not "zero"
    BehaviorProcessor<Object> processor = BehaviorProcessor.create("default");
    processor.onNext("zero");
    processor.onNext("one");
    processor.subscribe(observer);
    processor.onNext("two");
    processor.onNext("three");
  
    // observer will receive only onComplete
    BehaviorProcessor<Object> processor = BehaviorProcessor.create("default");
    processor.onNext("zero");
    processor.onNext("one");
    processor.onComplete();
    processor.subscribe(observer);
  
    // observer will receive only onError
    BehaviorProcessor<Object> processor = BehaviorProcessor.create("default");
    processor.onNext("zero");
    processor.onNext("one");
    processor.onError(new RuntimeException("error"));
    processor.subscribe(observer);
    } </pre>
   *
   * @param <T>
   *          the type of item expected to be observed and emitted by the Processor
   */
  public final class BehaviorProcessor<T> extends FlowableProcessor<T> {
      final AtomicReference<BehaviorSubscription<T>[]> subscribers;
  
      static final Object[] EMPTY_ARRAY = new Object[0];
  
      @SuppressWarnings("rawtypes")
      static final BehaviorSubscription[] EMPTY = new BehaviorSubscription[0];
  
      @SuppressWarnings("rawtypes")
      static final BehaviorSubscription[] TERMINATED = new BehaviorSubscription[0];
  
      final ReadWriteLock lock;
      final Lock readLock;
      final Lock writeLock;
  
      final AtomicReference<Object> value;
  
      final AtomicReference<Throwable> terminalEvent;
  
      long index;
  
      /**
       * Creates a {@link BehaviorProcessor} without a default item.
       *
       * @param <T>
       *            the type of item the Subject will emit
       * @return the constructed {@link BehaviorProcessor}
       */
      @CheckReturnValue
      public static <T> BehaviorProcessor<T> create() {
          return new BehaviorProcessor<T>();
      }
  
      /**
       * Creates a {@link BehaviorProcessor} that emits the last item it observed and all subsequent items to each
       * {@link Subscriber} that subscribes to it.
       *
       * @param <T>
       *            the type of item the Subject will emit
       * @param defaultValue
       *            the item that will be emitted first to any {@link Subscriber} as long as the
       *            {@link BehaviorProcessor} has not yet observed any items from its source {@code Observable}
       * @return the constructed {@link BehaviorProcessor}
       */
      @CheckReturnValue
      public static <T> BehaviorProcessor<T> createDefault(T defaultValue) {
          ObjectHelper.requireNonNull(defaultValue, "defaultValue is null");
          return new BehaviorProcessor<T>(defaultValue);
      }
  
      /**
       * Constructs an empty BehaviorProcessor.
       * @since 2.0
       */
      @SuppressWarnings("unchecked")
      BehaviorProcessor() {
          this.value = new AtomicReference<Object>();
          this.lock = new ReentrantReadWriteLock();
          this.readLock = lock.readLock();
          this.writeLock = lock.writeLock();
          this.subscribers = new AtomicReference<BehaviorSubscription<T>[]>(EMPTY);
          this.terminalEvent = new AtomicReference<Throwable>();
      }
  
      /**
       * Constructs a BehaviorProcessor with the given initial value.
       * @param defaultValue the initial value, not null (verified)
       * @throws NullPointerException if {@code defaultValue} is null
       * @since 2.0
       */
      BehaviorProcessor(T defaultValue) {
          this();
          this.value.lazySet(ObjectHelper.requireNonNull(defaultValue, "defaultValue is null"));
      }
  
      @Override
      protected void subscribeActual(Subscriber<? super T> s) {
          BehaviorSubscription<T> bs = new BehaviorSubscription<T>(s, this);
          s.onSubscribe(bs);
          if (add(bs)) {
              if (bs.cancelled) {
                  remove(bs);
              } else {
                  bs.emitFirst();
              }
          } else {
              Throwable ex = terminalEvent.get();
              if (ex == ExceptionHelper.TERMINATED) {
                  s.onComplete();
              } else {
                  s.onError(ex);
              }
          }
      }
  
      @Override
      public void onSubscribe(Subscription s) {
          if (terminalEvent.get() != null) {
              s.cancel();
              return;
          }
          s.request(Long.MAX_VALUE);
      }
  
      @Override
      public void onNext(T t) {
          if (t == null) {
              onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
              return;
          }
          if (terminalEvent.get() != null) {
              return;
          }
          Object o = NotificationLite.next(t);
          setCurrent(o);
          for (BehaviorSubscription<T> bs : subscribers.get()) {
              bs.emitNext(o, index);
          }
      }
  
      @Override
      public void onError(Throwable t) {
          if (t == null) {
              t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
          }
          if (!terminalEvent.compareAndSet(null, t)) {
              RxJavaCommonPlugins.onError(t);
              return;
          }
          Object o = NotificationLite.error(t);
          for (BehaviorSubscription<T> bs : terminate(o)) {
              bs.emitNext(o, index);
          }
      }
  
      @Override
      public void onComplete() {
          if (!terminalEvent.compareAndSet(null, ExceptionHelper.TERMINATED)) {
              return;
          }
          Object o = NotificationLite.complete();
          for (BehaviorSubscription<T> bs : terminate(o)) {
              bs.emitNext(o, index);  // relaxed read okay since this is the only mutator thread
          }
      }
  
      /**
       * Tries to emit the item to all currently subscribed Subscribers if all of them
       * has requested some value, returns false otherwise.
       * <p>
       * This method should be called in a sequential manner just like the onXXX methods
       * of the PublishProcessor.
       * <p>
       * Calling with null will terminate the PublishProcessor and a NullPointerException
       * is signalled to the Subscribers.
       * <p>History: 2.0.8 - experimental
       * @param t the item to emit, not null
       * @return true if the item was emitted to all Subscribers
       * @since 2.1
       */
      public boolean offer(T t) {
          if (t == null) {
              onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
              return true;
          }
          BehaviorSubscription<T>[] array = subscribers.get();
  
          for (BehaviorSubscription<T> s : array) {
              if (s.isFull()) {
                  return false;
              }
          }
  
          Object o = NotificationLite.next(t);
          setCurrent(o);
          for (BehaviorSubscription<T> bs : array) {
              bs.emitNext(o, index);
          }
          return true;
      }
  
      @Override
      public boolean hasSubscribers() {
          return subscribers.get().length != 0;
      }
  
  
      /* test support*/ int subscriberCount() {
          return subscribers.get().length;
      }
  
      @Override
      public Throwable getThrowable() {
          Object o = value.get();
          if (NotificationLite.isError(o)) {
              return NotificationLite.getError(o);
          }
          return null;
      }
  
      /**
       * Returns a single value the Subject currently has or null if no such value exists.
       * <p>The method is thread-safe.
       * @return a single value the Subject currently has or null if no such value exists
       */
      public T getValue() {
          Object o = value.get();
          if (NotificationLite.isComplete(o) || NotificationLite.isError(o)) {
              return null;
          }
          return NotificationLite.getValue(o);
      }
  
      /**
       * Returns an Object array containing snapshot all values of the Subject.
       * <p>The method is thread-safe.
       * @return the array containing the snapshot of all values of the Subject
       */
      public Object[] getValues() {
          @SuppressWarnings("unchecked")
          T[] a = (T[])EMPTY_ARRAY;
          T[] b = getValues(a);
          if (b == EMPTY_ARRAY) {
              return new Object[0];
          }
          return b;
  
      }
  
      /**
       * Returns a typed array containing a snapshot of all values of the Subject.
       * <p>The method follows the conventions of Collection.toArray by setting the array element
       * after the last value to null (if the capacity permits).
       * <p>The method is thread-safe.
       * @param array the target array to copy values into if it fits
       * @return the given array if the values fit into it or a new array containing all values
       */
      @SuppressWarnings("unchecked")
      public T[] getValues(T[] array) {
          Object o = value.get();
          if (o == null || NotificationLite.isComplete(o) || NotificationLite.isError(o)) {
              if (array.length != 0) {
                  array[0] = null;
              }
              return array;
          }
          T v = NotificationLite.getValue(o);
          if (array.length != 0) {
              array[0] = v;
              if (array.length != 1) {
                  array[1] = null;
              }
          } else {
              array = (T[])Array.newInstance(array.getClass().getComponentType(), 1);
              array[0] = v;
          }
          return array;
      }
  
      @Override
      public boolean hasComplete() {
          Object o = value.get();
          return NotificationLite.isComplete(o);
      }
  
      @Override
      public boolean hasThrowable() {
          Object o = value.get();
          return NotificationLite.isError(o);
      }
  
      /**
       * Returns true if the subject has any value.
       * <p>The method is thread-safe.
       * @return true if the subject has any value
       */
      public boolean hasValue() {
          Object o = value.get();
          return o != null && !NotificationLite.isComplete(o) && !NotificationLite.isError(o);
      }
  
  
      boolean add(BehaviorSubscription<T> rs) {
          for (;;) {
              BehaviorSubscription<T>[] a = subscribers.get();
              if (a == TERMINATED) {
                  return false;
              }
              int len = a.length;
              @SuppressWarnings("unchecked")
              BehaviorSubscription<T>[] b = new BehaviorSubscription[len + 1];
              System.arraycopy(a, 0, b, 0, len);
              b[len] = rs;
              if (subscribers.compareAndSet(a, b)) {
                  return true;
              }
          }
      }
  
      @SuppressWarnings("unchecked")
      void remove(BehaviorSubscription<T> rs) {
          for (;;) {
              BehaviorSubscription<T>[] a = subscribers.get();
              if (a == TERMINATED || a == EMPTY) {
                  return;
              }
              int len = a.length;
              int j = -1;
              for (int i = 0; i < len; i++) {
                  if (a[i] == rs) {
                      j = i;
                      break;
                  }
              }
  
              if (j < 0) {
                  return;
              }
              BehaviorSubscription<T>[] b;
              if (len == 1) {
                  b = EMPTY;
              } else {
                  b = new BehaviorSubscription[len - 1];
                  System.arraycopy(a, 0, b, 0, j);
                  System.arraycopy(a, j + 1, b, j, len - j - 1);
              }
              if (subscribers.compareAndSet(a, b)) {
                  return;
              }
          }
      }
  
      @SuppressWarnings("unchecked")
      BehaviorSubscription<T>[] terminate(Object terminalValue) {
  
          BehaviorSubscription<T>[] a = subscribers.get();
          if (a != TERMINATED) {
              a = subscribers.getAndSet(TERMINATED);
              if (a != TERMINATED) {
                  // either this or atomics with lots of allocation
                  setCurrent(terminalValue);
              }
          }
  
          return a;
      }
  
      void setCurrent(Object o) {
          Lock wl = writeLock;
          wl.lock();
          index++;
          value.lazySet(o);
          wl.unlock();
      }
  
      static final class BehaviorSubscription<T> extends AtomicLong implements Subscription, NonThrowingPredicate<Object> {
  
          private static final long serialVersionUID = 3293175281126227086L;
  
          final Subscriber<? super T> actual;
          final BehaviorProcessor<T> state;
  
          boolean next;
          boolean emitting;
          AppendOnlyLinkedArrayList<Object> queue;
  
          boolean fastPath;
  
          volatile boolean cancelled;
  
          long index;
  
          BehaviorSubscription(Subscriber<? super T> actual, BehaviorProcessor<T> state) {
              this.actual = actual;
              this.state = state;
          }
  
          @Override
          public void request(long n) {
              if (SubscriptionHelper.validate(n)) {
                  BackpressureHelper.add(this, n);
              }
          }
  
          @Override
          public void cancel() {
              if (!cancelled) {
                  cancelled = true;
  
                  state.remove(this);
              }
          }
  
          void emitFirst() {
              if (cancelled) {
                  return;
              }
              Object o;
              synchronized (this) {
                  if (cancelled) {
                      return;
                  }
                  if (next) {
                      return;
                  }
  
                  BehaviorProcessor<T> s = state;
  
                  Lock readLock = s.readLock;
                  readLock.lock();
                  index = s.index;
                  o = s.value.get();
                  readLock.unlock();
  
                  emitting = o != null;
                  next = true;
              }
  
              if (o != null) {
                  if (test(o)) {
                      return;
                  }
  
                  emitLoop();
              }
          }
  
          void emitNext(Object value, long stateIndex) {
              if (cancelled) {
                  return;
              }
              if (!fastPath) {
                  synchronized (this) {
                      if (cancelled) {
                          return;
                      }
                      if (index == stateIndex) {
                          return;
                      }
                      if (emitting) {
                          AppendOnlyLinkedArrayList<Object> q = queue;
                          if (q == null) {
                              q = new AppendOnlyLinkedArrayList<Object>(4);
                              queue = q;
                          }
                          q.add(value);
                          return;
                      }
                      next = true;
                  }
                  fastPath = true;
              }
  
              test(value);
          }
  
          @Override
          public boolean test(Object o) {
              if (cancelled) {
                  return true;
              }
  
              if (NotificationLite.isComplete(o)) {
                  actual.onComplete();
                  return true;
              } else
              if (NotificationLite.isError(o)) {
                  actual.onError(NotificationLite.getError(o));
                  return true;
              }
  
              long r = get();
              if (r != 0L) {
                  actual.onNext(NotificationLite.<T>getValue(o));
                  if (r != Long.MAX_VALUE) {
                      decrementAndGet();
                  }
                  return false;
              }
              cancel();
              actual.onError(new MissingBackpressureException("Could not deliver value due to lack of requests"));
              return true;
          }
  
          void emitLoop() {
              for (;;) {
                  if (cancelled) {
                      return;
                  }
                  AppendOnlyLinkedArrayList<Object> q;
                  synchronized (this) {
                      q = queue;
                      if (q == null) {
                          emitting = false;
                          return;
                      }
                      queue = null;
                  }
  
                  q.forEachWhile(this);
              }
          }
  
          public boolean isFull() {
              return get() == 0L;
          }
      }
  }
  