@Test public void shouldDrainAllTuplesFromSinglePort () { final TupleQueue tupleQueue = new SingleThreadedTupleQueue( 1 ); final Tuple tuple = new Tuple(); tupleQueue.offer( tuple ); final GreedyDrainer greedyDrainer = new GreedyDrainer( 1 ); final TuplesImpl tuples = new TuplesImpl( 1 ); final boolean success = greedyDrainer.drain( null, new TupleQueue[] { tupleQueue }, key -> tuples ); assertFalse( success ); assertThat( tuples.getNonEmptyPortCount(), equalTo( 1 ) ); assertThat( tuples.getTupleCount( 0 ), equalTo( 1 ) ); assertTrue( tuple == tuples.getTupleOrFail( 0, 0 ) ); assertThat( tupleQueue.size(), equalTo( 0 ) ); }
@Override public boolean drain ( final boolean maySkipBlocking, final PartitionKey key, final TupleQueue[] queues, final Function<PartitionKey, TuplesImpl> tuplesSupplier ) { checkArgument( queues != null ); checkArgument( queues.length == inputPortCount ); checkArgument( tuplesSupplier != null ); boolean empty = true; for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { if ( !queues[ portIndex ].isEmpty() ) { empty = false; break; } } if ( empty ) { return false; } final TuplesImpl tuples = tuplesSupplier.apply( key ); for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { queues[ portIndex ].poll( Integer.MAX_VALUE, tuples.getTuplesModifiable( portIndex ) ); } return false; }
GreedyDrainer implements TupleQueueDrainer { @Override public boolean drain ( final boolean maySkipBlocking, final PartitionKey key, final TupleQueue[] queues, final Function<PartitionKey, TuplesImpl> tuplesSupplier ) { checkArgument( queues != null ); checkArgument( queues.length == inputPortCount ); checkArgument( tuplesSupplier != null ); boolean empty = true; for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { if ( !queues[ portIndex ].isEmpty() ) { empty = false; break; } } if ( empty ) { return false; } final TuplesImpl tuples = tuplesSupplier.apply( key ); for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { queues[ portIndex ].poll( Integer.MAX_VALUE, tuples.getTuplesModifiable( portIndex ) ); } return false; } }
GreedyDrainer implements TupleQueueDrainer { @Override public boolean drain ( final boolean maySkipBlocking, final PartitionKey key, final TupleQueue[] queues, final Function<PartitionKey, TuplesImpl> tuplesSupplier ) { checkArgument( queues != null ); checkArgument( queues.length == inputPortCount ); checkArgument( tuplesSupplier != null ); boolean empty = true; for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { if ( !queues[ portIndex ].isEmpty() ) { empty = false; break; } } if ( empty ) { return false; } final TuplesImpl tuples = tuplesSupplier.apply( key ); for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { queues[ portIndex ].poll( Integer.MAX_VALUE, tuples.getTuplesModifiable( portIndex ) ); } return false; } GreedyDrainer( final int inputPortCount ); }
GreedyDrainer implements TupleQueueDrainer { @Override public boolean drain ( final boolean maySkipBlocking, final PartitionKey key, final TupleQueue[] queues, final Function<PartitionKey, TuplesImpl> tuplesSupplier ) { checkArgument( queues != null ); checkArgument( queues.length == inputPortCount ); checkArgument( tuplesSupplier != null ); boolean empty = true; for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { if ( !queues[ portIndex ].isEmpty() ) { empty = false; break; } } if ( empty ) { return false; } final TuplesImpl tuples = tuplesSupplier.apply( key ); for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { queues[ portIndex ].poll( Integer.MAX_VALUE, tuples.getTuplesModifiable( portIndex ) ); } return false; } GreedyDrainer( final int inputPortCount ); @Override boolean drain( final boolean maySkipBlocking, final PartitionKey key, final TupleQueue[] queues,
                           final Function<PartitionKey, TuplesImpl> tuplesSupplier ); }
GreedyDrainer implements TupleQueueDrainer { @Override public boolean drain ( final boolean maySkipBlocking, final PartitionKey key, final TupleQueue[] queues, final Function<PartitionKey, TuplesImpl> tuplesSupplier ) { checkArgument( queues != null ); checkArgument( queues.length == inputPortCount ); checkArgument( tuplesSupplier != null ); boolean empty = true; for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { if ( !queues[ portIndex ].isEmpty() ) { empty = false; break; } } if ( empty ) { return false; } final TuplesImpl tuples = tuplesSupplier.apply( key ); for ( int portIndex = 0; portIndex < inputPortCount; portIndex++ ) { queues[ portIndex ].poll( Integer.MAX_VALUE, tuples.getTuplesModifiable( portIndex ) ); } return false; } GreedyDrainer( final int inputPortCount ); @Override boolean drain( final boolean maySkipBlocking, final PartitionKey key, final TupleQueue[] queues,
                           final Function<PartitionKey, TuplesImpl> tuplesSupplier );  }