/*
 * Copyright (c) 2010-2018. Axon Framework
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.axonframework.eventsourcing.eventstore.inmemory;

import org.axonframework.eventhandling.GenericEventMessage;
import org.axonframework.eventhandling.TrackedEventMessage;
import org.axonframework.eventsourcing.eventstore.EventStorageEngineTest;
import org.junit.jupiter.api.*;

import java.util.Optional;
import java.util.OptionalLong;
import java.util.stream.Stream;

import static org.junit.jupiter.api.Assertions.*;

/**
 * Test class validating the {@link InMemoryEventStorageEngine}.
 *
 * @author Rene de Waele
 */
class InMemoryEventStorageEngineTest extends EventStorageEngineTest {

    private InMemoryEventStorageEngine testSubject;

    @BeforeEach
    void setUp() {
        testSubject = new InMemoryEventStorageEngine();
        setTestSubject(testSubject);
    }

    @Test
    void testPublishedEventsEmittedToExistingStreams() {
        Stream<? extends TrackedEventMessage<?>> stream = testSubject.readEvents(null, true);
        testSubject.appendEvents(GenericEventMessage.asEventMessage("test"));

        assertTrue(stream.findFirst().isPresent());
    }

    @Test
    void testPublishedEventsEmittedToExistingStreams_WithOffset() {
        testSubject = new InMemoryEventStorageEngine(1);
        Stream<? extends TrackedEventMessage<?>> stream = testSubject.readEvents(null, true);
        testSubject.appendEvents(GenericEventMessage.asEventMessage("test"));

        Optional<? extends TrackedEventMessage<?>> optionalResult = stream.findFirst();
        assertTrue(optionalResult.isPresent());
        OptionalLong optionalResultPosition = optionalResult.get().trackingToken().position();
        assertTrue(optionalResultPosition.isPresent());
        assertEquals(1, optionalResultPosition.getAsLong());
    }
}

@@ -663,16 +663,69 @@
 emory;%0A%0A
+import org.axonframework.eventhandling.EventMessage;%0A
 import o
@@ -887,37 +887,292 @@
 tstore.E
-ventStorageEngineTest
+mbeddedEventStore;%0Aimport org.axonframework.eventsourcing.eventstore.EventStorageEngineTest;%0Aimport org.axonframework.eventsourcing.eventstore.EventStore;%0Aimport org.axonframework.messaging.unitofwork.DefaultUnitOfWork;%0Aimport org.axonframework.messaging.unitofwork.UnitOfWork
 ;%0Aimport
@@ -1512,16 +1512,120 @@
 Test %7B%0A%0A
+    private static final EventMessage%3CObject%3E TEST_EVENT = GenericEventMessage.asEventMessage(%22test%22);%0A%0A
     priv
@@ -2002,50 +2002,18 @@
 nts(
-GenericEventMessage.asEventMessage(%22test%22)
+TEST_EVENT
 );%0A%0A
@@ -2337,50 +2337,18 @@
 nts(
-GenericEventMessage.asEventMessage(%22test%22)
+TEST_EVENT
 );%0A%0A
@@ -2700,11 +2700,1499 @@
 );%0A    %7D
+%0A%0A    /**%0A     * This test issues the publication of an %7B@link EventMessage%7D through the %7B@link EventStore%7D, as this guarantees%0A     * the process of attaching the operation to the %7B@link UnitOfWork%7D are correctly invoked. This allows for%0A     * validation of issue #1056 (https://github.com/AxonFramework/AxonFramework/issues/1056), which defines that we%0A     * should ensure the usage of the %7B@link InMemoryEventStorageEngine%7D will rollback stored events correctly.%0A     */%0A    @Test%0A    void testEventsAreNotStoredWhenTheUnitOfWorkIsRolledBack() %7B%0A        // Given an EmbeddedEventStore and UnitOfWork...%0A        EventStore eventStore = EmbeddedEventStore.builder()%0A                                                  .storageEngine(testSubject)%0A                                                  .build();%0A        UnitOfWork%3CEventMessage%3CObject%3E%3E unitOfWork = DefaultUnitOfWork.startAndGet(TEST_EVENT);%0A%0A        // when _only_ publishing...%0A        eventStore.publish(TEST_EVENT);%0A%0A        // then there are no events in the storage engine, since the UnitOfWork is not committed yet.%0A        Stream%3C? extends TrackedEventMessage%3C?%3E%3E eventStream = testSubject.readEvents(null, true);%0A        assertEquals(0L, eventStream.count());%0A%0A        // When rolling back the UnitOfWork...%0A        unitOfWork.rollback();%0A%0A        // then there are *still* no events in the storage engine.%0A        eventStream = testSubject.readEvents(null, true);%0A        assertEquals(0L, eventStream.count());%0A    %7D
 %0A%7D%0A

Add test to validate rollback of event storage