@Test @DisplayName("remove live messages which correspond to a started `CatchUpJob`") void removeLiveMessagesIfCatchUpStarted() { InboxMessage toDeliver = toDeliver(targetOne, type); InboxMessage anotherToDeliver = copyWithNewId(toDeliver); InboxMessage delivered = delivered(targetOne, type); InboxMessage differentTarget = delivered(targetTwo, type); Conveyor conveyor = new Conveyor( ImmutableList.of(toDeliver, anotherToDeliver, delivered, differentTarget), 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, 0); assertContainsExactly(conveyor.removals(), toDeliver, anotherToDeliver); assertContainsExactly(conveyor.iterator(), delivered, 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);  }