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);  }
@Test @DisplayName("remove live messages which correspond to a started `CatchUpJob`") void removeLiveMessagesIfCatchUpStarted() { InboxMessage toDeliver = toDeliver(targetOne, type); InboxMessage anotherToDeliver = copyWithNewId(toDeliver); InboxMessage delivered = delivered(targetOne, type); InboxMessage differentTarget = delivered(targetTwo, type); Conveyor conveyor = new Conveyor( ImmutableList.of(toDeliver, anotherToDeliver, delivered, differentTarget), new DeliveredMessages() ); CatchUp job = catchUpJob(type, IN_PROGRESS, currentTime(), ImmutableList.of(targetOne)); CatchUpStation station = new CatchUpStation(MemoizingAction.empty(), ImmutableList.of(job)); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 0); assertContainsExactly(conveyor.removals(), toDeliver, anotherToDeliver); assertContainsExactly(conveyor.iterator(), delivered, differentTarget); }
@Test @DisplayName("deliver the messages in `CATCH_UP` status " + "which correspond to a started `CatchUpJob` and remove duplicates") void matchAndRunDeliveryAction() { InboxMessage toCatchUp = catchingUp(targetOne, type); InboxMessage duplicateCopy = copyWithNewId(toCatchUp); InboxMessage anotherToCatchUp = catchingUp(targetOne, type); InboxMessage alreadyDelivered = delivered(targetOne, type); InboxMessage differentTarget = catchingUp(targetTwo, type); ImmutableList<InboxMessage> initialContents = ImmutableList.of(toCatchUp, anotherToCatchUp, duplicateCopy, alreadyDelivered, differentTarget); Conveyor conveyor = new Conveyor(initialContents, new DeliveredMessages()); CatchUp job = catchUpJob(type, IN_PROGRESS, currentTime(), ImmutableList.of(targetOne)); CatchUpStation station = new CatchUpStation(MemoizingAction.empty(), ImmutableList.of(job)); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 2); assertContainsExactly(conveyor.recentlyDelivered(), copyWithStatus(toCatchUp, DELIVERED), copyWithStatus(anotherToCatchUp, DELIVERED), alreadyDelivered); assertContainsExactly(conveyor.removals(), duplicateCopy); assertContainsExactly(conveyor.iterator(), copyWithStatus(toCatchUp, DELIVERED), copyWithStatus(anotherToCatchUp, DELIVERED), alreadyDelivered, differentTarget); }
@Test @DisplayName("mark as `CATCH_UP` those live messages " + "which correspond to a finalizing `CatchUpJob`") void markLiveMessagesCatchUpIfJobFinalizing() { InboxMessage toDeliver = toDeliver(targetOne, type); InboxMessage anotherToDeliver = copyWithNewId(toDeliver); InboxMessage delivered = delivered(targetOne, type); InboxMessage differentTarget = delivered(targetTwo, type); Conveyor conveyor = new Conveyor( ImmutableList.of(toDeliver, anotherToDeliver, delivered, differentTarget), new DeliveredMessages() ); CatchUp job = catchUpJob(type, FINALIZING, currentTime(), ImmutableList.of(targetOne)); CatchUpStation station = new CatchUpStation(MemoizingAction.empty(), ImmutableList.of(job)); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 0); assertContainsExactly(conveyor.iterator(), copyWithStatus(toDeliver, TO_CATCH_UP), copyWithStatus(anotherToDeliver, TO_CATCH_UP), delivered, differentTarget); }
@Test @DisplayName("deduplicate and deliver all matching `CATCH_UP` messages " + "if the respective `CatchUpJob` is completed, " + "keeping the `CATCH_UP` messages in their storage for a bit longer") void deduplicateAndDeliverWhenJobCompleted() { InboxMessage toCatchUp = catchingUp(targetOne, type); InboxMessage moreToCatchUp = catchingUp(targetOne, type); InboxMessage toDeliver = toDeliver(targetOne, type); InboxMessage duplicateToCatchUp = copyWithNewId(toCatchUp); InboxMessage duplicateToDeliver = copyWithStatus(copyWithNewId(toCatchUp), TO_DELIVER); Conveyor conveyor = new Conveyor( ImmutableList.of(toCatchUp, moreToCatchUp, toDeliver, duplicateToCatchUp, duplicateToDeliver), new DeliveredMessages() ); CatchUp job = catchUpJob(type, COMPLETED, currentTime(), ImmutableList.of(targetOne)); MemoizingAction action = MemoizingAction.empty(); CatchUpStation station = new CatchUpStation(action, ImmutableList.of(job)); Station.Result result = station.process(conveyor); assertDeliveredCount(result, 2); Collection<InboxMessage> deliveredMessages = checkNotNull(action.passedMessages()); assertContainsExactly(deliveredMessages, toCatchUp, moreToCatchUp ); Map<InboxMessageId, InboxMessage> remaindersById = stream(conveyor.iterator()).collect(toMap(InboxMessage::getId, message -> message)); assertThat(remaindersById).hasSize(3); assertKeptForLonger(toCatchUp.getId(), remaindersById); assertKeptForLonger(moreToCatchUp.getId(), remaindersById); assertNotKeptForLonger(toDeliver.getId(), remaindersById); }
@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) ); }