@Test @DisplayName("run the delivery action " + "for the messages in `CATCH_UP` status sorting them beforehand") void sortCatchUpMessagesBeforeCallingToAction() { Timestamp now = currentTime(); Timestamp secondBefore = subtract(now, fromSeconds(1)); Timestamp twoSecondsBefore = subtract(now, fromSeconds(2)); Timestamp threeSecondsBefore = subtract(now, fromSeconds(3)); InboxMessage toCatchUp1 = catchingUp(targetOne, type, threeSecondsBefore); InboxMessage toCatchUp2 = catchingUp(targetOne, type, twoSecondsBefore); InboxMessage toCatchUp3 = catchingUp(targetOne, type, secondBefore); InboxMessage toCatchUp4 = catchingUp(targetOne, type, now); Conveyor conveyor = new Conveyor( ImmutableList.of(toCatchUp3, toCatchUp2, toCatchUp4, toCatchUp1), new DeliveredMessages() ); CatchUp job = catchUpJob(type, IN_PROGRESS, currentTime(), ImmutableList.of(targetOne)); MemoizingAction action = MemoizingAction.empty(); CatchUpStation station = new CatchUpStation(action, ImmutableList.of(job)); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 4); List<InboxMessage> deliveredMessages = checkNotNull(action.passedMessages()); assertThat(deliveredMessages).containsExactlyElementsIn( ImmutableList.of(toCatchUp1, toCatchUp2, toCatchUp3, toCatchUp4) ); }
@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);  }