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);  }
@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); }
@Test @DisplayName("remove duplicates prior to dispatching") void removeDuplicates() { InboxMessage toDeliver = toDeliver(targetOne, type); InboxMessage duplicate = copyWithNewId(toDeliver); InboxMessage anotherDuplicate = copyWithNewId(toDeliver); InboxMessage alreadyDelivered = delivered(targetOne, type); InboxMessage duplicateOfDelivered = copyWithNewId(copyWithStatus(alreadyDelivered, TO_DELIVER)); InboxMessage toCatchUp = catchingUp(targetOne, type); ImmutableList<InboxMessage> initialContents = ImmutableList.of(toDeliver, duplicate, anotherDuplicate, alreadyDelivered, duplicateOfDelivered, 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, 1); List<InboxMessage> passedToAction = checkNotNull(action.passedMessages()); assertContainsExactly(passedToAction, toDeliver); assertContainsExactly(conveyor.removals(), duplicate, anotherDuplicate, duplicateOfDelivered); assertContainsExactly(conveyor.iterator(), copyWithStatus(toDeliver, DELIVERED), alreadyDelivered, toCatchUp); }
@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) ); }
@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); }