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.Cancellable;
  import io.reactivex.rxjava3.internal.disposables.*;
  import io.reactivex.rxjava3.internal.util.*;
  import io.reactivex.rxjava3.operators.SimpleQueue;
  import io.reactivex.rxjava3.operators.SpscLinkedArrayQueue;
  import io.reactivex.rxjava3.plugins.RxJavaPlugins;
  
  public final class ObservableCreate<T> extends Observable<T> {
      final ObservableOnSubscribe<T> source;
  
      public ObservableCreate(ObservableOnSubscribe<T> source) {
          this.source = source;
      }
  
      @Override
      protected void subscribeActual(Observer<? super T> observer) {
          CreateEmitter<T> parent = new CreateEmitter<>(observer);
          observer.onSubscribe(parent);
  
          try {
              source.subscribe(parent);
          } catch (Throwable ex) {
              Exceptions.throwIfFatal(ex);
              parent.onError(ex);
          }
      }
  
      static final class CreateEmitter<T>
      extends AtomicReference<Disposable>
      implements ObservableEmitter<T>, Disposable {
  
          private static final long serialVersionUID = -3434801548987643227L;
  
          final Observer<? super T> observer;
  
          CreateEmitter(Observer<? super T> observer) {
              this.observer = observer;
          }
  
          @Override
          public void onNext(T t) {
              if (t == null) {
                  onError(ExceptionHelper.createNullPointerException("onNext called with a null value."));
                  return;
              }
              if (!isDisposed()) {
                  observer.onNext(t);
              }
          }
  
          @Override
          public void onError(Throwable t) {
              if (!tryOnError(t)) {
                  RxJavaPlugins.onError(t);
              }
          }
  
          @Override
          public boolean tryOnError(Throwable t) {
              if (t == null) {
                  t = ExceptionHelper.createNullPointerException("onError called with a null Throwable.");
              }
              if (!isDisposed()) {
                  try {
                      observer.onError(t);
                  } finally {
                      dispose();
                  }
                  return true;
              }
              return false;
          }
  
          @Override
          public void onComplete() {
              if (!isDisposed()) {
                  try {
                      observer.onComplete();
                  } finally {
                      dispose();
                  }
              }
          }
  
          @Override
          public void setDisposable(Disposable d) {
              DisposableHelper.set(this, d);
          }
  
          @Override
          public void setCancellable(Cancellable c) {
              setDisposable(new CancellableDisposable(c));
          }
  
          @Override
          public ObservableEmitter<T> serialize() {
              return new SerializedEmitter<>(this);
          }
  
          @Override
          public void dispose() {
              DisposableHelper.dispose(this);
          }
  
          @Override
          public boolean isDisposed() {
              return DisposableHelper.isDisposed(get());
          }
  
          @Override
          public String toString() {
              return String.format("%s{%s}", getClass().getSimpleName(), super.toString());
          }
      }
  
      /**
       * Serializes calls to onNext, onError and onComplete.
       *
       * @param <T> the value type
       */
      static final class SerializedEmitter<T>
      extends AtomicInteger
      implements ObservableEmitter<T> {
  
          private static final long serialVersionUID = 4883307006032401862L;
  
          final ObservableEmitter<T> emitter;
  
          final AtomicThrowable errors;
  
          final SpscLinkedArrayQueue<T> queue;
  
          volatile boolean done;
  
          SerializedEmitter(ObservableEmitter<T> emitter) {
              this.emitter = emitter;
              this.errors = new AtomicThrowable();
              this.queue = new SpscLinkedArrayQueue<>(16);
          }
  
          @Override
          public void onNext(T t) {
              if (done || emitter.isDisposed()) {
                  return;
              }
              if (t == null) {
                  onError(ExceptionHelper.createNullPointerException("onNext called with a null value."));
                  return;
              }
              if (get() == 0 && compareAndSet(0, 1)) {
                  emitter.onNext(t);
                  if (decrementAndGet() == 0) {
                      return;
                  }
              } else {
                  SimpleQueue<T> q = queue;
                  synchronized (q) {
                      q.offer(t);
                  }
                  if (getAndIncrement() != 0) {
                      return;
                  }
              }
              drainLoop();
          }
  
          @Override
          public void onError(Throwable t) {
              if (!tryOnError(t)) {
                  RxJavaPlugins.onError(t);
              }
          }
  
          @Override
          public boolean tryOnError(Throwable t) {
              if (done || emitter.isDisposed()) {
                  return false;
              }
              if (t == null) {
                  t = ExceptionHelper.createNullPointerException("onError called with a null Throwable.");
              }
              if (errors.tryAddThrowable(t)) {
                  done = true;
                  drain();
                  return true;
              }
              return false;
          }
  
          @Override
          public void onComplete() {
              if (done || emitter.isDisposed()) {
                  return;
              }
              done = true;
              drain();
          }
  
          void drain() {
              if (getAndIncrement() == 0) {
                  drainLoop();
              }
          }
  
          void drainLoop() {
              ObservableEmitter<T> e = emitter;
              SpscLinkedArrayQueue<T> q = queue;
              AtomicThrowable errors = this.errors;
              int missed = 1;
              for (;;) {
  
                  for (;;) {
                      if (e.isDisposed()) {
                          q.clear();
                          return;
                      }
  
                      if (errors.get() != null) {
                          q.clear();
                          errors.tryTerminateConsumer(e);
                          return;
                      }
  
                      boolean d = done;
                      T v = q.poll();
  
                      boolean empty = v == null;
  
                      if (d && empty) {
                          e.onComplete();
                          return;
                      }
  
                      if (empty) {
                          break;
                      }
  
                      e.onNext(v);
                  }
  
                  missed = addAndGet(-missed);
                  if (missed == 0) {
                      break;
                  }
              }
          }
  
          @Override
          public void setDisposable(Disposable d) {
              emitter.setDisposable(d);
          }
  
          @Override
          public void setCancellable(Cancellable c) {
              emitter.setCancellable(c);
          }
  
          @Override
          public boolean isDisposed() {
              return emitter.isDisposed();
          }
  
          @Override
          public ObservableEmitter<T> serialize() {
              return this;
          }
  
          @Override
          public String toString() {
              return emitter.toString();
          }
      }
  
  }
  