@Test @DisplayName("deduplicate and deliver all matching `CATCH_UP` messages " + "if the respective `CatchUpJob` is completed, " + "keeping the `CATCH_UP` messages in their storage for a bit longer") void deduplicateAndDeliverWhenJobCompleted() { InboxMessage toCatchUp = catchingUp(targetOne, type); InboxMessage moreToCatchUp = catchingUp(targetOne, type); InboxMessage toDeliver = toDeliver(targetOne, type); InboxMessage duplicateToCatchUp = copyWithNewId(toCatchUp); InboxMessage duplicateToDeliver = copyWithStatus(copyWithNewId(toCatchUp), TO_DELIVER); Conveyor conveyor = new Conveyor( ImmutableList.of(toCatchUp, moreToCatchUp, toDeliver, duplicateToCatchUp, duplicateToDeliver), new DeliveredMessages() ); CatchUp job = catchUpJob(type, COMPLETED, currentTime(), ImmutableList.of(targetOne)); MemoizingAction action = MemoizingAction.empty(); CatchUpStation station = new CatchUpStation(action, ImmutableList.of(job)); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 2); Collection<InboxMessage> deliveredMessages = checkNotNull(action.passedMessages()); assertContainsExactly(deliveredMessages, toCatchUp, moreToCatchUp ); Map<InboxMessageId, InboxMessage> remaindersById = stream(conveyor.iterator()).collect(toMap(InboxMessage::getId, message -> message)); assertThat(remaindersById).hasSize(3); assertKeptForLonger(toCatchUp.getId(), remaindersById); assertKeptForLonger(moreToCatchUp.getId(), remaindersById); assertNotKeptForLonger(toDeliver.getId(), remaindersById); }
@Override public final Result process(Conveyor conveyor) { JobFilter jobFilter = new JobFilter(jobs, conveyor); Collection<InboxMessage> toDispatch = jobFilter.messagesToDispatch(); return dispatch(toDispatch, conveyor); }
CatchUpStation extends Station { @Override public final Result process(Conveyor conveyor) { JobFilter jobFilter = new JobFilter(jobs, conveyor); Collection<InboxMessage> toDispatch = jobFilter.messagesToDispatch(); return dispatch(toDispatch, conveyor); } }
CatchUpStation extends Station { @Override public final Result process(Conveyor conveyor) { JobFilter jobFilter = new JobFilter(jobs, conveyor); Collection<InboxMessage> toDispatch = jobFilter.messagesToDispatch(); return dispatch(toDispatch, conveyor); } CatchUpStation(DeliveryAction action, Iterable<CatchUp> jobs); }
CatchUpStation extends Station { @Override public final Result process(Conveyor conveyor) { JobFilter jobFilter = new JobFilter(jobs, conveyor); Collection<InboxMessage> toDispatch = jobFilter.messagesToDispatch(); return dispatch(toDispatch, conveyor); } CatchUpStation(DeliveryAction action, Iterable<CatchUp> jobs); @Override final Result process(Conveyor conveyor); }
CatchUpStation extends Station { @Override public final Result process(Conveyor conveyor) { JobFilter jobFilter = new JobFilter(jobs, conveyor); Collection<InboxMessage> toDispatch = jobFilter.messagesToDispatch(); return dispatch(toDispatch, conveyor); } CatchUpStation(DeliveryAction action, Iterable<CatchUp> jobs); @Override final Result process(Conveyor conveyor);  }