ConcurrentInboundEdgeStream { public static InboundEdgeStream create( @Nonnull ConcurrentConveyor<Object> conveyor, int ordinal, int priority, boolean waitForAllBarriers, @Nonnull String debugName, @Nullable ComparatorEx<?> comparator ) { if (comparator == null) { return new RoundRobinDrain(conveyor, ordinal, priority, debugName, waitForAllBarriers); } else { return new OrderedDrain(conveyor, ordinal, priority, debugName, comparator); } } private  ConcurrentInboundEdgeStream(); static InboundEdgeStream create(
            @Nonnull ConcurrentConveyor<Object> conveyor,
            int ordinal,
            int priority,
            boolean waitForAllBarriers,
            @Nonnull String debugName,
            @Nullable ComparatorEx<?> comparator
    );  }
@Test public void when_receivingBarriers_then_waitForBarrier() { stream = ConcurrentInboundEdgeStream.create(conveyor, 0, 0, true, "cies", null); add(q1, barrier(0)); add(q2, 1); drainAndAssert(MADE_PROGRESS, 1); add(q1, 2); drainAndAssert(NO_PROGRESS); add(q2, barrier(0)); drainAndAssert(MADE_PROGRESS, barrier(0)); drainAndAssert(MADE_PROGRESS, 2); }
@Test public void when_receivingBarriersWhileDone_then_coalesce() { stream = ConcurrentInboundEdgeStream.create(conveyor, 0, 0, true, "cies", null); add(q1, 1, barrier(0)); add(q2, DONE_ITEM); drainAndAssert(MADE_PROGRESS, 1, barrier(0)); add(q1, DONE_ITEM); drainAndAssert(DONE); }