public class StreamNotFoundException extends Exception {

    public StreamNotFoundException() {
    }

    public StreamNotFoundException(String message) {
        super(message);
    }

    public StreamNotFoundException(Throwable cause) {
        super(cause);
    }

    public StreamNotFoundException(String message, Throwable cause) {
        super(message, cause);
    }
    
} public class ConcurrencyException extends Exception {

    public ConcurrencyException() {
    }

    public ConcurrencyException(String message) {
        super(message);
    }

    public ConcurrencyException(Throwable cause) {
        super(cause);
    }

    public ConcurrencyException(String message, Throwable cause) {
        super(message, cause);
    }
    
} public interface EventStore {

    /**
     * Tests whether the stream identified by {@code streamId} exists in the
     * bucket identified by {@code bucketId}.
     * 
     * @param bucketId  the identifier of the bucket to which the stream belongs
     * @param streamId  the identifier of the stream that is tested for existence
     * @return  whether the specified stream exists
     */
    boolean existsStream(String bucketId, String streamId);

    /**
     * Opens the latest version of the stream identified by {@code streamId} in the
     * bucket identified by {@code bucketId} for reading.
     * 
     * @param bucketId  the identifier of the bucket to which the stream belongs
     * @param streamId  the identifier of the stream that is opened for reading
     * @return  the requested event stream as readable event stream
     * @throws StreamNotFoundException  if a stream with the given identifier cannot be found in the bucket
     */
    ReadableEventStream openStreamForReading(String bucketId, String streamId)
            throws StreamNotFoundException;

    /**
     * Opens the stream identified by {@code streamId} in the
     * bucket identified by {@code bucketId} for reading, but only with
     * a maximum version of {@code maxVersion}.
     * Can be used to load previous versions of the event stream for reading.
     * If {@code maxVersion} is larger than the latest stream version,
     * the latest available version of the stream will be loaded.
     * 
     * @param bucketId  the identifier of the bucket to which the stream belongs
     * @param streamId  the identifier of the stream that is opened for reading
     * @param maxVersion  the requested version or {@code Long.MAX_VALUE} to load the latest version
     * @return  the requested event stream as readable event stream
     * @throws StreamNotFoundException  if a stream with the given identifier cannot be found in the bucket
     */
    ReadableEventStream openStreamForReading(String bucketId, String streamId, long maxVersion)
            throws StreamNotFoundException;

    /**
     * Creates a new stream with identifier {@code streamId} in the bucket
     * identified by {@code bucketId}.
     * 
     * 
     * @param bucketId  the identifier of the bucket to which the stream belongs
     * @param streamId  the identifier of the stream that is created
     * @return  the requested event stream as writable event stream
     */
    WritableEventStream createStream(String bucketId, String streamId);

    /**
     * Open the version {@code version} of the stream identified by {@code streamId}
     * in the bucket identified by {@code bucketId} for writing.
     * 
     * @param bucketId  the identifier of the bucket to which the stream belongs
     * @param streamId  the identifier of the stream that is opened for reading
     * @param version   the requested version of the stream
     * @return  the requested event stream as writable event stream
     */
    WritableEventStream openStreamForWriting(String bucketId, String streamId, long version);

} public class MockPersistence implements EventStorePersistence {

    private static final String RESET = "RESET";
    private final List<ChangeSet> changeSets = new ArrayList<>();

    public MockPersistence() { }

    @Override
    public boolean existsStream(String bucketId, String streamId) {
        return !changeSets.isEmpty();
    }

    @Override
    public Iterator<ChangeSet> allChanges(String bucketId) {
        return changeSets.iterator();
    }

    @Override
    public Iterator<ChangeSet> getFrom(String bucketId, String streamId, long minVersion, long maxVersion) 
            throws StreamNotFoundException {
        if (maxVersion == Long.MAX_VALUE)
            maxVersion = Integer.MAX_VALUE;
        int max = Math.min(changeSets.size(), (int) maxVersion);
        int min = Math.min(changeSets.size(), (int) minVersion);
        return changeSets.subList(min, max).iterator();
    }

    @Override
    public void persistChanges(ChangeSet changeSet) 
            throws ConcurrencyException, DuplicateCommitException {
        if (RESET.equals(changeSet.streamId())) {
            this.cleanup();
            return;
        }
        // version 0 -> changeset.size() == 0, i.e. wrong if size > 0 
        // version 1 -> size == 1
        // etc

        // we add the changeSet first, such that we can ignore the
        // exception in tests where we actually want to create an invalid stream.
        changeSets.add(changeSet);
        if (changeSets.size() != changeSet.streamVersion())
            throw new ConcurrencyException();
    }

    private void cleanup() {
        changeSets.clear();
    }

    public static ChangeSet resetCommand() {
        return new TestChangeSet("DEFAULT_BUCKET", RESET);
    }
    
} public interface ChangeSet extends Serializable {

    /**
     * Identifies the bucket to which the event stream belongs.
     * 
     * @return  the identifier
     */
    String bucketId();

    /**
     * Identifies the event stream to which this belongs.
     * 
     * @return  the identifier
     */
    String streamId();

    /**
     * Gets the version of the event stream to which this applies.
     * 
     * @return  the version
     */
    long streamVersion();

    /**
     * Gets the unique identifier for this ChangeSet within the stream.
     * 
     * @return the unique identifer
     */
    String changeSetId();
    
    /**
     * Gets an iterator to the events contained within this ChangeSet.
     * 
     * @return  the iterator
     */
    Iterator<Serializable> events();
    
} public class DuplicateCommitException extends Exception {

    public DuplicateCommitException() {
    }

    public DuplicateCommitException(String message) {
        super(message);
    }

    public DuplicateCommitException(Throwable cause) {
        super(cause);
    }

    public DuplicateCommitException(String message, Throwable cause) {
        super(message, cause);
    }
    
} public interface ReadableEventStream extends VersionedEventStream {

    /**
     * Gets an iterator to the events contained within this stream.
     * 
     * @return  the iterator
     */
    Iterator<Serializable> events();

} public interface WritableEventStream extends VersionedEventStream {

    /**
     * Append the given event to the event stream.
     * 
     * @param event  the event to be appended, not null
     */
    void append(Serializable event);

    /**
     * Commits the changes to durable storage.
     * 
     * @param commitId  the value that uniquely identifies the commit, null returns empty-handed
     * @throws DuplicateCommitException  if a commit with the same id already exists in the bucket
     * @throws ConcurrencyException  if a {@link ChangeSet} with the same streamVersion already exists
     *   in the durable storage.
     */
    void commit(String commitId) 
            throws DuplicateCommitException, ConcurrencyException;

    /**
     * Clear all uncommitted changes.
     */
    void rollback();
    
} public interface EventStoreCommitListener {

    /**
     * Receives a notification that a change set has been committed to the
     * event store.
     * This method must be thread-safe.
     * <p>
     * See the {@link EventStoreCommitNotifier} documentation on the transaction
     * semantics and quality guarantees given for this function.
     * 
     * @param notification  the actual notification
     */
    void receive(EventStoreCommitNotification notification);
    
} public interface EventStoreCommitNotification {

    /**
     * Returns the change set that has been committed to the event store.
     * 
     * @return  the committed {@link ChangeSet}
     */
    ChangeSet changes();
    
} public interface EventStoreCommitNotifier {

    /**
     * Notifies all registered listeners about the committed {@link ChangeSet}.
     * Non-delivered notifications are not persisted during application
     * restarts to avoid expensive 2-phase-commits.  If the server crashes or
     * stops while the notification is in progress, clients are expected to 
     * recover manually on the next server startup (e.g., by replaying the full
     * event store on application startup using {@link EventStorePersistence#allChanges}).
     * 
     * @param changeSet  the change set that has been committed, not null
     */
    void notifyListeners(ChangeSet changeSet);

    /**
     * Adds an interested listener for the given bucket.
     * The listener must not be registered in that bucket already.
     * 
     * @param bucketId  the bucket that the listener is interested in
     * @param listener  the listener that shall be added, not null
     */
    void addListener(String bucketId, EventStoreCommitListener listener);

    /**
     * Removes a registered listener from the given bucket.
     * The listener must be registered in the specified bucket
     * 
     * @param bucketId  the bucket that the listener was registered for
     * @param listener  the listener that shall be removed, not null
     */
    void removeListener(String bucketId, EventStoreCommitListener listener);
    
} public interface EventStorePersistence {

    /**
     * Tests whether an event stream with identifier {@code streamId} exists
     * in the bucket identified by {@code bucketId}.
     * 
     * @param bucketId  the identifier of the bucket to which the stream belongs, not null
     * @param streamId  the identifier of the stream that is tested for existence, not null
     * @return  whether the specified stream exists
     */
    boolean existsStream(String bucketId, String streamId);
    
    /**
     * Gets an iterator to all changes persisted to the given bucket.
     * The order of events is consistent on a per-stream basis, but no
     * other ordering guarantee can be given.
     * <p>
     * Some persistence implementations support deferred ("streamed") loading
     * of events from the persistence store.
     * It depends on the persistence implementation whether {@link #allChanges}
     * requires an open transaction and requires {@link Iterator#next()} to be
     * called within this open transaction.
     * 
     * @param bucketId  the identifier of the bucket from which the changes are fetched, not null
     * @return  the iterator to the changes
     */
    Iterator<ChangeSet> allChanges(String bucketId);

    /**
     * Gets an iterator to all changes ({@link ChangeSet}s) between version
     * {@code minVersion} (exclusive) and version {@code maxVersion} (inclusive)
     * in the specified event stream.
     * The order of the respective {@link ChangeSet#streamVersion} is guaranteed
     * to be strictly increasing.
     * <p>
     * Some persistence implementations support deferred ("streamed") loading
     * of events from the persistence store.
     * It depends on the persistence implementation whether {@link #getFrom}
     * requires an open transaction and requires {@link Iterator#next()} to be
     * called within this open transaction.
     * 
     * @param bucketId  the identifier of the bucket to which the stream belongs, not null
     * @param streamId  the identifier of the stream that is to be retrieved from, not null
     * @param minVersion  the minimum version (exclusive) of the {@link ChangeSet} to fetch
     * @param maxVersion  the maximum version (inclusive) of the {@link ChangeSet} to fetch
     * @throws StreamNotFoundException if the given stream cannot be found
     * @return  the iterator to the {@link ChangeSet}s
     */
    Iterator<ChangeSet> getFrom(
            String bucketId,
            String streamId,
            long minVersion,
            long maxVersion) throws StreamNotFoundException;

    /**
     * Persists the given {@link ChangeSet} to the durable storage.
     * 
     * @param changeSet  the changes to be persisted, not null
     * @throws ConcurrencyException  if there already exists a {@link ChangeSet}
     *  within the respective stream that has the same {@link ChangeSet#streamVersion() version}
     *  (Not supported with all implementations)
     * @throws DuplicateCommitException  if there already exists a {@link ChangeSet}
     *  with the same id (Not supported with all implementations)
     */
    void persistChanges(ChangeSet changeSet)
            throws ConcurrencyException, DuplicateCommitException;
    
} @ConcurrencyManagement(ConcurrencyManagementType.CONTAINER)
public class SyncEventStoreCommitNotifier
        extends AbstractEventStoreCommitNotifier 
        implements EventStoreCommitNotifier {

    @Override
    @Lock(LockType.READ)
    public void notifyListeners(ChangeSet changeSet) {
        if (changeSet == null)
            throw new IllegalArgumentException("changeSet must not be null");
        this.performNotification(changeSet);
    }
    
} public class IteratorUtils {
    
    /** 
     * Gets a list based on an iterator. 
     * As the wrapped Iterator is traversed, an {@link ArrayList} of its values
     * is created. At the end, the list is returned.

     * @param <T> the event type
     * @param it  the iterator to be traversed
     * @return   the list of values found in the traversal of the iterator
     */
    public static <T> List<T> toList(Iterator<T> it) {
        List<T> list = new ArrayList<>();
        while (it.hasNext())
            list.add(it.next());
        return list;
    }
    
}
import org.jeeventstore.StreamNotFoundException;
import org.jeeventstore.ConcurrencyException;
import org.jeeventstore.EventStore;
import org.jeeventstore.persistence.MockPersistence;
import java.io.File;
import java.io.Serializable;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import javax.ejb.EJB;
import javax.ejb.EJBException;
import org.jboss.arquillian.container.test.api.Deployment;
import org.jboss.arquillian.testng.Arquillian;
import org.jboss.shrinkwrap.api.Archive;
import org.jboss.shrinkwrap.api.ShrinkWrap;
import org.jboss.shrinkwrap.api.spec.EnterpriseArchive;
import org.jboss.shrinkwrap.api.spec.JavaArchive;
import org.jeeventstore.ChangeSet;
import org.jeeventstore.DuplicateCommitException;
import org.jeeventstore.ReadableEventStream;
import org.jeeventstore.WritableEventStream;
import org.jeeventstore.EventStoreCommitListener;
import org.jeeventstore.EventStoreCommitNotification;
import org.jeeventstore.EventStoreCommitNotifier;
import org.jeeventstore.EventStorePersistence;
import org.jeeventstore.notifier.SyncEventStoreCommitNotifier;
import org.jeeventstore.util.IteratorUtils;
import static org.testng.Assert.*;
import org.testng.annotations.Test;
/*
 * Copyright (c) 2013-2014 Red Rainbow IT Solutions GmbH, Germany
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to
 * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
 * the Software, and to permit persons to whom the Software is furnished to do so,
 * subject to the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
 * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
 * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
 * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

package org.jeeventstore.store;


public class OptimisticEventStoreServiceTest extends Arquillian 
    implements EventStoreCommitListener {

    public final static String BUCKET_ID = "BUCKET_ID";

    // these must be static to have them set from different threads
    protected static boolean caught = false;
    protected static ChangeSet caughtCS = null;

    @Override
    public void receive(EventStoreCommitNotification notification) {
        caught = true;
        caughtCS = notification.changes();
    }

    @Deployment
    public static Archive<?> deployment() {
        EnterpriseArchive ear = ShrinkWrap.create(EnterpriseArchive.class, "test.ear")
                .addAsModule(ShrinkWrap.create(JavaArchive.class, "ejb.jar")
                        .addAsManifestResource(new File("src/test/resources/META-INF/beans.xml"))
                        .addAsManifestResource(new File(
                                "src/test/resources/META-INF/ejb-jar-OptimisticEventStoreServiceTest.xml"),
                                "ejb-jar.xml")
                        .addPackage(ChangeSet.class.getPackage())
                        .addPackage(EventStoreCommitNotifier.class.getPackage())