RateLimitedBatcher implements Observable.Operator<Batch<T, I>, T> { public static <T extends Batchable<?>, I> RateLimitedBatcher<T, I> create(TokenBucket tokenBucket, long initialDelay, long maxDelay, IndexExtractor<T, I> indexExtractor, EmissionStrategy emissionStrategy, String metricsRoot, Registry registry, Scheduler scheduler) { return new RateLimitedBatcher<T, I>(tokenBucket, initialDelay, maxDelay, indexExtractor, emissionStrategy, metricsRoot, registry, scheduler); } private  RateLimitedBatcher(TokenBucket tokenBucket,
                               long initialDelay,
                               long maxDelay,
                               IndexExtractor<T, I> indexExtractor,
                               EmissionStrategy emissionStrategy,
                               String metricsRoot,
                               Registry registry,
                               Scheduler scheduler); static RateLimitedBatcher<T, I> create(TokenBucket tokenBucket,
                                                                              long initialDelay,
                                                                              long maxDelay,
                                                                              IndexExtractor<T, I> indexExtractor,
                                                                              EmissionStrategy emissionStrategy,
                                                                              String metricsRoot,
                                                                              Registry registry,
                                                                              Scheduler scheduler); @Override Subscriber<? super T> call(final Subscriber<? super Batch<T, I>> downstream);  }
@Test public void onCompletedIsNotSentAfterOnError() { final RateLimitedBatcher<BatchableOperationMock, String> batcher = buildBatcher(minimumTimeInQueueMs); final Subject<BatchableOperationMock, BatchableOperationMock> updates = PublishSubject.<BatchableOperationMock>create().toSerialized(); final AssertableSubscriber<Batch<BatchableOperationMock, String>> subscriber = updates.lift(batcher).test(); testScheduler.triggerActions(); subscriber.assertNoTerminalEvent().assertNoValues(); updates.onError(new RuntimeException("some problem")); testScheduler.triggerActions(); subscriber.assertNotCompleted().assertError(RuntimeException.class); updates.onCompleted(); subscriber.assertNotCompleted(); }
@Test public void onErrorIsNotSentAfterOnCompleted() { final RateLimitedBatcher<BatchableOperationMock, String> batcher = buildBatcher(minimumTimeInQueueMs); final Subject<BatchableOperationMock, BatchableOperationMock> updates = PublishSubject.<BatchableOperationMock>create().toSerialized(); final AssertableSubscriber<Batch<BatchableOperationMock, String>> subscriber = updates.lift(batcher).test(); testScheduler.triggerActions(); subscriber.assertNoTerminalEvent().assertNoValues(); updates.onCompleted(); testScheduler.advanceTimeBy(minimumTimeInQueueMs, TimeUnit.MILLISECONDS); subscriber.assertNoErrors().assertCompleted(); updates.onError(new RuntimeException("some problem")); subscriber.assertNoErrors(); }
@Test public void ignoreErrorsFromDownstream() { final Instant now = Instant.ofEpochMilli(testScheduler.now()); final RateLimitedBatcher<BatchableOperationMock, String> batcher = buildBatcher(minimumTimeInQueueMs); final Subject<BatchableOperationMock, BatchableOperationMock> updates = PublishSubject.<BatchableOperationMock>create().toSerialized(); final AssertableSubscriber<?> subscriber = updates.lift(batcher) .lift(new ExceptionThrowingOperator("some error happened")) .test(); testScheduler.triggerActions(); subscriber.assertNoTerminalEvent().assertNoValues(); for (int i = 0; i < 10; i++) { updates.onNext(new BatchableOperationMock(LOW, now, "resource2", "sub2", "create")); testScheduler.advanceTimeBy(minimumTimeInQueueMs, TimeUnit.MILLISECONDS); subscriber.assertNoTerminalEvent().assertNoValues(); } updates.onCompleted(); testScheduler.advanceTimeBy(minimumTimeInQueueMs, TimeUnit.MILLISECONDS); subscriber.assertNoValues().assertCompleted(); }