@Test void testEventsScheduledForHandling() { EventMessage<?> message1 = createEvent("aggregate1", 1); EventMessage<?> message2 = createEvent("aggregate2", 1); testSubject.handle(Arrays.asList(message1, message2), mock(Consumer.class)); verify(executor, times(2)).execute(isA(Runnable.class)); }
@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);  }