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.observable;
  
  import java.util.concurrent.atomic.*;
  
  import io.reactivex.rxjava3.core.*;
  import io.reactivex.rxjava3.disposables.Disposable;
  import io.reactivex.rxjava3.exceptions.Exceptions;
  import io.reactivex.rxjava3.functions.Consumer;
  import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
  import io.reactivex.rxjava3.internal.fuseable.HasUpstreamObservableSource;
  import io.reactivex.rxjava3.internal.util.ExceptionHelper;
  import io.reactivex.rxjava3.observables.ConnectableObservable;
  import io.reactivex.rxjava3.plugins.RxJavaPlugins;
  
  /**
   * Shares a single underlying connection to the upstream ObservableSource
   * and multicasts events to all subscribed observers until the upstream
   * completes or the connection is disposed.
   * <p>
   * The difference to ObservablePublish is that when the upstream terminates,
   * late observers will receive that terminal event until the connection is
   * disposed and the ConnectableObservable is reset to its fresh state.
   *
   * @param <T> the element type
   * @since 2.2.10
   */
  public final class ObservablePublish<T> extends ConnectableObservable<T>
  implements HasUpstreamObservableSource<T> {
  
      final ObservableSource<T> source;
  
      final AtomicReference<PublishConnection<T>> current;
  
      public ObservablePublish(ObservableSource<T> source) {
          this.source = source;
          this.current = new AtomicReference<>();
      }
  
      @Override
      public void connect(Consumer<? super Disposable> connection) {
          boolean doConnect = false;
          PublishConnection<T> conn;
  
          for (;;) {
              conn = current.get();
  
              if (conn == null || conn.isDisposed()) {
                  PublishConnection<T> fresh = new PublishConnection<>(current);
                  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(Observer<? super T> observer) {
          PublishConnection<T> conn;
  
          for (;;) {
              conn = current.get();
              // we don't create a fresh connection if the current is terminated
              if (conn == null) {
                  PublishConnection<T> fresh = new PublishConnection<>(current);
                  if (!current.compareAndSet(conn, fresh)) {
                      continue;
                  }
                  conn = fresh;
              }
              break;
          }
  
          InnerDisposable<T> inner = new InnerDisposable<>(observer, conn);
          observer.onSubscribe(inner);
          if (conn.add(inner)) {
              if (inner.isDisposed()) {
                  conn.remove(inner);
              }
              return;
          }
          // Late observers will be simply terminated
          Throwable error = conn.error;
          if (error != null) {
              observer.onError(error);
          } else {
              observer.onComplete();
          }
      }
  
      @Override
      public void reset() {
          PublishConnection<T> conn = current.get();
          if (conn != null && conn.isDisposed()) {
              current.compareAndSet(conn, null);
          }
      }
  
      @Override
      public ObservableSource<T> source() {
          return source;
      }
  
      static final class PublishConnection<T>
      extends AtomicReference<InnerDisposable<T>[]>
      implements Observer<T>, Disposable {
  
          private static final long serialVersionUID = -3251430252873581268L;
  
          final AtomicBoolean connect;
  
          final AtomicReference<PublishConnection<T>> current;
  
          final AtomicReference<Disposable> upstream;
  
          @SuppressWarnings("rawtypes")
          static final InnerDisposable[] EMPTY = new InnerDisposable[0];
  
          @SuppressWarnings("rawtypes")
          static final InnerDisposable[] TERMINATED = new InnerDisposable[0];
  
          Throwable error;
  
          @SuppressWarnings("unchecked")
          PublishConnection(AtomicReference<PublishConnection<T>> current) {
              this.connect = new AtomicBoolean();
              this.current = current;
              this.upstream = new AtomicReference<>();
              lazySet(EMPTY);
          }
  
          @SuppressWarnings("unchecked")
          @Override
          public void dispose() {
              getAndSet(TERMINATED);
              current.compareAndSet(this, null);
              DisposableHelper.dispose(upstream);
          }
  
          @Override
          public boolean isDisposed() {
              return get() == TERMINATED;
          }
  
          @Override
          public void onSubscribe(Disposable d) {
              DisposableHelper.setOnce(upstream, d);
          }
  
          @Override
          public void onNext(T t) {
              for (InnerDisposable<T> inner : get()) {
                  inner.downstream.onNext(t);
              }
          }
  
          @Override
          @SuppressWarnings("unchecked")
          public void onError(Throwable e) {
              if (upstream.get() != DisposableHelper.DISPOSED) {
                  error = e;
                  upstream.lazySet(DisposableHelper.DISPOSED);
                  for (InnerDisposable<T> inner : getAndSet(TERMINATED)) {
                      inner.downstream.onError(e);
                  }
              } else {
                  RxJavaPlugins.onError(e);
              }
          }
  
          @Override
          @SuppressWarnings("unchecked")
          public void onComplete() {
              upstream.lazySet(DisposableHelper.DISPOSED);
              for (InnerDisposable<T> inner : getAndSet(TERMINATED)) {
                  inner.downstream.onComplete();
              }
          }
  
          public boolean add(InnerDisposable<T> inner) {
              for (;;) {
                  InnerDisposable<T>[] a = get();
                  if (a == TERMINATED) {
                      return false;
                  }
                  int n = a.length;
                  @SuppressWarnings("unchecked")
                  InnerDisposable<T>[] b = new InnerDisposable[n + 1];
                  System.arraycopy(a, 0, b, 0, n);
                  b[n] = inner;
                  if (compareAndSet(a, b)) {
                      return true;
                  }
              }
          }
  
          @SuppressWarnings("unchecked")
          public void remove(InnerDisposable<T> inner) {
              for (;;) {
                  InnerDisposable<T>[] a = get();
                  int n = a.length;
                  if (n == 0) {
                      return;
                  }
  
                  int j = -1;
                  for (int i = 0; i < n; i++) {
                      if (a[i] == inner) {
                          j = i;
                          break;
                      }
                  }
  
                  if (j < 0) {
                      return;
                  }
                  InnerDisposable<T>[] b = EMPTY;
                  if (n != 1) {
                      b = new InnerDisposable[n - 1];
                      System.arraycopy(a, 0, b, 0, j);
                      System.arraycopy(a, j + 1, b, j, n - j - 1);
                  }
                  if (compareAndSet(a, b)) {
                      return;
                  }
              }
          }
      }
  
      /**
       * Intercepts the dispose signal from the downstream and
       * removes itself from the connection's observers array
       * at most once.
       * @param <T> the element type
       */
      static final class InnerDisposable<T>
      extends AtomicReference<PublishConnection<T>>
      implements Disposable {
  
          private static final long serialVersionUID = 7463222674719692880L;
  
          final Observer<? super T> downstream;
  
          InnerDisposable(Observer<? super T> downstream, PublishConnection<T> parent) {
              this.downstream = downstream;
              lazySet(parent);
          }
  
          @Override
          public void dispose() {
              PublishConnection<T> p = getAndSet(null);
              if (p != null) {
                  p.remove(this);
              }
          }
  
          @Override
          public boolean isDisposed() {
              return get() == null;
          }
      }
  }
  
  