StreamAdapter { public Stream<T> stream() { int characteristics = 0; return StreamSupport.stream(() -> Spliterators.spliteratorUnknownSize(new Iterator<T>() { @Override public boolean hasNext() { return !blockingQueue.isEmpty() || !streamFinished.get(); } @Override public synchronized T next() { T result = null; consumerThread.set(new ConsumerThread(Thread.currentThread(), blockingQueue.isEmpty())); try { result = blockingQueue.take(); } catch (InterruptedException e) { if (!streamFinished.get()) { throw new RuntimeException(e); } } finally { consumerThread.set(null); } return result; } }, characteristics), characteristics, false).filter(item -> item != null); } StreamAdapter(); synchronized void finishStream(); synchronized boolean addItem(T item); Stream<T> stream();  }
@Test public void testStreamAdapter_NormalExecution() { StreamAdapter<Object> adapter = new StreamAdapter<>(); ConsumerTask<Object> consumerTask = new ConsumerTask<>(adapter.stream()); new Thread(consumerTask).start(); addItemsAndFinish(adapter, new Object(), new Object()); consumerTask.awaitCompletion(); assertEquals(2, consumerTask.getItemsConsumed()); assertSame(ConsumerTask.CompletionKind.NORMAL, consumerTask.getCompletionKind()); }