code: public static <T> Stream<T> observe(final BlockingQueue<T> queue, final BooleanSupplier hasMore, final long maxWaitIntervalInMillis,
              final Runnable onComplete) {
          N.checkArgNotNull(queue, "queue");
          N.checkArgNotNull(hasMore, "hasMore");
          N.checkArgPositive(maxWaitIntervalInMillis, "maxWaitIntervalInMillis");
          N.checkArgNotNull(onComplete, "onComplete");
  
          final Iterator<T> iter = new ObjIterator<T>() {
              private T next = null;
  
              @Override
              public boolean hasNext() {
                  if (next == null && (hasMore.getAsBoolean() || queue.size() > 0)) {
                      try {
                          do {
                              next = queue.poll(maxWaitIntervalInMillis, TimeUnit.MILLISECONDS);
                          } while (next == null && (hasMore.getAsBoolean() || queue.size() > 0));
                      } catch (InterruptedException e) {
                          throw new RuntimeException(e);
                      }
                  }
  
                  return next != null;
              }
  
              @Override
              public T next() {
                  if (hasNext() == false) {
                      throw new NoSuchElementException();
                  }
  
                  final T res = next;
                  next = null;
                  return res;
              }
          };
  
          return of(iter).onClose(onComplete);
      }
docs: Sample code:
  
  <pre>
  <code>
  final BlockingQueue<String> queue = new ArrayBlockingQueue<>(32);
  final MutableBoolean hasMore = MutableBoolean.of(true);
  Stream.observe(queue, () -> hasMore.value(), 10).filter(s -> s.startsWith("a")).asyncRun(s -> s.forEach(Fn.println()));
  N.asList("a", "b", "ab", "bc", "1", "a").forEach(queue::add);
  N.println("==================");
  hasMore.setFalse();
  N.sleep(50);
  N.println("==================");
  </code>
  </pre>
  
  @param queue
  @param hasMore
  @param maxWaitIntervalInMillis
  @param isCompleted it will will be set to {@code true} if Stream is completed and the upstream should not continue to put elements to queue when it's completed.
  This is an output parameter.
  @return