@Test @DisplayName("modify the messages and keep them for longer if the deduplication window is set") void keepMessagesForLongerIfDeduplicationWindowSet() { InboxMessage toDeliver = toDeliver(targetOne, type); InboxMessage differentTarget = toDeliver(targetTwo, type); InboxMessage alreadyDelivered = delivered(targetOne, type); InboxMessage toCatchUp = catchingUp(targetOne, type); ImmutableList<InboxMessage> initialContents = ImmutableList.of(toDeliver, differentTarget, alreadyDelivered, toCatchUp); Conveyor conveyor = new Conveyor(initialContents, new DeliveredMessages()); Station station = new LiveDeliveryStation(MemoizingAction.empty(), fromSeconds(100)); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 2); Map<InboxMessageId, InboxMessage> contentsById = stream(conveyor.iterator()).collect(toMap(InboxMessage::getId,msg -> msg)); assertKeptForLonger(toDeliver.getId(), contentsById); assertKeptForLonger(differentTarget.getId(), contentsById); assertNotKeptForLonger(alreadyDelivered.getId(), contentsById); assertNotKeptForLonger(toCatchUp.getId(), contentsById); }
@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);  }