public void consume(int eventCount,long timeoutInMillis,Predicate<Event> condition) throws TimeoutException {
  if (eventCount < 0)   throw new IllegalArgumentException("The eventCount may not be negative");
  if (eventCount == 0)   return;
  int eventsRemaining=eventCount;
  final long stopTime=System.currentTimeMillis() + timeoutInMillis;
  while (eventsRemaining > 0 && System.currentTimeMillis() < stopTime) {
    Event nextEvent=queue.poll();
    if (nextEvent != null) {
      if (condition.test(nextEvent)) {
        --eventsRemaining;
        consumedEvents.accept(nextEvent);
      }
 else {
        ignoredEvents.accept(nextEvent);
      }
    }
  }
  if (eventsRemaining > 0) {
    throw new TimeoutException("Received " + (eventCount - eventsRemaining) + " of "+ eventCount+ " in "+ timeoutInMillis+ "ms");
  }
}

public void consume ( int eventCount , long timeoutInMillis , Predicate < Event > condition ) throws TimeoutException { if ( eventCount < 0 ) throw new IllegalArgumentException ( "The eventCount may not be negative" ) ; if ( eventCount == 0 ) return ; int eventsRemaining = eventCount ; final long stopTime = System . currentTimeMillis ( ) + timeoutInMillis ; while ( eventsRemaining > 0 && System . currentTimeMillis ( ) < stopTime ) { Event nextEvent = queue . poll ( ) ; if ( nextEvent != null ) { if ( condition . test ( nextEvent ) ) { -- eventsRemaining ; consumedEvents . accept ( nextEvent ) ; } else { ignoredEvents . accept ( nextEvent ) ; } } } if ( eventsRemaining > 0 ) { throw new TimeoutException ( "Received " + ( eventCount - eventsRemaining ) + " of " + eventCount + " in " + timeoutInMillis + "ms" ) ; } }
blocks until the listener has consume the specified number of matching events , blocking at most the specified number of milliseconds . if this method has not reached the number of matching events and comes across events that do not satisfy the predicate , those events are consumed and ignored .
