@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(); }
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); }
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); } }
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); }
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); }
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);  }