/** Observe through individual queue per observer. */
 static final class ObserveOnSubscriber<T> extends Subscriber<T> implements Action0 {
  final Subscriber<? super T> child;
  final Scheduler.Worker recursiveScheduler;
  final NotificationLite<T> on;
  final boolean delayError;
  final Queue<Object> queue;
  /** The emission threshold that should trigger a replenishing request. */
  final int limit;

  // the status of the current stream
  volatile boolean finished;

  final AtomicLong requested = new AtomicLong();

  final AtomicLong counter = new AtomicLong();

  /**
   * The single exception if not null, should be written before setting finished (release) and read after
   * reading finished (acquire).
   */
  Throwable error;

  /** Remembers how many elements have been emitted before the requests run out. */
  long emitted;

  // do NOT pass the Subscriber through to couple the subscription chain ... unsubscribing on the parent should
  // not prevent anything downstream from consuming, which will happen if the Subscription is chained
  public ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> child, boolean delayError, int bufferSize) {
   this.child = child;
   this.recursiveScheduler = scheduler.createWorker();
   this.delayError = delayError;
   this.on = NotificationLite.instance();
   int calculatedSize = (bufferSize > 0) ? bufferSize : RxRingBuffer.SIZE;
   // this formula calculates the 75% of the bufferSize, rounded up to the next integer
   this.limit = calculatedSize - (calculatedSize >> 2);
   if (UnsafeAccess.isUnsafeAvailable()) {
    queue = new SpscArrayQueue<Object>(calculatedSize);
   } else {
    queue = new SpscAtomicArrayQueue<Object>(calculatedSize);
   }
   // signal that this is an async operator capable of receiving this many
   request(calculatedSize);
  }

  void init() {
   // don't want this code in the constructor because `this` can escape through the
   // setProducer call
   Subscriber<? super T> localChild = child;

   localChild.setProducer(new Producer() {

    @Override
    public void request(long n) {
     if (n > 0L) {
      BackpressureUtils.getAndAddRequest(requested, n);
      schedule();
     }
    }

   });
   localChild.add(recursiveScheduler);
   localChild.add(this);
  }

  @Override
  public void onNext(final T t) {
   if (isUnsubscribed() || finished) {
    return;
   }
   if (!queue.offer(on.next(t))) {
    onError(new MissingBackpressureException());
    return;
   }
   schedule();
  }

  @Override
  public void onCompleted() {
   if (isUnsubscribed() || finished) {
    return;
   }
   finished = true;
   schedule();
  }

  @Override
  public void onError(final Throwable e) {
   if (isUnsubscribed() || finished) {
    RxJavaHooks.onError(e);
    return;
   }
   error = e;
   finished = true;
   schedule();
  }

  protected void schedule() {
   if (counter.getAndIncrement() == 0) {
    recursiveScheduler.schedule(this);
   }
  }
 }