@Test public void testConsumer() { doReturn(new ConsumerResult<>()).when(mockConsumer).consume(any(), any()); doAnswer(new Answer() { int producerCalls = 0; @Override public Object answer(InvocationOnMock invocationOnMock) throws Throwable { producerCalls++; return producerCalls == 1 ? createProducerResult() : new ProducerResult(); } }).when(mockProducer).produce(any()); SynchronousProducerConsumerProcessor spyProcessor = spy(processor); spyProcessor.execute(); verify(mockConsumer, times(1)).consume(any(), any()); }
public void execute() { final Instant endProcessingTime = Instant.now().plusSeconds(runMaxSeconds); itemsProcessedCount = 0; producer.reset(); while(true) { ProducerResult<T> producerResult = producer.produce(endProcessingTime); if(producerResult.isInterrupted() || producerResult.getItemsProduced() == null || producerResult.getItemsProduced().size() == 0 || InstantUtil.isNowAfterOrEqual(endProcessingTime)) { break; } ConsumerResult<T> consumerResult = consumer.consume(producerResult.getItemsProduced(), endProcessingTime); itemsProcessedCount += consumerResult.getItemsConsumedCount(); if(consumerResult.isInterrupted()) { break; } } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) { processingTimeExpired(); } }
SynchronousProducerConsumerProcessor { public void execute() { final Instant endProcessingTime = Instant.now().plusSeconds(runMaxSeconds); itemsProcessedCount = 0; producer.reset(); while(true) { ProducerResult<T> producerResult = producer.produce(endProcessingTime); if(producerResult.isInterrupted() || producerResult.getItemsProduced() == null || producerResult.getItemsProduced().size() == 0 || InstantUtil.isNowAfterOrEqual(endProcessingTime)) { break; } ConsumerResult<T> consumerResult = consumer.consume(producerResult.getItemsProduced(), endProcessingTime); itemsProcessedCount += consumerResult.getItemsConsumedCount(); if(consumerResult.isInterrupted()) { break; } } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) { processingTimeExpired(); } } }
SynchronousProducerConsumerProcessor { public void execute() { final Instant endProcessingTime = Instant.now().plusSeconds(runMaxSeconds); itemsProcessedCount = 0; producer.reset(); while(true) { ProducerResult<T> producerResult = producer.produce(endProcessingTime); if(producerResult.isInterrupted() || producerResult.getItemsProduced() == null || producerResult.getItemsProduced().size() == 0 || InstantUtil.isNowAfterOrEqual(endProcessingTime)) { break; } ConsumerResult<T> consumerResult = consumer.consume(producerResult.getItemsProduced(), endProcessingTime); itemsProcessedCount += consumerResult.getItemsConsumedCount(); if(consumerResult.isInterrupted()) { break; } } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) { processingTimeExpired(); } } SynchronousProducerConsumerProcessor(Producer<T> producer, Consumer<T> consumer); }
SynchronousProducerConsumerProcessor { public void execute() { final Instant endProcessingTime = Instant.now().plusSeconds(runMaxSeconds); itemsProcessedCount = 0; producer.reset(); while(true) { ProducerResult<T> producerResult = producer.produce(endProcessingTime); if(producerResult.isInterrupted() || producerResult.getItemsProduced() == null || producerResult.getItemsProduced().size() == 0 || InstantUtil.isNowAfterOrEqual(endProcessingTime)) { break; } ConsumerResult<T> consumerResult = consumer.consume(producerResult.getItemsProduced(), endProcessingTime); itemsProcessedCount += consumerResult.getItemsConsumedCount(); if(consumerResult.isInterrupted()) { break; } } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) { processingTimeExpired(); } } SynchronousProducerConsumerProcessor(Producer<T> producer, Consumer<T> consumer); void execute(); int getItemsProcessedCount(); int getRunMaxSeconds(); SynchronousProducerConsumerProcessor<T> setRunMaxSeconds(int runMaxSeconds); }
SynchronousProducerConsumerProcessor { public void execute() { final Instant endProcessingTime = Instant.now().plusSeconds(runMaxSeconds); itemsProcessedCount = 0; producer.reset(); while(true) { ProducerResult<T> producerResult = producer.produce(endProcessingTime); if(producerResult.isInterrupted() || producerResult.getItemsProduced() == null || producerResult.getItemsProduced().size() == 0 || InstantUtil.isNowAfterOrEqual(endProcessingTime)) { break; } ConsumerResult<T> consumerResult = consumer.consume(producerResult.getItemsProduced(), endProcessingTime); itemsProcessedCount += consumerResult.getItemsConsumedCount(); if(consumerResult.isInterrupted()) { break; } } if(InstantUtil.isNowAfterOrEqual(endProcessingTime)) { processingTimeExpired(); } } SynchronousProducerConsumerProcessor(Producer<T> producer, Consumer<T> consumer); void execute(); int getItemsProcessedCount(); int getRunMaxSeconds(); SynchronousProducerConsumerProcessor<T> setRunMaxSeconds(int runMaxSeconds); static final int DEFAULT_RUN_MAX_SECONDS; }