@Test public void blacklist() throws InterruptedException { eventsLeft.set(50); AtomicInteger messagesReceived = new AtomicInteger(); AtomicBoolean completed = new AtomicBoolean(); AtomicBoolean failed = new AtomicBoolean(); TrackingEventProcessorManager.EventTracker tracker = testSubject.createEventTracker(100L, "", true, new StreamObserver<InputStream>() { @Override public void onNext( InputStream value) { messagesReceived.incrementAndGet(); } @Override public void onError(Throwable t) { failed.set(true); } @Override public void onCompleted() { completed.set(true); } }); tracker.addPermits(50); tracker.addBlacklist(Collections.singletonList(PayloadDescription.newBuilder() .setType("DemoType") .setRevision("1.0") .build())); tracker.start(); assertWithin(1, TimeUnit.SECONDS, () -> assertEquals(10, messagesReceived.get())); }
EventTracker createEventTracker(long trackingToken, String clientId, boolean forceReadingFromLeader, StreamObserver<InputStream> eventStream) { return new EventTracker(trackingToken, clientId, forceReadingFromLeader, eventStream); }
TrackingEventProcessorManager { EventTracker createEventTracker(long trackingToken, String clientId, boolean forceReadingFromLeader, StreamObserver<InputStream> eventStream) { return new EventTracker(trackingToken, clientId, forceReadingFromLeader, eventStream); } }
TrackingEventProcessorManager { EventTracker createEventTracker(long trackingToken, String clientId, boolean forceReadingFromLeader, StreamObserver<InputStream> eventStream) { return new EventTracker(trackingToken, clientId, forceReadingFromLeader, eventStream); } TrackingEventProcessorManager(EventStorageEngine eventStorageEngine, int blacklistedSendAfter);   TrackingEventProcessorManager(String context, Function<Long, CloseableIterator<SerializedEventWithToken>> iteratorBuilder, int blacklistedSendAfter); }
TrackingEventProcessorManager { EventTracker createEventTracker(long trackingToken, String clientId, boolean forceReadingFromLeader, StreamObserver<InputStream> eventStream) { return new EventTracker(trackingToken, clientId, forceReadingFromLeader, eventStream); } TrackingEventProcessorManager(EventStorageEngine eventStorageEngine, int blacklistedSendAfter);   TrackingEventProcessorManager(String context, Function<Long, CloseableIterator<SerializedEventWithToken>> iteratorBuilder, int blacklistedSendAfter); void reschedule(); void stopAll(); void stopAllWhereNotAllowedReadingFromFollower(); void validateActiveConnections(long minLastPermits); void close(); }
TrackingEventProcessorManager { EventTracker createEventTracker(long trackingToken, String clientId, boolean forceReadingFromLeader, StreamObserver<InputStream> eventStream) { return new EventTracker(trackingToken, clientId, forceReadingFromLeader, eventStream); } TrackingEventProcessorManager(EventStorageEngine eventStorageEngine, int blacklistedSendAfter);   TrackingEventProcessorManager(String context, Function<Long, CloseableIterator<SerializedEventWithToken>> iteratorBuilder, int blacklistedSendAfter); void reschedule(); void stopAll(); void stopAllWhereNotAllowedReadingFromFollower(); void validateActiveConnections(long minLastPermits); void close();  }