@Test public void shouldDrainAllTuplesWithAtLeastTupleAvailabilityByCountSatisfied () { final TupleQueue tupleQueue = new MultiThreadedTupleQueue( 2 ); final Tuple tuple1 = new Tuple(); tupleQueue.offer( tuple1 ); final Tuple tuple2 = new Tuple(); tupleQueue.offer( tuple2 ); drainer.setParameters( AT_LEAST, 1 ); final boolean success = drainer.drain( null, new TupleQueue[] { tupleQueue }, tuplesSupplier ); assertTrue( success ); assertThat( result.getNonEmptyPortCount(), equalTo( 1 ) ); assertThat( result.getTupleCount( DEFAULT_PORT_INDEX ), equalTo( 2 ) ); assertTrue( tuple1 == result.getTupleOrFail( DEFAULT_PORT_INDEX, 0 ) ); assertTrue( tuple2 == result.getTupleOrFail( DEFAULT_PORT_INDEX, 1 ) ); assertThat( tupleQueue.size(), equalTo( 0 ) ); }
public final void setParameters ( final TupleAvailabilityByCount tupleAvailabilityByCount, final int tupleCount ) { checkArgument( tupleAvailabilityByCount != null ); checkArgument( tupleCount > 0, "invalid tuple count %s", tupleCount ); this.tupleCountToCheck = tupleCount; this.tupleCountToPoll = tupleAvailabilityByCount == EXACT ? tupleCount : max( tupleCount, maxBatchSize ); }
SinglePortDrainer implements TupleQueueDrainer { public final void setParameters ( final TupleAvailabilityByCount tupleAvailabilityByCount, final int tupleCount ) { checkArgument( tupleAvailabilityByCount != null ); checkArgument( tupleCount > 0, "invalid tuple count %s", tupleCount ); this.tupleCountToCheck = tupleCount; this.tupleCountToPoll = tupleAvailabilityByCount == EXACT ? tupleCount : max( tupleCount, maxBatchSize ); } }
SinglePortDrainer implements TupleQueueDrainer { public final void setParameters ( final TupleAvailabilityByCount tupleAvailabilityByCount, final int tupleCount ) { checkArgument( tupleAvailabilityByCount != null ); checkArgument( tupleCount > 0, "invalid tuple count %s", tupleCount ); this.tupleCountToCheck = tupleCount; this.tupleCountToPoll = tupleAvailabilityByCount == EXACT ? tupleCount : max( tupleCount, maxBatchSize ); } SinglePortDrainer( final int maxBatchSize ); }
SinglePortDrainer implements TupleQueueDrainer { public final void setParameters ( final TupleAvailabilityByCount tupleAvailabilityByCount, final int tupleCount ) { checkArgument( tupleAvailabilityByCount != null ); checkArgument( tupleCount > 0, "invalid tuple count %s", tupleCount ); this.tupleCountToCheck = tupleCount; this.tupleCountToPoll = tupleAvailabilityByCount == EXACT ? tupleCount : max( tupleCount, maxBatchSize ); } SinglePortDrainer( final int maxBatchSize ); final void setParameters( final TupleAvailabilityByCount tupleAvailabilityByCount, final int tupleCount ); }
SinglePortDrainer implements TupleQueueDrainer { public final void setParameters ( final TupleAvailabilityByCount tupleAvailabilityByCount, final int tupleCount ) { checkArgument( tupleAvailabilityByCount != null ); checkArgument( tupleCount > 0, "invalid tuple count %s", tupleCount ); this.tupleCountToCheck = tupleCount; this.tupleCountToPoll = tupleAvailabilityByCount == EXACT ? tupleCount : max( tupleCount, maxBatchSize ); } SinglePortDrainer( final int maxBatchSize ); final void setParameters( final TupleAvailabilityByCount tupleAvailabilityByCount, final int tupleCount );  }