@Test public void testConsume() { final int ITEMS_TO_CONSUME = 50; ConsumerResult consumerResult = consumer.consume( IntStream.range(0, ITEMS_TO_CONSUME).mapToObj(i -> Integer.toString(i)).collect(Collectors.toList()), Instant.now().plusSeconds(60)); Assert.assertEquals(consumerResult.getItemsConsumedCount(), ITEMS_TO_CONSUME); verify(consumer, times(ITEMS_TO_CONSUME)).updateAgendaProgress(any()); }
@Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingInstant) { ConsumerResult<String> consumerResult = new ConsumerResult<>(); if(collection == null) return consumerResult; int reclaimCount = 0; for(String agendaProgressId : collection) { if(updateAgendaProgress(agendaProgressId)) reclaimCount++; if(InstantUtil.isNowAfterOrEqual(endProcessingInstant)) break; } return consumerResult.setItemsConsumedCount(reclaimCount); }
AgendaProgressTimeoutConsumer implements Consumer<String> { @Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingInstant) { ConsumerResult<String> consumerResult = new ConsumerResult<>(); if(collection == null) return consumerResult; int reclaimCount = 0; for(String agendaProgressId : collection) { if(updateAgendaProgress(agendaProgressId)) reclaimCount++; if(InstantUtil.isNowAfterOrEqual(endProcessingInstant)) break; } return consumerResult.setItemsConsumedCount(reclaimCount); } }
AgendaProgressTimeoutConsumer implements Consumer<String> { @Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingInstant) { ConsumerResult<String> consumerResult = new ConsumerResult<>(); if(collection == null) return consumerResult; int reclaimCount = 0; for(String agendaProgressId : collection) { if(updateAgendaProgress(agendaProgressId)) reclaimCount++; if(InstantUtil.isNowAfterOrEqual(endProcessingInstant)) break; } return consumerResult.setItemsConsumedCount(reclaimCount); } AgendaProgressTimeoutConsumer(HttpObjectClient<AgendaProgress> agendaProgressClient); }
AgendaProgressTimeoutConsumer implements Consumer<String> { @Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingInstant) { ConsumerResult<String> consumerResult = new ConsumerResult<>(); if(collection == null) return consumerResult; int reclaimCount = 0; for(String agendaProgressId : collection) { if(updateAgendaProgress(agendaProgressId)) reclaimCount++; if(InstantUtil.isNowAfterOrEqual(endProcessingInstant)) break; } return consumerResult.setItemsConsumedCount(reclaimCount); } AgendaProgressTimeoutConsumer(HttpObjectClient<AgendaProgress> agendaProgressClient); @Override ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingInstant); boolean getLogReclaimOnly(); AgendaProgressTimeoutConsumer setLogReclaimOnly(boolean logReclaimOnly); }
AgendaProgressTimeoutConsumer implements Consumer<String> { @Override public ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingInstant) { ConsumerResult<String> consumerResult = new ConsumerResult<>(); if(collection == null) return consumerResult; int reclaimCount = 0; for(String agendaProgressId : collection) { if(updateAgendaProgress(agendaProgressId)) reclaimCount++; if(InstantUtil.isNowAfterOrEqual(endProcessingInstant)) break; } return consumerResult.setItemsConsumedCount(reclaimCount); } AgendaProgressTimeoutConsumer(HttpObjectClient<AgendaProgress> agendaProgressClient); @Override ConsumerResult<String> consume(Collection<String> collection, Instant endProcessingInstant); boolean getLogReclaimOnly(); AgendaProgressTimeoutConsumer setLogReclaimOnly(boolean logReclaimOnly);  }