Resolve remaining TODO's and failing (non-integration) tests

We still need to resolve ignored EventStreamDecorator#decorateForAppend() tests though.

amqp/src/main/java/org/axonframework/contextsupport/spring/amqp/EventBusBeanDefinitionParser.java

@@ -17,7 +17,7 @@
 package org.axonframework.contextsupport.spring.amqp;
 
 import org.axonframework.eventhandling.amqp.spring.ListenerContainerLifecycleManager;
-import org.axonframework.eventhandling.amqp.spring.SpringAMQPEventBus;
+import org.axonframework.eventhandling.amqp.spring.SpringAMQPTerminal;
 import org.springframework.beans.MutablePropertyValues;
 import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
@@ -65,7 +65,7 @@ public class EventBusBeanDefinitionParser extends AbstractBeanDefinitionParser {
     @Override
     protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
         GenericBeanDefinition terminalDefinition = new GenericBeanDefinition();
-        terminalDefinition.setBeanClass(SpringAMQPEventBus.class);
+        terminalDefinition.setBeanClass(SpringAMQPTerminal.class);
         GenericBeanDefinition listenerContainerDefinition = createContainerManager(element, parserContext);
         final String containerBeanName = resolveId(element, terminalDefinition, parserContext)
                 + CONTAINER_MANAGER_SUFFIX;

amqp/src/main/java/org/axonframework/eventhandling/amqp/spring/EventProcessorMessageListener.java

@@ -33,7 +33,7 @@
 
 /**
  * MessageListener implementation that deserializes incoming messages and forwards them to one or more event processors.
- * The <code>byte[]</code> making up the message payload must the format as used by the {@link SpringAMQPEventBus}.
+ * The <code>byte[]</code> making up the message payload must the format as used by the {@link SpringAMQPTerminal}.
  *
  * @author Allard Buijze
  * @since 2.0

amqp/src/main/java/org/axonframework/eventhandling/amqp/spring/SpringAMQPEventBus.java → amqp/src/main/java/org/axonframework/eventhandling/amqp/spring/SpringAMQPTerminal.java

@@ -21,11 +21,9 @@
 import org.axonframework.common.Assert;
 import org.axonframework.common.AxonConfigurationException;
 import org.axonframework.common.Registration;
-import org.axonframework.eventhandling.AbstractEventBus;
+import org.axonframework.eventhandling.EventBus;
 import org.axonframework.eventhandling.EventMessage;
 import org.axonframework.eventhandling.amqp.*;
-import org.axonframework.eventsourcing.eventstore.TrackingEventStream;
-import org.axonframework.eventsourcing.eventstore.TrackingToken;
 import org.axonframework.messaging.unitofwork.CurrentUnitOfWork;
 import org.axonframework.serialization.Serializer;
 import org.slf4j.Logger;
@@ -40,6 +38,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
 
@@ -52,13 +51,13 @@
  * @author Allard Buijze
  * @since 2.0
  */
-public class SpringAMQPEventBus extends AbstractEventBus implements InitializingBean, ApplicationContextAware {
+public class SpringAMQPTerminal implements InitializingBean, ApplicationContextAware {
 
-    //todo turn this into an event processor
-
-    private static final Logger logger = LoggerFactory.getLogger(SpringAMQPEventBus.class);
+    private static final Logger logger = LoggerFactory.getLogger(SpringAMQPTerminal.class);
     private static final String DEFAULT_EXCHANGE_NAME = "Axon.EventBus";
 
+    private final EventBus eventBus;
+
     private ConnectionFactory connectionFactory;
     private String exchangeName = DEFAULT_EXCHANGE_NAME;
     private boolean isTransactional = false;
@@ -70,9 +69,21 @@ public class SpringAMQPEventBus extends AbstractEventBus implements Initializing
     private RoutingKeyResolver routingKeyResolver;
     private boolean waitForAck;
     private long publisherAckTimeout;
+    private Registration eventBusRegistration;
 
-    @Override
-    protected void prepareCommit(List<? extends EventMessage<?>> events) {
+    public SpringAMQPTerminal(EventBus eventBus) {
+        this.eventBus = eventBus;
+    }
+
+    public void start() {
+        eventBusRegistration = eventBus.subscribe(this::send);
+    }
+
+    public void shutDown() {
+        Optional.ofNullable(eventBusRegistration).ifPresent(Registration::cancel);
+    }
+
+    protected void send(List<? extends EventMessage<?>> events) {
         Channel channel = connectionFactory.createConnection().createChannel(isTransactional);
         try {
             if (isTransactional) {
@@ -179,13 +190,6 @@ private void tryRollback(Channel channel) {
         }
     }
 
-    @Override
-    public TrackingEventStream streamEvents(TrackingToken trackingToken) {
-        //todo can be removed when this is modified to be an event processor
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
     public Registration subscribe(Consumer<List<? extends EventMessage<?>>> eventProcessor) {
         AMQPConsumerConfiguration config = new DefaultAMQPConsumerConfiguration(eventProcessor.toString());
         return getListenerContainerLifecycleManager().registerEventProcessor(eventProcessor, config, messageConverter);
@@ -289,7 +293,7 @@ public void setPublisherAckTimeout(long publisherAckTimeout) {
     /**
      * Sets the ConnectionFactory providing the Connections and Channels to send messages on. The SpringAMQPTerminal
      * does not cache or reuse connections. Providing a ConnectionFactory instance that caches connections will prevent
-     * new connections to be opened for each invocation to {@link #publish(EventMessage[])}
+     * new connections to be opened for each invocation to {@link #accept(List)}
      * <p>
      * Defaults to an autowired Connection Factory.
      *

amqp/src/main/resources/org/axonframework/contextsupport/spring/amqp/axon-amqp-3.0.xsd

@@ -36,7 +36,7 @@
             </xsd:documentation>
             <xsd:appinfo>
                 <tool:annotation>
-                    <tool:exports type="org.axonframework.eventhandling.amqp.spring.SpringAMQPEventBus"/>
+                    <tool:exports type="org.axonframework.eventhandling.amqp.spring.SpringAMQPTerminal"/>
                     <tool:exports type="org.axonframework.eventhandling.amqp.spring.ListenerContainerLifecycleManager"/>
                 </tool:annotation>
             </xsd:appinfo>

amqp/src/test/java/org/axonframework/contextsupport/spring/amqp/EventBusBeanDefinitionParserTest.java

@@ -21,7 +21,7 @@
 import org.axonframework.eventhandling.amqp.AMQPMessageConverter;
 import org.axonframework.eventhandling.amqp.RoutingKeyResolver;
 import org.axonframework.eventhandling.amqp.spring.SpringAMQPConsumerConfiguration;
-import org.axonframework.eventhandling.amqp.spring.SpringAMQPEventBus;
+import org.axonframework.eventhandling.amqp.spring.SpringAMQPTerminal;
 import org.axonframework.serialization.Serializer;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -64,7 +64,7 @@ public void testContextStartup() {
         assertNotNull(applicationContext);
         assertNotNull(beanFactory);
         BeanDefinition terminal1 = beanFactory.getBeanDefinition("terminal1");
-        assertEquals(SpringAMQPEventBus.class.getName(), terminal1.getBeanClassName());
+        assertEquals(SpringAMQPTerminal.class.getName(), terminal1.getBeanClassName());
         assertTrue(property(terminal1, "listenerContainerLifecycleManager") instanceof BeanReference);
 
         assertEquals("connectionFactory", property(terminal1, "connectionFactory", BeanReference.class).getBeanName());

amqp/src/test/java/org/axonframework/eventhandling/amqp/spring/SpringAMQPEventBusIntegrationTest.java → amqp/src/test/java/org/axonframework/eventhandling/amqp/spring/SpringAMQPTerminalIntegrationTest.java

@@ -48,7 +48,7 @@
 @ContextConfiguration(locations = "/META-INF/spring/messaging-context.xml")
 @RunWith(SpringJUnit4ClassRunner.class)
 @Ignore
-public class SpringAMQPEventBusIntegrationTest {
+public class SpringAMQPTerminalIntegrationTest {
 
     private static final int EVENT_COUNT = 100;
     private static final int THREAD_COUNT = 10;
@@ -57,7 +57,7 @@ public class SpringAMQPEventBusIntegrationTest {
     @Autowired
     private ConnectionFactory connectionFactory;
     @Autowired
-    private SpringAMQPEventBus terminal;
+    private SpringAMQPTerminal terminal;
 
     @Autowired
     @Qualifier("eventProcessor1")

amqp/src/test/java/org/axonframework/eventhandling/amqp/spring/SpringAMQPEventBusTest.java → amqp/src/test/java/org/axonframework/eventhandling/amqp/spring/SpringAMQPTerminalTest.java

@@ -19,6 +19,7 @@
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
 import org.axonframework.eventhandling.GenericEventMessage;
+import org.axonframework.eventhandling.SimpleEventBus;
 import org.axonframework.eventhandling.amqp.DefaultAMQPMessageConverter;
 import org.axonframework.eventhandling.amqp.EventPublicationFailedException;
 import org.axonframework.messaging.unitofwork.CurrentUnitOfWork;
@@ -44,29 +45,33 @@
 /**
  * @author Allard Buijze
  */
-public class SpringAMQPEventBusTest {
+public class SpringAMQPTerminalTest {
 
     private static final Charset UTF_8 = Charset.forName("UTF-8");
-    private SpringAMQPEventBus testSubject;
+    private SpringAMQPTerminal testSubject;
     private ConnectionFactory connectionFactory;
     private Serializer serializer;
+    private SimpleEventBus eventBus;
 
     @Before
     public void setUp() throws Exception {
-        testSubject = new SpringAMQPEventBus();
+        eventBus = new SimpleEventBus();
+        testSubject = new SpringAMQPTerminal(eventBus);
         connectionFactory = mock(ConnectionFactory.class);
         serializer = mock(Serializer.class);
         testSubject.setConnectionFactory(connectionFactory);
         testSubject.setExchangeName("mockExchange");
         testSubject.setTransactional(true);
         testSubject.setMessageConverter(new DefaultAMQPMessageConverter(serializer));
+        testSubject.start();
     }
 
     @After
     public void tearDown() {
         while (CurrentUnitOfWork.isStarted()) {
             CurrentUnitOfWork.get().rollback();
         }
+        testSubject.shutDown();
     }
 
     @Test
@@ -80,7 +85,7 @@ public void testSendMessage_NoUnitOfWork() throws IOException {
                 .thenReturn(new SimpleSerializedObject<>("Message".getBytes(UTF_8), byte[].class, "String", "0"));
         when(serializer.serialize(message.getMetaData(), byte[].class))
                 .thenReturn(new SerializedMetaData<>(new byte[0], byte[].class));
-        testSubject.publish(message);
+        eventBus.publish(message);
 
         verify(transactionalChannel).basicPublish(eq("mockExchange"), eq("java.lang"),
                                                   eq(false), eq(false),
@@ -103,7 +108,7 @@ public void testSendMessage_WithTransactionalUnitOfWork() throws IOException {
                 .thenReturn(new SimpleSerializedObject<>("Message".getBytes(UTF_8), byte[].class, "String", "0"));
         when(serializer.serialize(message.getMetaData(), byte[].class))
                 .thenReturn(new SerializedMetaData<>(new byte[0], byte[].class));
-        testSubject.publish(message);
+        eventBus.publish(message);
 
         uow.commit();
         verify(transactionalChannel).basicPublish(eq("mockExchange"), eq("java.lang"),
@@ -127,7 +132,7 @@ public void testSendMessage_WithTransactionalUnitOfWork_ChannelClosedBeforeCommi
                 .thenReturn(new SimpleSerializedObject<>("Message".getBytes(UTF_8), byte[].class, "String", "0"));
         when(serializer.serialize(message.getMetaData(), byte[].class))
                 .thenReturn(new SerializedMetaData<>(new byte[0], byte[].class));
-        testSubject.publish(message);
+        eventBus.publish(message);
 
         try {
             uow.commit();
@@ -157,7 +162,7 @@ public void testSendMessage_WithUnitOfWorkRollback() throws IOException {
                 .thenReturn(new SimpleSerializedObject<>("Message".getBytes(UTF_8), byte[].class, "String", "0"));
         when(serializer.serialize(message.getMetaData(), byte[].class))
                 .thenReturn(new SerializedMetaData<>(new byte[0], byte[].class));
-        testSubject.publish(message);
+        eventBus.publish(message);
 
         verify(transactionalChannel, never()).txRollback();
         verify(transactionalChannel, never()).txCommit();
@@ -193,7 +198,7 @@ public void testSendMessageWithPublisherAck_UnitOfWorkCommitted()
 
         UnitOfWork<?> uow = DefaultUnitOfWork.startAndGet(message);
 
-        testSubject.publish(message);
+        eventBus.publish(message);
         verify(channel, never()).waitForConfirms();
 
         uow.commit();
@@ -239,7 +244,7 @@ public void testSendMessageWithPublisherAck_NoActiveUnitOfWork() throws Interrup
         when(serializer.serialize(message.getMetaData(), byte[].class))
                 .thenReturn(new SerializedMetaData<>(new byte[0], byte[].class));
 
-        testSubject.publish(message);
+        eventBus.publish(message);
         verify(channel).confirmSelect();
         verify(channel).basicPublish(eq("mockExchange"), eq("java.lang"),
                                      eq(false), eq(false),

core/src/main/java/org/axonframework/eventhandling/AbstractEventProcessor.java

@@ -71,20 +71,22 @@ public void accept(List<? extends EventMessage<?>> eventMessages) {
         Map<? extends EventMessage<?>, MessageMonitor.MonitorCallback> monitorCallbacks =
                 eventMessages.stream().collect(toMap(Function.identity(), messageMonitor::onMessageIngested));
         UnitOfWork<? extends EventMessage<?>> unitOfWork = new BatchingUnitOfWork<>(eventMessages);
-        unitOfWork.onRollback(uow -> errorHandler
-                .handleError(getName(), uow.getExecutionResult().getExceptionResult(), eventMessages,
-                             () -> accept(eventMessages)));
-        unitOfWork.onCleanup(uow -> {
-            MessageMonitor.MonitorCallback callback = monitorCallbacks.get(uow.getMessage());
-            if (uow.isRolledBack()) {
-                callback.reportFailure(uow.getExecutionResult().getExceptionResult());
-            } else {
-                callback.reportSuccess();
-            }
-        });
         try {
             unitOfWork.executeWithResult(
-                    () -> new DefaultInterceptorChain<>(unitOfWork, interceptors, eventHandlerInvoker).proceed(),
+                    () -> {
+                        unitOfWork.onRollback(uow -> errorHandler
+                                .handleError(getName(), uow.getExecutionResult().getExceptionResult(), eventMessages,
+                                             () -> accept(eventMessages)));
+                        unitOfWork.onCleanup(uow -> {
+                            MessageMonitor.MonitorCallback callback = monitorCallbacks.get(uow.getMessage());
+                            if (uow.isRolledBack()) {
+                                callback.reportFailure(uow.getExecutionResult().getExceptionResult());
+                            } else {
+                                callback.reportSuccess();
+                            }
+                        });
+                        return new DefaultInterceptorChain<>(unitOfWork, interceptors, eventHandlerInvoker).proceed();
+                    },
                     rollbackConfiguration);
         } catch (Exception e) {
             throw new EventProcessingException(

core/src/main/java/org/axonframework/eventhandling/saga/AnnotatedSagaManager.java

@@ -43,9 +43,6 @@ public AnnotatedSagaManager(Class<T> sagaType, SagaRepository<T> sagaRepository,
         this(sagaType, sagaRepository, sagaFactory, new DefaultSagaMetaModelFactory().modelOf(sagaType));
     }
 
-    /**
-     * TODO: Javadoc
-     */
     public AnnotatedSagaManager(Class<T> sagaType, SagaRepository<T> sagaRepository, Callable<T> sagaFactory,
                                 SagaModel<T> sagaMetaModel) {
         super(sagaType, sagaRepository, sagaFactory);

core/src/main/java/org/axonframework/eventhandling/saga/SagaRepository.java

@@ -51,9 +51,6 @@ public interface SagaRepository<T> {
      */
     Saga<T> load(String sagaIdentifier);
 
-    /**
-     * TODO: Fix docs
-     */
     Saga<T> newInstance(Callable<T> factoryMethod);
 
 }

core/src/main/java/org/axonframework/eventsourcing/AbstractSnapshotter.java

@@ -43,9 +43,8 @@ public abstract class AbstractSnapshotter implements Snapshotter {
     private final TransactionManager transactionManager;
 
     /**
-     * Initializes the Snapshotter to append snapshots in the given {@code eventStore}. This snapshotter will
-     * create the snapshots in the process that triggers them, and save them into the Event Store without any
-     * transaction.
+     * Initializes the Snapshotter to append snapshots in the given {@code eventStore}. This snapshotter will create the
+     * snapshots in the process that triggers them, and save them into the Event Store without any transaction.
      *
      * @param eventStorageEngine the EventStore instance to store snapshots in
      */
@@ -54,36 +53,37 @@ protected AbstractSnapshotter(EventStorageEngine eventStorageEngine) {
     }
 
     /**
-     * Initializes the Snapshotter to append snapshots in the given {@code eventStore}. This snapshotter will create
-     * the snapshots in the process that triggers them, and save them into the Event Store in a transaction managed by
-     * the given {@code transactionManager}.
+     * Initializes the Snapshotter to append snapshots in the given {@code eventStore}. This snapshotter will create the
+     * snapshots in the process that triggers them, and save them into the Event Store in a transaction managed by the
+     * given {@code transactionManager}.
      *
-     * @param eventStorageEngine         the EventStore instance to store snapshots in
+     * @param eventStorageEngine the EventStore instance to store snapshots in
      * @param transactionManager The transaction manager to create the surrounding transaction with
      */
     protected AbstractSnapshotter(EventStorageEngine eventStorageEngine, TransactionManager transactionManager) {
         this(eventStorageEngine, DirectExecutor.INSTANCE, transactionManager);
     }
 
     /**
-     * Initializes the Snapshotter to append snapshots in the given {@code eventStore}. This snapshotter will create
-     * the snapshots in the process provided by the given {@code executor}, and save them into the Event Store in a
+     * Initializes the Snapshotter to append snapshots in the given {@code eventStore}. This snapshotter will create the
+     * snapshots in the process provided by the given {@code executor}, and save them into the Event Store in a
      * transaction managed by the given {@code transactionManager}.
      *
-     * @param eventStorageEngine         The EventStore instance to store snapshots in
+     * @param eventStorageEngine The EventStore instance to store snapshots in
      * @param executor           The executor that handles the actual snapshot creation process
      * @param transactionManager The transaction manager to create the surrounding transaction with
      */
-    protected AbstractSnapshotter(EventStorageEngine eventStorageEngine, Executor executor, TransactionManager transactionManager) {
+    protected AbstractSnapshotter(EventStorageEngine eventStorageEngine, Executor executor,
+                                  TransactionManager transactionManager) {
         this.eventStorageEngine = eventStorageEngine;
         this.executor = executor;
         this.transactionManager = transactionManager;
     }
 
     @Override
     public void scheduleSnapshot(Class<?> aggregateType, String aggregateIdentifier) {
-        executor.execute(() -> transactionManager
-                .executeInTransaction(new SilentTask(createSnapshotterTask(aggregateType, aggregateIdentifier))));
+        executor.execute(new SilentTask(() -> transactionManager
+                .executeInTransaction(createSnapshotterTask(aggregateType, aggregateIdentifier))));
     }
 
     /**
@@ -98,9 +98,8 @@ protected Runnable createSnapshotterTask(Class<?> aggregateType, String aggregat
     }
 
     /**
-     * Creates a snapshot event for an aggregate of which passed events are available in the given
-     * {@code eventStream}. May return {@code null} to indicate a snapshot event is not necessary or
-     * appropriate for the given event stream.
+     * Creates a snapshot event for an aggregate of which passed events are available in the given {@code eventStream}.
+     * May return {@code null} to indicate a snapshot event is not necessary or appropriate for the given event stream.
      *
      * @param aggregateType       The aggregate's type identifier
      * @param aggregateIdentifier The identifier of the aggregate to create a snapshot for
@@ -165,7 +164,7 @@ private CreateSnapshotTask(Class<?> aggregateType, String identifier) {
 
         @Override
         public void run() {
-            DomainEventStream eventStream = eventStream = eventStorageEngine.readEvents(identifier);
+            DomainEventStream eventStream = eventStorageEngine.readEvents(identifier);
             // a snapshot should only be stored if the snapshot replaces at least more than one event
             long firstEventSequenceNumber = eventStream.peek().getSequenceNumber();
             DomainEventMessage snapshotEvent = createSnapshot(aggregateType, identifier, eventStream);