@Test void testOrderingOfEvents() throws Exception { testSubject = new AsynchronousEventProcessingStrategy(Executors.newSingleThreadExecutor(), new SequentialPolicy()); final List<EventMessage> ackedMessages = Collections.synchronizedList(new ArrayList<>()); EventMessage<?> event1 = createEvent(1); EventMessage<?> event2 = createEvent(2); final Consumer<List<? extends EventMessage<?>>> processor = mock(Consumer.class); CountDownLatch latch = new CountDownLatch(2); doAnswer(invocation -> { List<? extends EventMessage<?>> events = (List) invocation.getArguments()[0]; events.forEach(e -> { ackedMessages.add(e); latch.countDown(); }); return null; }).when(processor).accept(anyList()); new DefaultUnitOfWork<>(null).execute(() -> { testSubject.handle(Collections.singletonList(event1), processor); testSubject.handle(Collections.singletonList(event2), processor); }); latch.await(); InOrder inOrder = Mockito.inOrder(processor, processor); inOrder.verify(processor).accept(Arrays.asList(event1, event2)); assertEquals(2, ackedMessages.size()); assertEquals(event1, ackedMessages.get(0)); assertEquals(event2, ackedMessages.get(1)); }
@Override public void handle(List<? extends EventMessage<?>> events, Consumer<List<? extends EventMessage<?>>> processor) { if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWorkRoot = CurrentUnitOfWork.get().root(); unitOfWorkRoot.getOrComputeResource(scheduledEventsKey, key -> { List<EventMessage<?>> allEvents = new ArrayList<>(); unitOfWorkRoot.afterCommit(uow -> schedule(allEvents, processor)); return allEvents; }).addAll(events); } else { schedule(events, processor); } }
AsynchronousEventProcessingStrategy implements EventProcessingStrategy { @Override public void handle(List<? extends EventMessage<?>> events, Consumer<List<? extends EventMessage<?>>> processor) { if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWorkRoot = CurrentUnitOfWork.get().root(); unitOfWorkRoot.getOrComputeResource(scheduledEventsKey, key -> { List<EventMessage<?>> allEvents = new ArrayList<>(); unitOfWorkRoot.afterCommit(uow -> schedule(allEvents, processor)); return allEvents; }).addAll(events); } else { schedule(events, processor); } } }
AsynchronousEventProcessingStrategy implements EventProcessingStrategy { @Override public void handle(List<? extends EventMessage<?>> events, Consumer<List<? extends EventMessage<?>>> processor) { if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWorkRoot = CurrentUnitOfWork.get().root(); unitOfWorkRoot.getOrComputeResource(scheduledEventsKey, key -> { List<EventMessage<?>> allEvents = new ArrayList<>(); unitOfWorkRoot.afterCommit(uow -> schedule(allEvents, processor)); return allEvents; }).addAll(events); } else { schedule(events, processor); } } AsynchronousEventProcessingStrategy(Executor executor,
                                               SequencingPolicy<? super EventMessage<?>> sequencingPolicy); }
AsynchronousEventProcessingStrategy implements EventProcessingStrategy { @Override public void handle(List<? extends EventMessage<?>> events, Consumer<List<? extends EventMessage<?>>> processor) { if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWorkRoot = CurrentUnitOfWork.get().root(); unitOfWorkRoot.getOrComputeResource(scheduledEventsKey, key -> { List<EventMessage<?>> allEvents = new ArrayList<>(); unitOfWorkRoot.afterCommit(uow -> schedule(allEvents, processor)); return allEvents; }).addAll(events); } else { schedule(events, processor); } } AsynchronousEventProcessingStrategy(Executor executor,
                                               SequencingPolicy<? super EventMessage<?>> sequencingPolicy); @Override void handle(List<? extends EventMessage<?>> events, Consumer<List<? extends EventMessage<?>>> processor); }
AsynchronousEventProcessingStrategy implements EventProcessingStrategy { @Override public void handle(List<? extends EventMessage<?>> events, Consumer<List<? extends EventMessage<?>>> processor) { if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWorkRoot = CurrentUnitOfWork.get().root(); unitOfWorkRoot.getOrComputeResource(scheduledEventsKey, key -> { List<EventMessage<?>> allEvents = new ArrayList<>(); unitOfWorkRoot.afterCommit(uow -> schedule(allEvents, processor)); return allEvents; }).addAll(events); } else { schedule(events, processor); } } AsynchronousEventProcessingStrategy(Executor executor,
                                               SequencingPolicy<? super EventMessage<?>> sequencingPolicy); @Override void handle(List<? extends EventMessage<?>> events, Consumer<List<? extends EventMessage<?>>> processor);  }