/** 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);   }  } }