@Test(expected = IllegalStateException.class) public void testGetNextRecordsWithoutStarting() { verify(executorService, times(0)).execute(any()); getRecordsCache.drainQueueForRequests(); }
@VisibleForTesting synchronized void drainQueueForRequests() { final PrefetchRecordsRetrieved recordsToDeliver = peekNextResult(); if (publisherSession.hasDemandToPublish() && canDispatchRecord(recordsToDeliver)) { subscriber.onNext(recordsToDeliver.prepareForPublish()); recordsToDeliver.dispatched(); lastEventDeliveryTime = Instant.now(); } }
PrefetchRecordsPublisher implements RecordsPublisher { @VisibleForTesting synchronized void drainQueueForRequests() { final PrefetchRecordsRetrieved recordsToDeliver = peekNextResult(); if (publisherSession.hasDemandToPublish() && canDispatchRecord(recordsToDeliver)) { subscriber.onNext(recordsToDeliver.prepareForPublish()); recordsToDeliver.dispatched(); lastEventDeliveryTime = Instant.now(); } } }
PrefetchRecordsPublisher implements RecordsPublisher { @VisibleForTesting synchronized void drainQueueForRequests() { final PrefetchRecordsRetrieved recordsToDeliver = peekNextResult(); if (publisherSession.hasDemandToPublish() && canDispatchRecord(recordsToDeliver)) { subscriber.onNext(recordsToDeliver.prepareForPublish()); recordsToDeliver.dispatched(); lastEventDeliveryTime = Instant.now(); } } PrefetchRecordsPublisher(final int maxPendingProcessRecordsInput, final int maxByteSize, final int maxRecordsCount,
                                    final int maxRecordsPerCall,
                                    @NonNull final GetRecordsRetrievalStrategy getRecordsRetrievalStrategy,
                                    @NonNull final ExecutorService executorService,
                                    final long idleMillisBetweenCalls,
                                    @NonNull final MetricsFactory metricsFactory,
                                    @NonNull final String operation,
                                    @NonNull final String shardId); }
PrefetchRecordsPublisher implements RecordsPublisher { @VisibleForTesting synchronized void drainQueueForRequests() { final PrefetchRecordsRetrieved recordsToDeliver = peekNextResult(); if (publisherSession.hasDemandToPublish() && canDispatchRecord(recordsToDeliver)) { subscriber.onNext(recordsToDeliver.prepareForPublish()); recordsToDeliver.dispatched(); lastEventDeliveryTime = Instant.now(); } } PrefetchRecordsPublisher(final int maxPendingProcessRecordsInput, final int maxByteSize, final int maxRecordsCount,
                                    final int maxRecordsPerCall,
                                    @NonNull final GetRecordsRetrievalStrategy getRecordsRetrievalStrategy,
                                    @NonNull final ExecutorService executorService,
                                    final long idleMillisBetweenCalls,
                                    @NonNull final MetricsFactory metricsFactory,
                                    @NonNull final String operation,
                                    @NonNull final String shardId); @Override void start(ExtendedSequenceNumber extendedSequenceNumber, InitialPositionInStreamExtended initialPositionInStreamExtended); @Override void shutdown(); @Override RequestDetails getLastSuccessfulRequestDetails(); @Override void restartFrom(RecordsRetrieved recordsRetrieved); @Override void subscribe(Subscriber<? super RecordsRetrieved> s); @Override synchronized void notify(RecordsDeliveryAck recordsDeliveryAck); }
PrefetchRecordsPublisher implements RecordsPublisher { @VisibleForTesting synchronized void drainQueueForRequests() { final PrefetchRecordsRetrieved recordsToDeliver = peekNextResult(); if (publisherSession.hasDemandToPublish() && canDispatchRecord(recordsToDeliver)) { subscriber.onNext(recordsToDeliver.prepareForPublish()); recordsToDeliver.dispatched(); lastEventDeliveryTime = Instant.now(); } } PrefetchRecordsPublisher(final int maxPendingProcessRecordsInput, final int maxByteSize, final int maxRecordsCount,
                                    final int maxRecordsPerCall,
                                    @NonNull final GetRecordsRetrievalStrategy getRecordsRetrievalStrategy,
                                    @NonNull final ExecutorService executorService,
                                    final long idleMillisBetweenCalls,
                                    @NonNull final MetricsFactory metricsFactory,
                                    @NonNull final String operation,
                                    @NonNull final String shardId); @Override void start(ExtendedSequenceNumber extendedSequenceNumber, InitialPositionInStreamExtended initialPositionInStreamExtended); @Override void shutdown(); @Override RequestDetails getLastSuccessfulRequestDetails(); @Override void restartFrom(RecordsRetrieved recordsRetrieved); @Override void subscribe(Subscriber<? super RecordsRetrieved> s); @Override synchronized void notify(RecordsDeliveryAck recordsDeliveryAck);  }