DefaultOperatorQueue implements OperatorQueue { @Override public void drain ( final boolean maySkipBlocking, final TupleQueueDrainer drainer, final Function<PartitionKey, TuplesImpl> tuplesSupplier ) { if ( drainer.drain( maySkipBlocking, null, tupleQueues, tuplesSupplier ) ) { int count = 1; while ( count++ < drainLimit && drainer.drain( true, null, tupleQueues, tuplesSupplier ) ) { } } } DefaultOperatorQueue( final String operatorId,
                                  final int inputPortCount,
                                  final ThreadingPref threadingPref,
                                  final TupleQueue[] tupleQueues,
                                  final int drainLimit ); @Override String getOperatorId(); @Override int getInputPortCount(); @Override int offer( final int portIndex, final List<Tuple> tuples ); @Override int offer( final int portIndex, final List<Tuple> tuples, final int fromIndex ); @Override void drain( final boolean maySkipBlocking,
                        final TupleQueueDrainer drainer,
                        final Function<PartitionKey, TuplesImpl> tuplesSupplier ); @Override void clear(); @Override void setTupleCounts( final int[] tupleCounts, final TupleAvailabilityByPort tupleAvailabilityByPort ); @Override boolean isEmpty(); @Override void ensureCapacity( final int capacity ); ThreadingPref getThreadingPref(); TupleQueue getTupleQueue( final int portIndex );  }
@Test public void testDrain () { final int inputPortCount = 3; final OperatorQueue operatorQueue = createOperatorQueue( inputPortCount, threadingPref ); final TuplesImpl input = createTuples( inputPortCount, TUPLE_QUEUE_SIZE ); for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { operatorQueue.offer( portIndex, input.getTuples( portIndex ) ); } final List<TuplesImpl> results = new ArrayList<>(); final Function<PartitionKey, TuplesImpl> tuplesSupplier = objects -> { final TuplesImpl tuples = new TuplesImpl( inputPortCount ); results.add( tuples ); return tuples; }; final MultiPortDrainer drainer = new NonBlockingMultiPortDisjunctiveDrainer( inputPortCount, 100 ); final int requiredTupleCount = TUPLE_QUEUE_SIZE / 2; drainer.setParameters( TupleAvailabilityByCount.EXACT, new int[] { 0, 1, 2 }, new int[] { requiredTupleCount, requiredTupleCount, requiredTupleCount } ); operatorQueue.drain( drainer, tuplesSupplier ); assertThat( results.size(), equalTo( 2 ) ); }