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.operators.flowable;
  
  import java.util.concurrent.atomic.*;
  
  import org.reactivestreams.*;
  
  import io.reactivex.rxjava3.core.FlowableSubscriber;
  import io.reactivex.rxjava3.disposables.Disposable;
  import io.reactivex.rxjava3.exceptions.*;
  import io.reactivex.rxjava3.flowables.ConnectableFlowable;
  import io.reactivex.rxjava3.functions.Consumer;
  import io.reactivex.rxjava3.internal.fuseable.*;
  import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
  import io.reactivex.rxjava3.internal.util.*;
  import io.reactivex.rxjava3.operators.QueueSubscription;
  import io.reactivex.rxjava3.operators.SimpleQueue;
  import io.reactivex.rxjava3.operators.SpscArrayQueue;
  import io.reactivex.rxjava3.plugins.RxJavaPlugins;
  
  /**
   * Shares a single underlying connection to the upstream Publisher
   * and multicasts events to all subscribed subscribers until the upstream
   * completes or the connection is disposed.
   * <p>
   * The difference to FlowablePublish is that when the upstream terminates,
   * late subscribers will receive that terminal event until the connection is
   * disposed and the ConnectableFlowable is reset to its fresh state.
   *
   * @param <T> the element type
   * @since 2.2.10
   */
  public final class FlowablePublish<T> extends ConnectableFlowable<T>
  implements HasUpstreamPublisher<T> {
  
      final Publisher<T> source;
  
      final int bufferSize;
  
      final AtomicReference<PublishConnection<T>> current;
  
      public FlowablePublish(Publisher<T> source, int bufferSize) {
          this.source = source;
          this.bufferSize = bufferSize;
          this.current = new AtomicReference<>();
      }
  
      @Override
      public Publisher<T> source() {
          return source;
      }
  
      @Override
      public void connect(Consumer<? super Disposable> connection) {
          PublishConnection<T> conn;
          boolean doConnect = false;
  
          for (;;) {
              conn = current.get();
  
              if (conn == null || conn.isDisposed()) {
                  PublishConnection<T> fresh = new PublishConnection<>(current, bufferSize);
                  if (!current.compareAndSet(conn, fresh)) {
                      continue;
                  }
                  conn = fresh;
              }
  
              doConnect = !conn.connect.get() && conn.connect.compareAndSet(false, true);
              break;
          }
  
          try {
              connection.accept(conn);
          } catch (Throwable ex) {
              Exceptions.throwIfFatal(ex);
              throw ExceptionHelper.wrapOrThrow(ex);
          }
  
          if (doConnect) {
              source.subscribe(conn);
          }
      }
  
      @Override
      protected void subscribeActual(Subscriber<? super T> s) {
          PublishConnection<T> conn;
  
          for (;;) {
              conn = current.get();
  
              // don't create a fresh connection if the current is disposed
              if (conn == null) {
                  PublishConnection<T> fresh = new PublishConnection<>(current, bufferSize);
                  if (!current.compareAndSet(conn, fresh)) {
                      continue;
                  }
                  conn = fresh;
              }
  
              break;
          }
  
          InnerSubscription<T> inner = new InnerSubscription<>(s, conn);
          s.onSubscribe(inner);
  
          if (conn.add(inner)) {
              if (inner.isCancelled()) {
                  conn.remove(inner);
              } else {
                  conn.drain();
              }
              return;
          }
  
          Throwable ex = conn.error;
          if (ex != null) {
              inner.downstream.onError(ex);
          } else {
              inner.downstream.onComplete();
          }
      }
  
      @Override
      public void reset() {
          PublishConnection<T> conn = current.get();
          if (conn != null && conn.isDisposed()) {
              current.compareAndSet(conn, null);
          }
      }
  
      static final class PublishConnection<T>
      extends AtomicInteger
      implements FlowableSubscriber<T>, Disposable {
  
          private static final long serialVersionUID = -1672047311619175801L;
  
          final AtomicReference<PublishConnection<T>> current;
  
          final AtomicReference<Subscription> upstream;
  
          final AtomicBoolean connect;
  
          final AtomicReference<InnerSubscription<T>[]> subscribers;
  
          final int bufferSize;
  
          volatile SimpleQueue<T> queue;
  
          int sourceMode;
  
          volatile boolean done;
          Throwable error;
  
          int consumed;
  
          @SuppressWarnings("rawtypes")
          static final InnerSubscription[] EMPTY = new InnerSubscription[0];
          @SuppressWarnings("rawtypes")
          static final InnerSubscription[] TERMINATED = new InnerSubscription[0];
  
          @SuppressWarnings("unchecked")
          PublishConnection(AtomicReference<PublishConnection<T>> current, int bufferSize) {
              this.current = current;
              this.upstream = new AtomicReference<>();
              this.connect = new AtomicBoolean();
              this.bufferSize = bufferSize;
              this.subscribers = new AtomicReference<>(EMPTY);
          }
  
          @SuppressWarnings("unchecked")
          @Override
          public void dispose() {
              subscribers.getAndSet(TERMINATED);
              current.compareAndSet(this, null);
              SubscriptionHelper.cancel(upstream);
          }
  
          @Override
          public boolean isDisposed() {
              return subscribers.get() == TERMINATED;
          }
  
          @Override
          public void onSubscribe(Subscription s) {
              if (SubscriptionHelper.setOnce(this.upstream, s)) {
                  if (s instanceof QueueSubscription) {
                      @SuppressWarnings("unchecked")
                      QueueSubscription<T> qs = (QueueSubscription<T>) s;
  
                      int m = qs.requestFusion(QueueSubscription.ANY | QueueSubscription.BOUNDARY);
                      if (m == QueueSubscription.SYNC) {
                          sourceMode = m;
                          queue = qs;
                          done = true;
                          drain();
                          return;
                      }
                      if (m == QueueSubscription.ASYNC) {
                          sourceMode = m;
                          queue = qs;
                          s.request(bufferSize);
                          return;
                      }
                  }
  
                  queue = new SpscArrayQueue<>(bufferSize);
  
                  s.request(bufferSize);
              }
          }
  
          @Override
          public void onNext(T t) {
              // we expect upstream to honor backpressure requests
              if (sourceMode == QueueSubscription.NONE && !queue.offer(t)) {
                  onError(new QueueOverflowException());
                  return;
              }
              // since many things can happen concurrently, we have a common dispatch
              // loop to act on the current state serially
              drain();
          }
  
          @Override
          public void onError(Throwable t) {
              if (done) {
                  RxJavaPlugins.onError(t);
              } else {
                  error = t;
                  done = true;
                  drain();
              }
          }
  
          @Override
          public void onComplete() {
              done = true;
              drain();
          }
  
          void drain() {
              if (getAndIncrement() != 0) {
                  return;
              }
  
              int missed = 1;
              SimpleQueue<T> queue = this.queue;
              int consumed = this.consumed;
              int limit = this.bufferSize - (this.bufferSize >> 2);
              boolean async = this.sourceMode != QueueSubscription.SYNC;
  
              outer:
              for (;;) {
                  if (queue != null) {
                      long minDemand = Long.MAX_VALUE;
                      boolean hasDemand = false;
  
                      InnerSubscription<T>[] consumers = subscribers.get();
  
                      for (InnerSubscription<T> inner : consumers) {
                          long request = inner.get();
                          if (request != Long.MIN_VALUE) {
                              hasDemand = true;
                              minDemand = Math.min(request - inner.emitted, minDemand);
                          }
                      }
  
                      if (!hasDemand) {
                          minDemand = 0L;
                      }
  
                      while (minDemand != 0L) {
                          boolean d = done;
                          T v;
  
                          try {
                              v = queue.poll();
                          } catch (Throwable ex) {
                              Exceptions.throwIfFatal(ex);
                              upstream.get().cancel();
                              queue.clear();
                              done = true;
                              signalError(ex);
                              return;
                          }
  
                          boolean empty = v == null;
  
                          if (checkTerminated(d, empty)) {
                              return;
                          }
  
                          if (empty) {
                              break;
                          }
  
                          for (InnerSubscription<T> inner : consumers) {
                              if (!inner.isCancelled()) {
                                  inner.downstream.onNext(v);
                                  inner.emitted++;
                              }
                          }
  
                          if (async && ++consumed == limit) {
                              consumed = 0;
                              upstream.get().request(limit);
                          }
                          minDemand--;
  
                          if (consumers != subscribers.get()) {
                              continue outer;
                          }
                      }
  
                      if (checkTerminated(done, queue.isEmpty())) {
                          return;
                      }
                  }
  
                  this.consumed = consumed;
                  missed = addAndGet(-missed);
                  if (missed == 0) {
                      break;
                  }
                  if (queue == null) {
                      queue = this.queue;
                  }
              }
          }
  
          @SuppressWarnings("unchecked")
          boolean checkTerminated(boolean isDone, boolean isEmpty) {
              if (isDone && isEmpty) {
                  Throwable ex = error;
  
                  if (ex != null) {
                      signalError(ex);
                  } else {
                      for (InnerSubscription<T> inner : subscribers.getAndSet(TERMINATED)) {
                          if (!inner.isCancelled()) {
                              inner.downstream.onComplete();
                          }
                      }
                  }
                  return true;
              }
              return false;
          }
  
          @SuppressWarnings("unchecked")
          void signalError(Throwable ex) {
              for (InnerSubscription<T> inner : subscribers.getAndSet(TERMINATED)) {
                  if (!inner.isCancelled()) {
                      inner.downstream.onError(ex);
                  }
              }
          }
  
          boolean add(InnerSubscription<T> inner) {
              // the state can change so we do a CAS loop to achieve atomicity
              for (;;) {
                  // get the current producer array
                  InnerSubscription<T>[] c = subscribers.get();
                  // if this subscriber-to-source reached a terminal state by receiving
                  // an onError or onComplete, just refuse to add the new producer
                  if (c == TERMINATED) {
                      return false;
                  }
                  // we perform a copy-on-write logic
                  int len = c.length;
                  @SuppressWarnings("unchecked")
                  InnerSubscription<T>[] u = new InnerSubscription[len + 1];
                  System.arraycopy(c, 0, u, 0, len);
                  u[len] = inner;
                  // try setting the subscribers array
                  if (subscribers.compareAndSet(c, u)) {
                      return true;
                  }
                  // if failed, some other operation succeeded (another add, remove or termination)
                  // so retry
              }
          }
  
          @SuppressWarnings("unchecked")
          void remove(InnerSubscription<T> inner) {
              // the state can change so we do a CAS loop to achieve atomicity
              for (;;) {
                  // let's read the current subscribers array
                  InnerSubscription<T>[] c = subscribers.get();
                  int len = c.length;
                  // if it is either empty or terminated, there is nothing to remove so we quit
                  if (len == 0) {
                      break;
                  }
                  // let's find the supplied producer in the array
                  // although this is O(n), we don't expect too many child subscribers in general
                  int j = -1;
                  for (int i = 0; i < len; i++) {
                      if (c[i] == inner) {
                          j = i;
                          break;
                      }
                  }
                  // we didn't find it so just quit
                  if (j < 0) {
                      return;
                  }
                  // we do copy-on-write logic here
                  InnerSubscription<T>[] u;
                  // we don't create a new empty array if producer was the single inhabitant
                  // but rather reuse an empty array
                  if (len == 1) {
                      u = EMPTY;
                  } else {
                      // otherwise, create a new array one less in size
                      u = new InnerSubscription[len - 1];
                      // copy elements being before the given producer
                      System.arraycopy(c, 0, u, 0, j);
                      // copy elements being after the given producer
                      System.arraycopy(c, j + 1, u, j, len - j - 1);
                  }
                  // try setting this new array as
                  if (subscribers.compareAndSet(c, u)) {
                      break;
                  }
                  // if we failed, it means something else happened
                  // (a concurrent add/remove or termination), we need to retry
              }
          }
      }
  
      static final class InnerSubscription<T> extends AtomicLong
      implements Subscription {
  
          private static final long serialVersionUID = 2845000326761540265L;
  
          final Subscriber<? super T> downstream;
  
          final PublishConnection<T> parent;
  
          long emitted;
  
          InnerSubscription(Subscriber<? super T> downstream, PublishConnection<T> parent) {
              this.downstream = downstream;
              this.parent = parent;
          }
  
          @Override
          public void request(long n) {
              if (SubscriptionHelper.validate(n)) {
                  BackpressureHelper.addCancel(this, n);
                  parent.drain();
              }
          }
  
          @Override
          public void cancel() {
              if (getAndSet(Long.MIN_VALUE) != Long.MIN_VALUE) {
                  parent.remove(this);
                  parent.drain();
              }
          }
  
          public boolean isCancelled() {
              return get() == Long.MIN_VALUE;
          }
      }
  }
  