code: @Override
      public void onComplete() {
  
  
          counter.active.decrementAndGet();
          counter.subscription.removeValue(subscription);
          if (queue != null && counter.active.get() == 0) {
  
              if (counter.completable) {
  
                  if(counter.closing.compareAndSet(false,true)) {
                      counter.closed = true;
                      queue.addContinuation(new Continuation(
                              () -> {
                                  final List current = new ArrayList();
                                  while (queue.size() > 0) {
                                      try {
  
                                          current.add(queue.get());
                                      }catch(ClosedQueueException e){
  
                                          break;
                                      }
                                  }
  
                                  throw new ClosedQueueException(
                                          current);
                              }));
  
                      queue.close();
                  }
              }
  
          }
  
      }
docs: /* (non-Javadoc)
  @see org.reactivestreams.Subscriber#onComplete()