@Test @DisplayName("deliver the messages in `CATCH_UP` status " + "which correspond to a started `CatchUpJob` and remove duplicates") void matchAndRunDeliveryAction() { InboxMessage toCatchUp = catchingUp(targetOne, type); InboxMessage duplicateCopy = copyWithNewId(toCatchUp); InboxMessage anotherToCatchUp = catchingUp(targetOne, type); InboxMessage alreadyDelivered = delivered(targetOne, type); InboxMessage differentTarget = catchingUp(targetTwo, type); ImmutableList<InboxMessage> initialContents = ImmutableList.of(toCatchUp, anotherToCatchUp, duplicateCopy, alreadyDelivered, differentTarget); Conveyor conveyor = new Conveyor(initialContents, new DeliveredMessages()); CatchUp job = catchUpJob(type, IN_PROGRESS, currentTime(), ImmutableList.of(targetOne)); CatchUpStation station = new CatchUpStation(MemoizingAction.empty(), ImmutableList.of(job)); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 2); assertContainsExactly(conveyor.recentlyDelivered(), copyWithStatus(toCatchUp, DELIVERED), copyWithStatus(anotherToCatchUp, DELIVERED), alreadyDelivered); assertContainsExactly(conveyor.removals(), duplicateCopy); assertContainsExactly(conveyor.iterator(), copyWithStatus(toCatchUp, DELIVERED), copyWithStatus(anotherToCatchUp, DELIVERED), alreadyDelivered, differentTarget); }
@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);  }