### Question:
TriangleStream extends Algorithm<TriangleStream, Stream<TriangleStream.Result>> { @Override public Stream<Result> compute() { submitTasks(); final TerminationFlag flag = getTerminationFlag(); final Iterator<Result> it = new Iterator<Result>() { @Override public boolean hasNext() { return flag.running() && (runningThreads.get() > 0 || !resultQueue.isEmpty()); } @Override public Result next() { Result result = null; while (hasNext() && result == null) { result = resultQueue.poll(); } return result; } }; return StreamSupport .stream(Spliterators.spliteratorUnknownSize(it, 0), false) .filter(Objects::nonNull); } TriangleStream(Graph graph, ExecutorService executorService, int concurrency); @Override TriangleStream me(); @Override void release(); @Override Stream<Result> compute();  }### Answer:
@Test void testSequential() { TripleConsumer mock = mock(TripleConsumer.class); new TriangleStream(graph, Pools.DEFAULT, 1) .compute() .forEach(r -> mock.consume(r.nodeA, r.nodeB, r.nodeC)); verify(mock, times((int) TRIANGLES)).consume(eq(centerId), anyLong(), anyLong()); }
@Test void testParallel() { TripleConsumer mock = mock(TripleConsumer.class); new TriangleStream(graph, Pools.DEFAULT, 8) .compute() .forEach(r -> mock.consume(r.nodeA, r.nodeB, r.nodeC)); verify(mock, times((int) TRIANGLES)).consume(eq(centerId), anyLong(), anyLong()); }