code: public static <T> Stream<T> observe(final BlockingQueue<T> queue, final Duration duration, final Runnable onComplete) {
          N.checkArgNotNull(queue, "queue");
          N.checkArgNotNull(duration, "duration");
          N.checkArgNotNull(onComplete, "onComplete");
  
          final long now = System.currentTimeMillis();
          final long endTime = duration.toMillis() >= Long.MAX_VALUE - now ? Long.MAX_VALUE : now + duration.toMillis();
  
          final Iterator<T> iter = new ObjIterator<T>() {
              private T next = null;
  
              @Override
              public boolean hasNext() {
                  if (next == null) {
                      final long curTime = System.currentTimeMillis();
  
                      if (curTime <= endTime) {
                          try {
                              next = queue.poll(endTime - curTime, TimeUnit.MILLISECONDS);
                          } 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);
  Stream.observe(queue, Duration.ofMillis(100)).filter(s -> s.startsWith("a")).asyncRun(s -> s.forEach(Fn.println()));
  N.asList("a", "b", "ab", "bc", "1", "a").forEach(queue::add);
  N.sleep(10);
  N.println("==================");
  N.sleep(100);
  N.println("==================");
  N.sleep(10);
  </code>
  </pre>
  
  @param queue
  @param duration
  @param onComplete
  @return