public class SimpleMessageBus implements MessageBus {

    /**
     * List of command handlers per command
     */
    private Map<String, CommandHandler<? extends Command>> commandHandlers;

    /**
     * List of event handlers per event
     */
    private Map<String, List<Class<? extends EventHandler<? extends Event>>>> eventHandlers;

    /**
     * Constructor
     */
    private SimpleMessageBus(){
        commandHandlers = new HashMap<String, CommandHandler<? extends Command>>();
        eventHandlers = new HashMap<String, List<Class<? extends EventHandler<? extends Event>>>>();
    }

    /**
     * Get the singleton instance
     * @return
     */
    public static SimpleMessageBus getInstance(){
        return InstanceHolder.INSTANCE;
    }

    /**
     * Creates a new instance
     * 
     * @return
     */
    private static SimpleMessageBus create(){
        return new SimpleMessageBus();
    }

    /**
     * Class to contain the singleton
     */
    private static class InstanceHolder{
        public static final SimpleMessageBus INSTANCE = SimpleMessageBus.create();
    }

    @Override
    public <T extends Command> void registerCommandHandler(Class<T> aClass, CommandHandler<T> handler) {
        String key = aClass.getName();

        /*
         * Register only if not found. Commands *should*
         * only be registered to a single handler
         */
        if(!commandHandlers.containsKey(key)){
            commandHandlers.put(key, handler);
        }
    }

    @Override
    public <T extends Event, H extends EventHandler<T>> void registerEventHandler(Class<T> aClass, Class<H> handler) {

        String key = aClass.getName();

        if(!eventHandlers.containsKey(key)){
            eventHandlers.put(key, new ArrayList<Class<? extends EventHandler<? extends Event>>>());
        }

        eventHandlers.get(key).add(handler);
    }

    @Override
    public <T extends Event> void publish(T event) {
        publish(event, null);
    }

    /**
     * Publish an event asynchronously. Although the publish is
     * asynchronous it is still subject to the 60 second request time
     * limit. This is why the event handlers themselves are using
     * deferred tasks because they are not subject to the request time limit.
     * 
     * By executing this method asynchronously it allows the caller to continue 
     * without waiting for the individual event handlers to be added
     * to the queue for processing.
     * 
     */
    @Override
    public <T extends Event> void publish(T event, String queue) {

        ThreadExecutor executor = new ThreadExecutor();        
        executor.execute(new PublishTask(event,queue));
    }
    
    /**
     * Execute a command synchronously
     * 
     */
    @Override
    public <T extends Command> void send(T command) throws EventCollisionException, HydrationException, AggregateNotFoundException {

        if(command == null || commandHandlers.isEmpty())
            return;

        String key = command.getClass().getName();

        if(!commandHandlers.containsKey(key))
            return;

        CommandHandler<? extends Command> handlerForType = commandHandlers.get(key);

        @SuppressWarnings("unchecked")
        CommandHandler<T> handler = (CommandHandler<T>) handlerForType;
        handler.handle(command);
    }
    
    /**
     * Runnable task used to publish an event asynchronously.
     */
    private class PublishTask implements Runnable{

        private String queue;
        private Event event;
        
        public PublishTask(Event event, String queue){
            this.queue = queue;
            this.event = event;
        }
        
        @Override
        public void run() {
            if(event == null || eventHandlers.isEmpty())
                return;

            String key = event.getClass().getName();

            if(!eventHandlers.containsKey(key))
                return;

            List<Class<? extends EventHandler<? extends Event>>> handlersForType = eventHandlers.get(key);

            List<DeferredTask> taskList = new ArrayList<DeferredTask>();

            /*
             * Loop through all of the event handlers for the event being published 
             * and attempt to instantiate the handlers through reflection.
             */
            for(Class<? extends EventHandler<? extends Event>> handler : handlersForType){

                try{
                    /*
                     * get a constructor that expects a single parameter value that matches the event's type
                     */
                    Constructor<? extends EventHandler<? extends Event>> constructor = handler.getDeclaredConstructor(event.getClass());
                    constructor.setAccessible(true);
                    taskList.add(constructor.newInstance(event));

                } catch (NoSuchMethodException | SecurityException | InstantiationException | IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
                    /*
                     * unable to create an instance of the handler.
                     * more than likely this is caused by the handler
                     * not implementing a constructor that takes a single
                     * parameter instance of the event.
                     * 
                     *  There are several possibilities for this situation
                     *  - Publish a new event
                     *  - Log the exception
                     *  - Send an email to an administrator
                     *  - etc..
                     */
                }
            }

            if(taskList.isEmpty())
                return;

            Queue taskQueue = null;

            if(queue != null)
                taskQueue = QueueFactory.getQueue(queue);
            else
                taskQueue = QueueFactory.getDefaultQueue();

            for(DeferredTask task : taskList){
                taskQueue.addAsync(TaskOptions.Builder.withPayload(task));
            }
        }
    }
} public class ChangeAttendeeEmail implements Command {

	private UUID attendeeId = null;
	private String email = null;
    
    /**
     * Default constructor for serialization
     */
    public ChangeAttendeeEmail(){
    }

    /**
     * Constructor
     * 
     * @param firstName
     * @param lastName
     */
    public ChangeAttendeeEmail(UUID attendeeId, String email){
        this.attendeeId = attendeeId;
        this.email = email;
    }

    /**
     * Get the attendee the Id
     * 
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }

    /**
     * Get the email
     * 
     * @return
     */
    public String getEmail(){
        return email;
    }  
} public class ChangeAttendeeName implements Command{
    private String firstName = null;
    private String lastName = null;
    private UUID attendeeId = null;

    /**
     * Default constructor for serialization
     */
    public ChangeAttendeeName(){
    }

    /**
     * Constructor
     * 
     * @param firstName
     * @param lastName
     */
    public ChangeAttendeeName(UUID attendeeId, String firstName, String lastName){
        this.attendeeId = attendeeId;
        this.firstName = firstName;
        this.lastName = lastName;
    }

    /**
     * Get the attendee the Id
     * 
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }

    /**
     * Get the first name
     * 
     * @return
     */
    public String getFirstName(){
        return firstName;
    }

    /**
     * Get the last name
     * 
     * @return
     */
    public String getLastName(){
        return lastName;
    }
} public class ConfirmChangeEmail implements Command {

	private UUID attendeeId;
    private UUID confirmationId;
    
    
    /**
     * Default constructor for serialization
     */
    public ConfirmChangeEmail(){
    }
    
    /**
     * Constructor
     * 
     * @param attendeeId
     * @param confirmationId
     */
    public ConfirmChangeEmail(UUID attendeeId, UUID confirmationId){
    	this.attendeeId = attendeeId;
    	this.confirmationId = confirmationId;
    }
    

    /**
     * Get the confirmation id to verify the email change
     * 
     * @return
     */
    public UUID getConfirmationId(){
    	return confirmationId;
    }

    /**
     * Get the attendee Id
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }
} public class RegisterAttendee implements Command {

    private String email = null;
    private String firstName = null;
    private String lastName = null;
    private UUID attendeeId = null;

    /**
     * Default constructor for serialization
     */
    public RegisterAttendee(){
    }

    /**
     * Constructor
     * 
     * @param firstName
     * @param lastName
     */
    public RegisterAttendee(UUID attendeeId, String email, String firstName, String lastName){
        this.attendeeId = attendeeId;
        this.email = email;
        this.firstName = firstName;
        this.lastName = lastName;
    }

    /**
     * Get the attendee Id
     * 
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }

    /**
     * Get the first name
     * 
     * @return
     */
    public String getFirstName(){
        return firstName;
    }

    /**
     * Get the last name
     * 
     * @return
     */
    public String getLastName(){
        return lastName;
    }

    /**
     * Get the email
     * @return
     */
    public String getEmail(){
        return email;
    }
} public class ResolveDuplicateEmail implements Command{

    private UUID attendeeId;

    /**
     * Default constructor
     * 
     * @param attendeeId
     */
    public ResolveDuplicateEmail(UUID attendeeId){
        this.attendeeId = attendeeId;
    }

    /**
     * Get the attendee Id
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }
} public class AttendeeChangeEmailConfirmed implements Event, Serializable {

    private static final long serialVersionUID = 1L;

    private UUID attendeeId;
    private UUID confirmationId;
    private String email;
    
    /**
     * Default constructor for serialization
     */
    public AttendeeChangeEmailConfirmed(){}
    
    /**
     * Constructor
     * 
     * @param attendeeId
     * @param confirmationId
     * @param email
     */
    public AttendeeChangeEmailConfirmed(UUID attendeeId, UUID confirmationId, String email){
    	this.attendeeId = attendeeId;
    	this.confirmationId = confirmationId;
    	this.email = email;
    }
    
    /**
     * Get the email
     * @return
     */
    public String getEmail(){
        return  email;
    }
    
    /**
     * Get the confirmation id to verify the email change
     * 
     * @return
     */
    public UUID getConfirmationId(){
    	return confirmationId;
    }

    /**
     * Get the attendee Id
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }
} public class AttendeeDisabled implements Event, Serializable {

    private static final long serialVersionUID = 1L;

    private UUID attendeeId;
    private DisableReason reason;

    /**
     * Default constructor for serialization
     */
    public AttendeeDisabled(){}

    /**
     * Constructor
     * 
     * @param attendeeId
     * @param reason
     */
    public AttendeeDisabled(UUID attendeeId, DisableReason reason){
        this.attendeeId = attendeeId;
        this.reason = reason;
    }

    /**
     * Get the attendee Id
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }

    /**
     * Get the reason
     * 
     * @return
     */
    public DisableReason getReason(){
        return reason;
    }
} public class AttendeeEmailChanged implements Event, Serializable {

	private static final long serialVersionUID = 1L;

	private UUID attendeeId;
	private UUID confirmationId;
    private String email;

    /**
     * Default constructor for serialization
     */
    public AttendeeEmailChanged(){}

    /**
     * Constructor
     * 
     * @param attendeeId
     * @param firstName
     * @param lastName
     */
    public AttendeeEmailChanged(UUID attendeeId, String email){
        this.attendeeId = attendeeId;
        this.email = email;
        this.confirmationId = UUID.randomUUID();
    }

    /**
     * Get the email
     * @return
     */
    public String getEmail(){
        return  email;
    }
    
    /**
     * Get the confirmation id to verify the email change
     * 
     * @return
     */
    public UUID getConfirmationId(){
    	return confirmationId;
    }

    /**
     * Get the attendee Id
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }
} public class AttendeeNameChanged implements Event, Serializable{

    private static final long serialVersionUID = 1L;

    private UUID attendeeId;
    private String firstName;
    private String lastName;

    /**
     * Default constructor for serialization
     */
    public AttendeeNameChanged(){}

    /**
     * Constructor
     * 
     * @param attendeeId
     * @param firstName
     * @param lastName
     */
    public AttendeeNameChanged(UUID attendeeId, String firstName, String lastName){
        this.attendeeId = attendeeId;
        this.firstName = firstName;
        this.lastName = lastName;
    }

    /**
     * Get the first name
     * @return
     */
    public String getFirstName(){
        return  firstName;
    }

    /**
     * Get the last name
     * @return
     */
    public String getLastName(){
        return lastName;
    }

    /**
     * Get the attendee Id
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }
} public class AttendeeRegistered implements Event, Serializable {

    private static final long serialVersionUID = 1L;

    private UUID attendeeId;
    private String email;
    private String firstName;
    private String lastName;

    /**
     * Constructor for serialization
     */
    public AttendeeRegistered(){}

    /**
     * Constructor
     * 
     * @param attendeeId
     * @param email
     * @param firstName
     * @param lastName
     */
    public AttendeeRegistered(UUID attendeeId, String email, String firstName, String lastName){
        this.attendeeId = attendeeId;
        this.email = email;
        this.firstName = firstName;
        this.lastName = lastName;
    }

    /**
     * Get the first name
     * 
     * @return
     */
    public String getFirstName(){
        return  firstName;
    }

    /**
     * Get the last name
     * 
     * @return
     */
    public String getLastName(){
        return lastName;
    }

    /**
     * Get the attendee Id
     * 
     * @return
     */
    public UUID getAttendeeId(){
        return attendeeId;
    }

    /**
     * Get the email
     * 
     * @return
     */
    public String getEmail(){
        return email;
    }
} public class ChangeAttendeeEmailCommandHandler implements CommandHandler<ChangeAttendeeEmail> {

	@Override
	public void handle(ChangeAttendeeEmail command) throws EventCollisionException, HydrationException, AggregateNotFoundException {

		Repository<Attendee> repository = new EventRepository<Attendee>(Attendee.class, Constants.DOMAIN_EVENTS_PROCESSING_QUEUE);

        Attendee attendee = repository.getById(command.getAttendeeId());
        attendee.changeEmail(command.getEmail());
        repository.save(attendee);
		
	}

} public class ChangeAttendeeNameCommandHandler implements CommandHandler<ChangeAttendeeName> {

    @Override
    public void handle(ChangeAttendeeName command) throws EventCollisionException, HydrationException, AggregateNotFoundException {

        Repository<Attendee> repository = new EventRepository<Attendee>(Attendee.class, Constants.DOMAIN_EVENTS_PROCESSING_QUEUE);

        Attendee attendee = repository.getById(command.getAttendeeId());
        attendee.changeName(command.getFirstName(), command.getLastName());
        repository.save(attendee);
    }
} public class ConfirmChangeEmailCommandHandler implements CommandHandler<ConfirmChangeEmail>{

	@Override
	public void handle(ConfirmChangeEmail command) throws EventCollisionException, HydrationException, AggregateNotFoundException {
		
		Repository<Attendee> repository = new EventRepository<Attendee>(Attendee.class, Constants.DOMAIN_EVENTS_PROCESSING_QUEUE);

        Attendee attendee = repository.getById(command.getAttendeeId());
        attendee.confirmChangeEmail(command.getConfirmationId());
        repository.save(attendee);
		
	}

} public class RegisterAttendeeCommandHandler implements CommandHandler<RegisterAttendee> {

    @Override
    public void handle(RegisterAttendee command) throws EventCollisionException, HydrationException {

        Repository<Attendee> repository = new EventRepository<Attendee>(Attendee.class, Constants.DOMAIN_EVENTS_PROCESSING_QUEUE);

        Attendee attendee = null;
        
		try {
			attendee = repository.getById(command.getAttendeeId());
		} catch (AggregateNotFoundException e) {
			//Do nothing. If it isn't found continue to process
		}
        
        //if the aggregate Id already exists throw an exception
        if(attendee != null) throw new EventCollisionException(command.getAttendeeId(), 0, "Attendee Id already exists.");
        
        attendee = Attendee.create(command.getAttendeeId(), command.getEmail(), command.getFirstName(), command.getLastName());
        repository.save(attendee);
        
    }
} public class ResolveDuplicateEmailCommandHandler implements CommandHandler<ResolveDuplicateEmail>{

    @Override
    public void handle(ResolveDuplicateEmail command) throws EventCollisionException, HydrationException, AggregateNotFoundException {

        Repository<Attendee> repository = new EventRepository<Attendee>(Attendee.class, Constants.DOMAIN_EVENTS_PROCESSING_QUEUE);

        Attendee attendee = repository.getById(command.getAttendeeId());
        attendee.disable(DisableReason.DUPLICATE);
        repository.save(attendee);

    }
} public class AttendeeChangeEmailConfirmedEventHandler extends EventHandler<AttendeeChangeEmailConfirmed> {

	private static final long serialVersionUID = 1L;

	public AttendeeChangeEmailConfirmedEventHandler(AttendeeChangeEmailConfirmed event) {
		super(event);
	}

	@Override
	public void run() {
		
		DatastoreService datastoreService = DatastoreServiceFactory.getDatastoreService();
        Transaction transaction = datastoreService.beginTransaction();

        Key attendeeKey = KeyFactory.createKey("Attendee", event.getAttendeeId().toString());
        Entity attendee = null;

        try {
            attendee = datastoreService.get(attendeeKey);
            attendee.setProperty("Email", event.getEmail());
            datastoreService.put(attendee);
            transaction.commit();

        } catch (EntityNotFoundException e) {
            MessageLog.log(e);
        } finally {
            if(transaction != null && transaction.isActive())
                transaction.rollback();
        }
		
	}

} public class AttendeeDisabledEventHandler extends EventHandler<AttendeeDisabled> {

    private static final long serialVersionUID = 1L;

    public AttendeeDisabledEventHandler(AttendeeDisabled event) {
        super(event);
    }

    @Override
    public void run() {

        MessageLog.log(String.format("Attendee %s Disabled.", event.getAttendeeId().toString()));

        //here is where you may send an email to the attendee notifying them that they have been disabled.
    }
} public class AttendeeEmailChangedEventHandler extends EventHandler<AttendeeEmailChanged>{

	private static final long serialVersionUID = 1L;

	public AttendeeEmailChangedEventHandler(AttendeeEmailChanged event) {
		super(event);
		// TODO Auto-generated constructor stub
	}

	@Override
    public void run() {

        MessageLog.log(String.format("Attendee Id %s requested to change email address to : %s.", event.getAttendeeId().toString(), event.getEmail()));

        //here is where you may send an email to the attendee with a link to "confirm" the change.
    }
} public class AttendeeNameChangedEventHandler extends EventHandler<AttendeeNameChanged>{

    private static final long serialVersionUID = 1L;

    public AttendeeNameChangedEventHandler(AttendeeNameChanged event){
        super(event);
    }

    @Override
    public void run() {

        DatastoreService datastoreService = DatastoreServiceFactory.getDatastoreService();
        Transaction transaction = datastoreService.beginTransaction();

        Key attendeeKey = KeyFactory.createKey("Attendee", event.getAttendeeId().toString());
        Entity attendee = null;

        try {
            attendee = datastoreService.get(attendeeKey);
            attendee.setProperty("FirstName", event.getFirstName());
            attendee.setProperty("LastName", event.getLastName());
            datastoreService.put(attendee);
            transaction.commit();

        } catch (EntityNotFoundException e) {
            MessageLog.log(e);
        } finally {
            if(transaction != null && transaction.isActive())
                transaction.rollback();
        }
    }
} public class AttendeeRegisteredEventHandler extends EventHandler<AttendeeRegistered>{

    private static final long serialVersionUID = 1L;

    public AttendeeRegisteredEventHandler(AttendeeRegistered event){
        super(event);
    }

    @Override
    public void run() {

        DatastoreService datastoreService = DatastoreServiceFactory.getDatastoreService();

        /*
         * Run a query against the read store to see if the email has already been registered
         */
        Filter emailFilter = new FilterPredicate("Email", FilterOperator.EQUAL, event.getEmail());

        Query attendeeQuery = new Query("Attendee")
        .setFilter(emailFilter)
        .setKeysOnly();

        PreparedQuery preparedQuery = datastoreService.prepare(attendeeQuery);
        List<Entity> matchingEmails = preparedQuery.asList(FetchOptions.Builder.withDefaults());

        /*
         * Create an attendee key based upon the UUID
         */
        Key attendeeKey = KeyFactory.createKey("Attendee", event.getAttendeeId().toString());
        Entity attendee = null;

        Transaction transaction = datastoreService.beginTransaction();

        try {
            attendee = datastoreService.get(attendeeKey);

            //do nothing if found. This is a new registration

        } catch (EntityNotFoundException e) {

            boolean duplicateExists = matchingEmails != null && !matchingEmails.isEmpty();

            attendee = new Entity(attendeeKey);
            attendee.setProperty("FirstName", event.getFirstName());
            attendee.setProperty("LastName", event.getLastName());
            attendee.setProperty("Email", event.getEmail());
            attendee.setProperty("IsEnabled", !duplicateExists);
            datastoreService.put(attendee);
            transaction.commit();

            /*
             * If a match is found. Send a compensating command.
             * You'll notice we're still allowing the read model to 
             * be populated
             */
            if(duplicateExists){
                try {
                    SimpleMessageBus.getInstance().send(new ResolveDuplicateEmail(event.getAttendeeId()));
                } catch (EventCollisionException | HydrationException | AggregateNotFoundException commandException) {
                    MessageLog.log(commandException);
                    //may want to publish message to administrator
                }
            }

        } finally {
            if(transaction != null && transaction.isActive())
                transaction.rollback();
        }
    }
}
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import com.cqrs.appengine.core.messaging.SimpleMessageBus;
import com.cqrs.appengine.sample.commands.ChangeAttendeeEmail;
import com.cqrs.appengine.sample.commands.ChangeAttendeeName;
import com.cqrs.appengine.sample.commands.ConfirmChangeEmail;
import com.cqrs.appengine.sample.commands.RegisterAttendee;
import com.cqrs.appengine.sample.commands.ResolveDuplicateEmail;
import com.cqrs.appengine.sample.domain.AttendeeChangeEmailConfirmed;
import com.cqrs.appengine.sample.domain.AttendeeDisabled;
import com.cqrs.appengine.sample.domain.AttendeeEmailChanged;
import com.cqrs.appengine.sample.domain.AttendeeNameChanged;
import com.cqrs.appengine.sample.domain.AttendeeRegistered;
import com.cqrs.appengine.sample.handlers.commands.ChangeAttendeeEmailCommandHandler;
import com.cqrs.appengine.sample.handlers.commands.ChangeAttendeeNameCommandHandler;
import com.cqrs.appengine.sample.handlers.commands.ConfirmChangeEmailCommandHandler;
import com.cqrs.appengine.sample.handlers.commands.RegisterAttendeeCommandHandler;
import com.cqrs.appengine.sample.handlers.commands.ResolveDuplicateEmailCommandHandler;
import com.cqrs.appengine.sample.handlers.events.AttendeeChangeEmailConfirmedEventHandler;
import com.cqrs.appengine.sample.handlers.events.AttendeeDisabledEventHandler;
import com.cqrs.appengine.sample.handlers.events.AttendeeEmailChangedEventHandler;
import com.cqrs.appengine.sample.handlers.events.AttendeeNameChangedEventHandler;
import com.cqrs.appengine.sample.handlers.events.AttendeeRegisteredEventHandler;
package com.cqrs.appengine.sample;



/**
 * Custom listener to handle command/event registration with the message
 * bus during warm-up
 */
public class ApplicationContextListener implements ServletContextListener {

    @Override
    public void contextDestroyed(ServletContextEvent event) {
        // App Engine does not currently invoke this method.	
    }

    @Override
    public void contextInitialized(ServletContextEvent event) {

        SimpleMessageBus.getInstance().registerCommandHandler(RegisterAttendee.class, new RegisterAttendeeCommandHandler());
        SimpleMessageBus.getInstance().registerCommandHandler(ChangeAttendeeName.class, new ChangeAttendeeNameCommandHandler());