libraries/eventsourcing/src/main/java/org/qi4j/library/eventsourcing/domain/source/AbstractEventStoreMixin.java

/**
 *
 * Copyright 2009-2010 Rickard Öberg AB
 *
 * 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.qi4j.library.eventsourcing.domain.source;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.qi4j.api.entity.Identity;
import org.qi4j.api.injection.scope.Structure;
import org.qi4j.api.injection.scope.This;
import org.qi4j.api.structure.Module;
import org.qi4j.api.type.ValueType;
import org.qi4j.io.Output;
import org.qi4j.io.Receiver;
import org.qi4j.io.Sender;
import org.qi4j.library.eventsourcing.domain.api.DomainEventValue;
import org.qi4j.library.eventsourcing.domain.api.UnitOfWorkDomainEventsValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static java.util.Collections.synchronizedList;

/**
 * Base implementation for EventStores.
 */
public abstract class AbstractEventStoreMixin
        implements EventStore, EventStream, EventStoreActivation
{
    @This
    protected Identity identity;

    protected Logger logger;
    protected ValueType domainEventType;
    protected ValueType eventsType;

    protected Lock lock = new ReentrantLock();

    @Structure
    protected Module module;

    private ExecutorService transactionNotifier;

    final private List<UnitOfWorkEventsListener> listeners = synchronizedList( new ArrayList<UnitOfWorkEventsListener>() );

    @Override
    public void activateEventStore() throws Exception
    {
        logger = LoggerFactory.getLogger( identity.identity().get() );

        domainEventType = module.valueDescriptor( DomainEventValue.class.getName() ).valueType();
        eventsType = module.valueDescriptor( UnitOfWorkDomainEventsValue.class.getName() ).valueType();

        transactionNotifier = Executors.newSingleThreadExecutor();
    }

    @Override
    public void passivateEventStore() throws Exception
    {
        transactionNotifier.shutdown();
        transactionNotifier.awaitTermination( 10000, TimeUnit.MILLISECONDS );
    }

    // UnitOfWorkEventsVisitor implementation
    // This is how transactions are put into the store


    @Override
    public Output<UnitOfWorkDomainEventsValue, IOException> storeEvents()
    {
        final Output<UnitOfWorkDomainEventsValue, IOException> storeOutput = storeEvents0();

        return new Output<UnitOfWorkDomainEventsValue, IOException>()
        {
            @Override
            public <SenderThrowableType extends Throwable> void receiveFrom( final Sender<? extends UnitOfWorkDomainEventsValue, SenderThrowableType> sender ) throws IOException, SenderThrowableType
            {
                final List<UnitOfWorkDomainEventsValue> events = new ArrayList<UnitOfWorkDomainEventsValue>(  );
                lock();
                try
                {
                    storeOutput.receiveFrom(new Sender<UnitOfWorkDomainEventsValue, SenderThrowableType>()
                    {
                        @Override
                        public <ReceiverThrowableType extends Throwable> void sendTo( final Receiver<? super UnitOfWorkDomainEventsValue, ReceiverThrowableType> receiver ) throws ReceiverThrowableType, SenderThrowableType
                        {
                            sender.sendTo( new Receiver<UnitOfWorkDomainEventsValue, ReceiverThrowableType>()
                            {
                                @Override
                                public void receive( UnitOfWorkDomainEventsValue item ) throws ReceiverThrowableType
                                {
                                    receiver.receive( item );
                                    events.add( item );
                                }
                            });
                        }
                    });

                } finally
                {
                    lock.unlock();
                }

                // Notify listeners
                transactionNotifier.submit( new Runnable()
                {
                    @Override
                    public void run()
                    {
                        synchronized(listeners)
                        {
                            for( UnitOfWorkEventsListener listener : listeners )
                            {
                                try
                                {
                                    listener.notifyTransactions( events );
                                } catch( Exception e )
                                {
                                    logger.warn( "Could not notify event listener", e );
                                }
                            }
                        }
                    }
                } );
            }
        };
    }

    // EventStream implementation
    @Override
    public void registerListener( UnitOfWorkEventsListener subscriber )
    {
        listeners.add( subscriber );
    }

    @Override
    public void unregisterListener( UnitOfWorkEventsListener subscriber )
    {
        listeners.remove( subscriber );
    }

    abstract protected Output<UnitOfWorkDomainEventsValue, IOException> storeEvents0();

    /**
     * Fix for this bug:
     * http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6822370
     */
    protected void lock()
    {
        while (true)
        {
            try
            {
                lock.tryLock( 1000, TimeUnit.MILLISECONDS );
                break;
            } catch (InterruptedException e)
            {
                // Try again
            }
        }
    }
}