@Test public void createEventTracker() throws InterruptedException { 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(5); tracker.start(); assertWithin(1, TimeUnit.SECONDS, () -> assertEquals(5, messagesReceived.get())); tracker.addPermits(10); assertWithin(1, TimeUnit.SECONDS, () -> assertEquals(10, messagesReceived.get())); eventsLeft.set(10); testSubject.reschedule(); assertWithin(1, TimeUnit.SECONDS, () -> assertEquals(15, messagesReceived.get())); tracker.stop(); assertTrue(completed.get()); assertWithin(10, TimeUnit.MILLISECONDS, () -> assertTrue(iteratorClosed.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();  }