@Test public void onErrorStopsProcessingTest() throws Exception { Throwable expected = new Throwable("Wheee"); addItemsToReturn(10); recordsPublisher.add(new ResponseItem(expected)); addItemsToReturn(10); setupNotifierAnswer(10); synchronized (processedNotifier) { subscriber.startSubscriptions(); processedNotifier.wait(5000); } for (int attempts = 0; attempts < 10; attempts++) { if (subscriber.retrievalFailure() != null) { break; } Thread.sleep(10); } verify(shardConsumer, times(10)).handleInput(argThat(eqProcessRecordsInput(processRecordsInput)), any(Subscription.class)); assertThat(subscriber.retrievalFailure(), equalTo(expected)); }
void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } }
ShardConsumerSubscriber implements Subscriber<RecordsRetrieved> { void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } } }
ShardConsumerSubscriber implements Subscriber<RecordsRetrieved> { void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } } @Deprecated  ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
                            ShardConsumer shardConsumer);   ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
            ShardConsumer shardConsumer, int readTimeoutsToIgnoreBeforeWarning); }
ShardConsumerSubscriber implements Subscriber<RecordsRetrieved> { void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } } @Deprecated  ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
                            ShardConsumer shardConsumer);   ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
            ShardConsumer shardConsumer, int readTimeoutsToIgnoreBeforeWarning); @Override void onSubscribe(Subscription s); @Override void onNext(RecordsRetrieved input); @Override void onError(Throwable t); @Override void onComplete(); void cancel(); }
ShardConsumerSubscriber implements Subscriber<RecordsRetrieved> { void startSubscriptions() { synchronized (lockObject) { lastRequestTime = Instant.now(); if (lastAccepted != null) { recordsPublisher.restartFrom(lastAccepted); } Flowable.fromPublisher(recordsPublisher).subscribeOn(scheduler).observeOn(scheduler, true, bufferSize) .subscribe(new ShardConsumerNotifyingSubscriber(this, recordsPublisher)); } } @Deprecated  ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
                            ShardConsumer shardConsumer);   ShardConsumerSubscriber(RecordsPublisher recordsPublisher, ExecutorService executorService, int bufferSize,
            ShardConsumer shardConsumer, int readTimeoutsToIgnoreBeforeWarning); @Override void onSubscribe(Subscription s); @Override void onNext(RecordsRetrieved input); @Override void onError(Throwable t); @Override void onComplete(); void cancel();  }