@Test @DisplayName("run the delivery action for all the messages in `TO_DELIVER` status") void runDeliveryActionForToDeliver() { InboxMessage toDeliver = toDeliver(targetOne, type); InboxMessage anotherToDeliver = toDeliver(targetOne, type); InboxMessage differentTarget = toDeliver(targetTwo, type); InboxMessage alreadyDelivered = delivered(targetOne, type); InboxMessage toCatchUp = catchingUp(targetOne, type); ImmutableList<InboxMessage> initialContents = ImmutableList.of(toDeliver, anotherToDeliver, differentTarget, alreadyDelivered, toCatchUp); Conveyor conveyor = new Conveyor(initialContents, new DeliveredMessages()); MemoizingAction action = MemoizingAction.empty(); Station station = new LiveDeliveryStation(action, noWindow()); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 3); List<InboxMessage> passedToAction = checkNotNull(action.passedMessages()); assertContainsExactly(passedToAction, toDeliver, anotherToDeliver, differentTarget); assertContainsExactly(conveyor.iterator(), copyWithStatus(toDeliver, DELIVERED), copyWithStatus(anotherToDeliver, DELIVERED), copyWithStatus(differentTarget, DELIVERED), alreadyDelivered, toCatchUp); }
@Override public final Result process(Conveyor conveyor) { FilterToDeliver filter = new FilterToDeliver(conveyor); Collection<InboxMessage> filtered = filter.messagesToDispatch(); if (filtered.isEmpty()) { return emptyResult(); } List<InboxMessage> toDispatch = deduplicateAndSort(filtered, conveyor); DeliveryErrors errors = action.executeFor(toDispatch); conveyor.markDelivered(toDispatch); Result result = new Result(toDispatch.size(), errors); return result; }
LiveDeliveryStation extends Station { @Override public final Result process(Conveyor conveyor) { FilterToDeliver filter = new FilterToDeliver(conveyor); Collection<InboxMessage> filtered = filter.messagesToDispatch(); if (filtered.isEmpty()) { return emptyResult(); } List<InboxMessage> toDispatch = deduplicateAndSort(filtered, conveyor); DeliveryErrors errors = action.executeFor(toDispatch); conveyor.markDelivered(toDispatch); Result result = new Result(toDispatch.size(), errors); return result; } }
LiveDeliveryStation extends Station { @Override public final Result process(Conveyor conveyor) { FilterToDeliver filter = new FilterToDeliver(conveyor); Collection<InboxMessage> filtered = filter.messagesToDispatch(); if (filtered.isEmpty()) { return emptyResult(); } List<InboxMessage> toDispatch = deduplicateAndSort(filtered, conveyor); DeliveryErrors errors = action.executeFor(toDispatch); conveyor.markDelivered(toDispatch); Result result = new Result(toDispatch.size(), errors); return result; } LiveDeliveryStation(DeliveryAction action, Duration deduplicationWindow); }
LiveDeliveryStation extends Station { @Override public final Result process(Conveyor conveyor) { FilterToDeliver filter = new FilterToDeliver(conveyor); Collection<InboxMessage> filtered = filter.messagesToDispatch(); if (filtered.isEmpty()) { return emptyResult(); } List<InboxMessage> toDispatch = deduplicateAndSort(filtered, conveyor); DeliveryErrors errors = action.executeFor(toDispatch); conveyor.markDelivered(toDispatch); Result result = new Result(toDispatch.size(), errors); return result; } LiveDeliveryStation(DeliveryAction action, Duration deduplicationWindow); @Override final Result process(Conveyor conveyor); }
LiveDeliveryStation extends Station { @Override public final Result process(Conveyor conveyor) { FilterToDeliver filter = new FilterToDeliver(conveyor); Collection<InboxMessage> filtered = filter.messagesToDispatch(); if (filtered.isEmpty()) { return emptyResult(); } List<InboxMessage> toDispatch = deduplicateAndSort(filtered, conveyor); DeliveryErrors errors = action.executeFor(toDispatch); conveyor.markDelivered(toDispatch); Result result = new Result(toDispatch.size(), errors); return result; } LiveDeliveryStation(DeliveryAction action, Duration deduplicationWindow); @Override final Result process(Conveyor conveyor);  }