@Test void testPublicationForbiddenDuringRootUowCommitPhase() { testSubject = spy(StubPublishingEventBus.builder().publicationPhase(UnitOfWork.Phase.COMMIT).build()); testSubject.publish(numberedEvent(1)); assertThrows(IllegalStateException.class, unitOfWork::commit); }
@Override public void publish(List<? extends EventMessage<?>> events) { List<MessageMonitor.MonitorCallback> ingested = events.stream().map(messageMonitor::onMessageIngested) .collect(Collectors.toList()); if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWork = CurrentUnitOfWork.get(); Assert.state(!unitOfWork.phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the current Unit of Work has already been " + "committed. Please start a new Unit of Work before publishing events."); Assert.state(!unitOfWork.root().phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the root Unit of Work has already been " + "committed."); unitOfWork.afterCommit(u -> ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess)); unitOfWork.onRollback(uow -> ingested.forEach( message -> message.reportFailure(uow.getExecutionResult().getExceptionResult()) )); eventsQueue(unitOfWork).addAll(events); } else { try { prepareCommit(intercept(events)); commit(events); afterCommit(events); ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess); } catch (Exception e) { ingested.forEach(m -> m.reportFailure(e)); throw e; } } }
AbstractEventBus implements EventBus { @Override public void publish(List<? extends EventMessage<?>> events) { List<MessageMonitor.MonitorCallback> ingested = events.stream().map(messageMonitor::onMessageIngested) .collect(Collectors.toList()); if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWork = CurrentUnitOfWork.get(); Assert.state(!unitOfWork.phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the current Unit of Work has already been " + "committed. Please start a new Unit of Work before publishing events."); Assert.state(!unitOfWork.root().phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the root Unit of Work has already been " + "committed."); unitOfWork.afterCommit(u -> ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess)); unitOfWork.onRollback(uow -> ingested.forEach( message -> message.reportFailure(uow.getExecutionResult().getExceptionResult()) )); eventsQueue(unitOfWork).addAll(events); } else { try { prepareCommit(intercept(events)); commit(events); afterCommit(events); ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess); } catch (Exception e) { ingested.forEach(m -> m.reportFailure(e)); throw e; } } } }
AbstractEventBus implements EventBus { @Override public void publish(List<? extends EventMessage<?>> events) { List<MessageMonitor.MonitorCallback> ingested = events.stream().map(messageMonitor::onMessageIngested) .collect(Collectors.toList()); if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWork = CurrentUnitOfWork.get(); Assert.state(!unitOfWork.phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the current Unit of Work has already been " + "committed. Please start a new Unit of Work before publishing events."); Assert.state(!unitOfWork.root().phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the root Unit of Work has already been " + "committed."); unitOfWork.afterCommit(u -> ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess)); unitOfWork.onRollback(uow -> ingested.forEach( message -> message.reportFailure(uow.getExecutionResult().getExceptionResult()) )); eventsQueue(unitOfWork).addAll(events); } else { try { prepareCommit(intercept(events)); commit(events); afterCommit(events); ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess); } catch (Exception e) { ingested.forEach(m -> m.reportFailure(e)); throw e; } } } protected  AbstractEventBus(Builder builder); }
AbstractEventBus implements EventBus { @Override public void publish(List<? extends EventMessage<?>> events) { List<MessageMonitor.MonitorCallback> ingested = events.stream().map(messageMonitor::onMessageIngested) .collect(Collectors.toList()); if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWork = CurrentUnitOfWork.get(); Assert.state(!unitOfWork.phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the current Unit of Work has already been " + "committed. Please start a new Unit of Work before publishing events."); Assert.state(!unitOfWork.root().phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the root Unit of Work has already been " + "committed."); unitOfWork.afterCommit(u -> ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess)); unitOfWork.onRollback(uow -> ingested.forEach( message -> message.reportFailure(uow.getExecutionResult().getExceptionResult()) )); eventsQueue(unitOfWork).addAll(events); } else { try { prepareCommit(intercept(events)); commit(events); afterCommit(events); ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess); } catch (Exception e) { ingested.forEach(m -> m.reportFailure(e)); throw e; } } } protected  AbstractEventBus(Builder builder); @Override Registration subscribe(Consumer<List<? extends EventMessage<?>>> eventProcessor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super EventMessage<?>> dispatchInterceptor); @Override void publish(List<? extends EventMessage<?>> events); }
AbstractEventBus implements EventBus { @Override public void publish(List<? extends EventMessage<?>> events) { List<MessageMonitor.MonitorCallback> ingested = events.stream().map(messageMonitor::onMessageIngested) .collect(Collectors.toList()); if (CurrentUnitOfWork.isStarted()) { UnitOfWork<?> unitOfWork = CurrentUnitOfWork.get(); Assert.state(!unitOfWork.phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the current Unit of Work has already been " + "committed. Please start a new Unit of Work before publishing events."); Assert.state(!unitOfWork.root().phase().isAfter(PREPARE_COMMIT), () -> "It is not allowed to publish events when the root Unit of Work has already been " + "committed."); unitOfWork.afterCommit(u -> ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess)); unitOfWork.onRollback(uow -> ingested.forEach( message -> message.reportFailure(uow.getExecutionResult().getExceptionResult()) )); eventsQueue(unitOfWork).addAll(events); } else { try { prepareCommit(intercept(events)); commit(events); afterCommit(events); ingested.forEach(MessageMonitor.MonitorCallback::reportSuccess); } catch (Exception e) { ingested.forEach(m -> m.reportFailure(e)); throw e; } } } protected  AbstractEventBus(Builder builder); @Override Registration subscribe(Consumer<List<? extends EventMessage<?>>> eventProcessor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super EventMessage<?>> dispatchInterceptor); @Override void publish(List<? extends EventMessage<?>> events);  }