@Test @DisplayName("sort messages prior to dispatching") void sort() { Timestamp now = currentTime(); Timestamp secondBefore = subtract(now, fromSeconds(1)); Timestamp twoSecondsBefore = subtract(now, fromSeconds(2)); Timestamp threeSecondsBefore = subtract(now, fromSeconds(3)); InboxMessage toDeliver1 = toDeliver(targetOne, type, threeSecondsBefore); InboxMessage toDeliver2 = toDeliver(targetTwo, type, twoSecondsBefore); InboxMessage toDeliver3 = toDeliver(targetOne, type, secondBefore); InboxMessage toDeliver4 = toDeliver(targetTwo, type, now); Conveyor conveyor = new Conveyor( ImmutableList.of(toDeliver2, toDeliver3, toDeliver4, toDeliver1), new DeliveredMessages() ); MemoizingAction action = MemoizingAction.empty(); Station station = new LiveDeliveryStation(action, noWindow()); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 4); List<InboxMessage> deliveredMessages = checkNotNull(action.passedMessages()); assertThat(deliveredMessages).containsExactlyElementsIn( ImmutableList.of(toDeliver1, toDeliver2, toDeliver3, toDeliver4) ); }
@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);  }