### Question:
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; }### Answer:
@Test(dataProvider = "noDataProvider") public void testNoDataProduced(Collection collection) { doReturn(new ProducerResult().setItemsProduced(collection)).when(mockProducer).produce(any()); processor.execute(); verify(mockConsumer, times(0)).consume(any(), any()); }
@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()); }