public interface Message {
	public Map<String, String> getHeaders();
	public String getId();
	public String getQueue();
	public String getBody();
} public interface MessageHandler {
	public void handle(Message message);
	public void handleBatch(List<Message> messages);
} public class AwsConfig {
	
	private String region;
	private String accessKey;
	private String secretKey;
	private DynamoDbConfig dynamo;
	private SqsConfig sqs;
	private List<KinesisConfig> kinesis;
	private S3Config s3;
	
	
	public AwsConfig(){}
	
	
	@JsonProperty("region")
	public String getRegion() {
		return region;
	}
	@JsonProperty("region")
	public void setRegion(String region) {
		this.region = region;
	}
	@JsonProperty("accessKey")
	public String getAccessKey() {
		return accessKey;
	}
	@JsonProperty("accessKey")
	public void setAccessKey(String accessKey) {
		this.accessKey = accessKey;
	}
	@JsonProperty("secretKey")
	public String getSecretKey() {
		return secretKey;
	}
	@JsonProperty("secretKey")
	public void setSecretKey(String secretKey) {
		this.secretKey = secretKey;
	}
	@JsonProperty("s3")
	public S3Config getS3() {
		return s3;
	}
	@JsonProperty("s3")
	public void setS3(S3Config s3) {
		this.s3 = s3;
	}
	@JsonProperty("dynamo")
	public DynamoDbConfig getDynamo() {
		return dynamo;
	}
	@JsonProperty("dynamo")
	public void setDynamo(DynamoDbConfig dynamo) {
		this.dynamo = dynamo;
	}
	@JsonProperty("kinesis")
	public List<KinesisConfig> getKinesis() {
		return kinesis;
	}
	@JsonProperty("kinesis")
	public void setKinesis(List<KinesisConfig> kinesis) {
		this.kinesis = kinesis;
	}
	@JsonProperty("sqs")
	public SqsConfig getSqs() {
		return sqs;
	}
	@JsonProperty("sqs")
	public void setSqs(SqsConfig sqs) {
		this.sqs = sqs;
	}
	
	
	public static class AwsConfigBuilder{
		private String region;
		private String accessKey;
		private String secretKey;
		private DynamoDbConfig dynamo;
		private SqsConfig sqs;
		private List<KinesisConfig> kinesis;
		private S3Config s3;

		public AwsConfigBuilder withRegion(final String region){
			this.region = region;
			return this;
		}
		
		public AwsConfigBuilder withAccessKey(final String accessKey){
			this.accessKey = accessKey;
			return this;
		}
		
		public AwsConfigBuilder withSecretKey(final String secretKey){
			this.secretKey = secretKey;
			return this;
		}
		
		public AwsConfigBuilder withDynamo(final DynamoDbConfig dynamo){
			this.dynamo = dynamo;
			return this;
		}
		
		
		public AwsConfigBuilder withSqs(final SqsConfig sqs){
			this.sqs = sqs;
			return this;
		}
		
		
		public AwsConfigBuilder withKinesis(final KinesisConfig ...kinesis){
			if(this.kinesis==null){
				this.kinesis = Lists.newArrayList(kinesis);
			}else{
				this.kinesis.addAll(Lists.newArrayList(kinesis));
			}
			return this;
		}
		
		public AwsConfigBuilder withS3(final S3Config s3){
			this.s3 = s3;
			return this;
		}
		
		
		public AwsConfig build(){
			final AwsConfig config = new AwsConfig();
			config.setRegion(this.region);
			config.setAccessKey(this.accessKey);
			config.setSecretKey(this.secretKey);
			config.setDynamo(this.dynamo);
			config.setKinesis(this.kinesis);
			config.setSqs(this.sqs);
			config.setS3(this.s3);
			return config;
		}
	}
} public class DynamoDbConfig {

	private String scanPackage;
	private boolean blockUntilReady = true;

	/** TODO - Remove this **/
	private List<Table> tables = new ArrayList<Table>();

	@JsonProperty(value="blockUntilReady")
	public boolean isBlockUntilReady() {
		return blockUntilReady;
	}
	@JsonProperty(value="blockUntilReady")
	public void setBlockUntilReady(boolean blockUntilReady) {
		this.blockUntilReady = blockUntilReady;
	}
	
	
	public DynamoDbConfig withScanPackage(final String scanPackage){
		this.scanPackage = scanPackage;
		return this;
	}
	public DynamoDbConfig withBlockingUntilReady(final boolean block){
		this.blockUntilReady = block;
		return this;
	}
	
	
	/***
	 * DynamoDb Table Configuration
	 * @author shawn
	 *
	 */
	public static class Table{
		
		private String name;
		private List<Key> keys = new ArrayList<Key>();
		private Long writeCapacityUnits;
		private Long readCapacityUnits;
		
		private List<AttributeDefinition> attributeDefinitions = new ArrayList<AttributeDefinition>();
		private List<LocalSecondaryIndex> localSecondaryIndexes = new ArrayList<LocalSecondaryIndex>();
		private List<GlobalSecondaryIndex> globalSecondaryIndexes = new ArrayList<GlobalSecondaryIndex>();

		
		@JsonProperty("name")
		public String getName() {
			return name;
		}
		@JsonProperty("name")
		public void setName(String name) {
			this.name = name;
		}

		/** Key Schema Elements, required for any attribute that is part of an index or primary key **/
		@JsonProperty("keys")
		public List<Key> getKeys() {
			return keys;
		}
		@JsonProperty("keys")
		public void setKeys(List<Key> keys) {
			this.keys = keys;
		}
		
		
		@JsonProperty("readCapacityUnits")
		/** Read Write capacity units **/
		public Long getReadCapacityUnits() {
			return readCapacityUnits;
		}
		@JsonProperty("readCapacityUnits")
		public void setReadCapacityUnits(Long readCapacityUnits) {
			this.readCapacityUnits = readCapacityUnits;
		}
		
		@JsonProperty("writeCapacityUnits")
		public Long getWriteCapacityUnits() {
			return writeCapacityUnits;
		}
		@JsonProperty("writeCapacityUnits")
		public void setWriteCapacityUnits(Long writeCapacityUnits) {
			this.writeCapacityUnits = writeCapacityUnits;
		}
		
	
		
		/** Attribute Definitions required for any attribute that is part of an index or primary key **/
		@JsonProperty("attributeDefinitions")
		public List<AttributeDefinition> getAttributeDefinitions() {
			return attributeDefinitions;
		}
		@JsonProperty("attributeDefinitions")
		public void setAttributeDefinitions(List<AttributeDefinition> attributeDefinitions) {
			this.attributeDefinitions = attributeDefinitions;
		}
		
		
		@JsonProperty("localSecondaryIndexes")
		public List<LocalSecondaryIndex> getLocalSecondaryIndexes() {
			return localSecondaryIndexes;
		}
		@JsonProperty("localSecondaryIndexes")
		public void setLocalSecondaryIndexes(List<LocalSecondaryIndex> localSecondaryIndexes) {
			this.localSecondaryIndexes = localSecondaryIndexes;
		}
		
		@JsonProperty("globalSecondaryIndexes")
		public List<GlobalSecondaryIndex> getGlobalSecondaryIndexes() {
			return globalSecondaryIndexes;
		}
		@JsonProperty("globalSecondaryIndexes")
		public void setGlobalSecondaryIndexes(List<GlobalSecondaryIndex> globalSecondaryIndexes) {
			this.globalSecondaryIndexes = globalSecondaryIndexes;
		}
	}
	

	
	
	/***
	 * Local Secondary Index Configuration
	 * @author shawn
	 */
	public static class LocalSecondaryIndex extends SecondaryIndex {
		public LocalSecondaryIndex(){}
	}


	/***
	 * Global Secondary Index Configuration
	 * @author shawn
	 */
	public static class GlobalSecondaryIndex extends SecondaryIndex{
		private Long readCapacityUnits;
		private Long writeCapacityUnits;
		
		public GlobalSecondaryIndex(){}
		
		@JsonProperty("readCapacityUnits")
		public Long getReadCapacityUnits() {
			return readCapacityUnits;
		}
		@JsonProperty("readCapacityUnits")
		public void setReadCapacityUnits(Long readCapacityUnits) {
			this.readCapacityUnits = readCapacityUnits;
		}
		
		@JsonProperty("writeCapacityUnits")
		public Long getWriteCapacityUnits() {
			return writeCapacityUnits;
		}
		@JsonProperty("writeCapacityUnits")
		public void setWriteCapacityUnits(Long writeCapacityUnits) {
			this.writeCapacityUnits = writeCapacityUnits;
		}
	}


	/***
	 * Common Attributes for all Secondary Indexes
	 * @author shawn
	 */
	public static class SecondaryIndex{ 
		
		private String name;
		private Projection projection; 
		private List<Key> keys = new ArrayList<Key>();
		
		protected SecondaryIndex(){}
		
		@JsonProperty("name")
		public String getName() {
			return name;
		}
		@JsonProperty("name")
		public void setName(String name) {
			this.name = name;
		}
		
		@JsonProperty("projection")
		public Projection getProjection() {
			return projection;
		}@JsonProperty("projection")
		public void setProjection(Projection projection) {
			this.projection = projection;
		}
		
		@JsonProperty("keys")
		public List<Key> getKeys() {
			return keys;
		}
		@JsonProperty("keys")
		public void setKeys(List<Key> keys) {
			this.keys = keys;
		}
	}
	
	
	/***
	 * Projection Configuration determines  which attributes are copied over with a given index
	 * @author shawn
	 *
	 */
	public static class Projection{
		private String type; // Valid Values are ALL, KEYS_ONLY, INCLUDE
		private List<String> nonKeyAttributes;
	
		public Projection(){}
		
		@JsonProperty("type")
		public String getType() {
			return type;
		}
		@JsonProperty("type")
		public void setType(String type) {
			this.type = type;
		}
		
		@JsonProperty("nonKeyAttributes")
		public List<String> getNonKeyAttributes() {
			return nonKeyAttributes;
		}
		@JsonProperty("nonKeyAttributes")
		public void setNonKeyAttributes(List<String> nonKeyAttributes) {
			this.nonKeyAttributes = nonKeyAttributes;
		}
	}
	
	

	/**
	 * Part of a primary key or index
	 * @author shawn
	 */
	public static class Key{
		public Key(){}
		private String attributeName;
		private String keyType;
		
		public Key(final String attributeName){
			this.attributeName = attributeName;
		}
		
		@JsonProperty("attributeName")
		public String getAttributeName() {
			return attributeName;
		}
		@JsonProperty("attributeName")
		public void setName(String attributeName) {
			this.attributeName = attributeName;
		}
		@JsonProperty("keyType")
		public String getKeyType() {
			return this.keyType;
		}
		@JsonProperty("keyType")
		public void setKeyType(final String keyType){
			this.keyType = keyType;
		}
	}
	
	

	
	/***
	 * An Attribute Definition for a Key Schema Element
	 * @author shawn
	 */
	public static class AttributeDefinition{
		private String name;
		private String type; // Valid Values are S, B, N
		
		@JsonProperty("name")
		public String getName() {
			return name;
		}
		@JsonProperty("name")
		public void setName(String name) {
			this.name = name;
		}
		
		@JsonProperty("type")
		public String getType() {
			return type;
		}
		@JsonProperty("type")
		public void setType(String type) {
			this.type = type;
		}
	}
	@JsonProperty("tables")
	public List<Table> getTables() {
		return tables;
	}
	@JsonProperty("tables")
	public void setTables(List<Table> tables) {
		this.tables = tables;
	}
	@JsonProperty("scanPackage")
	public String getPackagePrefix() {
		return scanPackage;
	}
	@JsonProperty("scanPackage")
	public void setPackagePrefix(String scanPackage) {
		this.scanPackage = scanPackage;
	}
} public class KinesisConfig {
	
	private String topic;
	private int shards;
	
	public KinesisConfig(){}
	
	
	@JsonProperty("topic")
	public String getTopic() {
		return topic;
	}
	@JsonProperty("topic")
	public void setTopic(String topic) {
		this.topic = topic;
	}
	public KinesisConfig withTopic(final String topic){
		this.topic = topic;
		return this;
	}
	
	
	@JsonProperty("shards")
	public int getShards() {
		return shards;
	}
	@JsonProperty("shards")
	public void setShards(int shards) {
		this.shards = shards;
	}
	public KinesisConfig withShards(final int shards){
		this.shards = shards;
		return this;
	}
} public class S3Config {
	
	private String defaultDrive;
	private List<String> buckets = new ArrayList<String>();

	/**
	 * get the default bucket to use if no bucket is specified
	 * @return bucketName
	 */
	@JsonProperty("defaultBucket")
	public String getDefaultDrive() {
		return defaultDrive;
	}
	/**
	 * Set the default bucket name to use in cases where no bucket is specified
	 * @param drive
	 */
	@JsonProperty("defaultBucket")
	public void setDefaultDrive(final String drive){
		this.defaultDrive = drive;
	}
	@JsonProperty("buckets")
	public List<String> getBuckets() {
		return buckets;
	}
	@JsonProperty("buckets")
	public void setBuckets(List<String> buckets) {
		this.buckets = buckets;
	}
	
	
	public S3Config withDefaultBucket(final String bucket){
		this.defaultDrive = bucket;
		return this;
	}
	public S3Config withBuckets(final String ... buckets){
		this.buckets.addAll(Lists.newArrayList(buckets));
		return this;
	}
} public class SqsConfig {

	private boolean produces = true;
	private boolean blockUntilReady = true;
	private List<SqsQueueConfig> queues = new ArrayList<SqsQueueConfig>();
	private List<SqsConsumerConfig> consumers = new ArrayList<SqsConsumerConfig>();
	
	@JsonProperty(value="queues")
	public List<SqsQueueConfig> getQueues() {
		return queues;
	}
	@JsonProperty(value="queues")
	public void setQueues(List<SqsQueueConfig> queues) {
		this.queues = queues;
	}
	@JsonProperty(value="consumers")
	public List<SqsConsumerConfig> getConsumers() {
		return consumers;
	}
	@JsonProperty(value="consumers")
	public void setConsumers(List<SqsConsumerConfig> consumers) {
		this.consumers = consumers;
	}
	@JsonProperty(value="produces")
	public boolean isProduces() {
		return produces;
	}
	@JsonProperty(value="produces")
	public void setProduces(boolean produces) {
		this.produces = produces;
	}
	@JsonProperty(value="blockUntilReady")
	public boolean isBlockUntilReady() {
		return blockUntilReady;
	}
	@JsonProperty(value="blockUntilReady")
	public void setBlockUntilReady(boolean blockUntilReady) {
		this.blockUntilReady = blockUntilReady;
	}
	
	
	
	public SqsConfig withProduces(boolean producers){
		this.produces = producers;
		return this;
	}
	public SqsConfig withBlockUntilReader(boolean block){
		this.blockUntilReady = block;
		return this;
	}
	public SqsConfig withQueues(SqsQueueConfig ...queues){
		this.queues.addAll(Lists.newArrayList(queues));
		return this;
	}
	public SqsConfig withConsumers(SqsConsumerConfig ...consumers){
		this.consumers.addAll(Lists.newArrayList(consumers));
		return this;
	}
} public class SqsConsumerConfig{
	private int threads = 1; 	//Default to a thread count of one for consumers
	private String queue;
	private String handlerKey;
	private int shutdownTimeMS = 2000;
	private MessageHandler handlerInstance;
	
	@JsonProperty("queue")
	public String getQueue() {
		return queue;
	}
	@JsonProperty("queue")
	public void setQueue(final String queue) {
		this.queue = queue;
	}
	public SqsConsumerConfig withQueue(final String queue) {
		this.queue = queue;
		return this;
	}
	
	@JsonProperty("threads")
	public int getThreads() {
		return threads;
	}
	@JsonProperty("threads")
	public void setThreads(final int threads) {
		this.threads = threads;
	}
	public SqsConsumerConfig withThreads(final int threads) {
		this.threads = threads;
		return this;
	}
	
	
	@JsonProperty("handler")
	public String getHandlerKey() {
		return handlerKey;
	}
	@JsonProperty("handler")
	public void setHandlerKey(String handler) {
		this.handlerKey = handler;
	}

	
	
	@JsonProperty("shutdownWaitMs")
	public int getShutdownTimeMS() {
		return shutdownTimeMS;
	}
	@JsonProperty("shutdownWaitMs")
	public void setShutdownTimeMS(int shutdownTimeMS) {
		this.shutdownTimeMS = shutdownTimeMS;
	}
	public SqsConsumerConfig withHandler(final int shutdownTimeMS) {
		this.shutdownTimeMS = shutdownTimeMS;
		return this;
	}
	
	public MessageHandler getHandlerInstance() {
		return handlerInstance;
	}
	public void setHandlerInstance(MessageHandler handlerInstance) {
		this.handlerInstance = handlerInstance;
	}
	public SqsConsumerConfig withHandler(final MessageHandler handler) {
		this.handlerInstance = handler;
		return this;
	}
	
	/***
	 * 
	 * @throws ClassNotFoundException
	 */
	public MessageHandler createHandler() throws ClassNotFoundException{
		if(this.handlerInstance==null){
			@SuppressWarnings("unchecked")
			Class<MessageHandler> handler = (Class<MessageHandler>)Class.forName(handlerKey);
			try{
				this.handlerInstance = handler.newInstance();	
			}catch(Exception ex){
				throw new RuntimeException(ex.getMessage(), ex);
			}	
		}
		return this.handlerInstance;
	}
	
	
	
	/***
	 * Makes a new copy of the configuration
	 * @return copiedConfig
	 */
	public SqsConsumerConfig copy() {
		final SqsConsumerConfig config = new SqsConsumerConfig();
		config.setHandlerInstance(this.handlerInstance);
		config.setHandlerKey(this.handlerKey);
		config.setQueue(this.queue);
		config.setThreads(this.threads);
		config.setShutdownTimeMS(this.shutdownTimeMS);
		return config;
	}
} public class SqsQueueConfig {
	
	/** The queue name **/
	private String name;
	
	
	/** If blockUntilQueue is ready in queue admin, the create queue process will wait this amount of time for the queue to become ready **/
	private int queueCreationTimeoutMS = 60000;
	
	
	/***
	 * The time in seconds that the delivery
     * of all messages in the queue will be delayed. An integer from 0 to 900
     * (15 minutes). The default for this attribute is 0 (zero).
	 */
	private int delaySeconds;
	public final static String DELAY_SECONDS = "DelaySeconds"; 
	
	
	
	/***
	 *  The limit of how many bytes a
     * message can contain before Amazon SQS rejects it. An integer from 1024
     * bytes (1 KiB) up to 262144 bytes (256 KiB). The default for this
     * attribute is 262144 (256 KiB)
	 */
	private int maximumMessageSize; 
	public final static String MAXIMUM_MESSAGE_SIZE = "MaximumMessageSize"; 
	
	
	
	/***
	 * - The number of seconds Amazon SQS retains a message. 
	 *   Integer representing seconds, from 60 (1minute) to 1209600 (14 days). 
	 *   The default for this attribute is 345600 (4 days).
	 */
	private int messageRetentionPeriod; 
	public final static String MESSAGE_RETENTION_PERIOD = "MessageRetentionPeriod"; 
	
	
	/***
	 * The default visibility timeout for
     * the queue. An integer from 0 to 43200 (12 hours). The default for this
     * attribute is 30.
	 */
	private int visibilityTimeout;
	public final static String VISIBILITY_TIMEOUT = "VisibilityTimeout"; 
	
	

	@JsonProperty("name")
	public String getName() {
		return name;
	}
	@JsonProperty("name")
	public void setName(String name) {
		this.name = name;
	}
	public SqsQueueConfig withName(final String name){
		this.name = name;
		return this;
	}
	
	/**
	 * When publishing messages to a queue where delay seconds is set, the queue will act as a delayed queue
	 * @return delaySeconds
	 */
	@JsonProperty("delaySeconds")
	public int getDelaySeconds() {
		return delaySeconds;
	}
	@JsonProperty("delaySeconds")
	public void setDelaySeconds(int delaySeconds) {
		this.delaySeconds = delaySeconds;
	}
	public SqsQueueConfig withDelay(final int seconds){
		this.delaySeconds = seconds;
		return this;
	}
	
	
	/***
	 * Maximum Message size
	 * @return maximumMessageSize - The maximum message size in bytes
	 */
	@JsonProperty("maximumMessageSize")
	public int getMaximumMessageSize() {
		return maximumMessageSize;
	}
	@JsonProperty("maximumMessageSize")
	public void setMaximumMessageSize(int maximumMessageSize) {
		this.maximumMessageSize = maximumMessageSize;
	}
	public SqsQueueConfig withMaxMessageSize(final int maxMessageSize){
		this.maximumMessageSize = maxMessageSize;
		return this;
	}
	
	
	/**
	 * Retention Period
	 * @return messageRetentionPeriod
	 */
	@JsonProperty("messageRetentionPeriod")
	public int getMessageRetentionPeriod() {
		return messageRetentionPeriod;
	}
	@JsonProperty("messageRetentionPeriod")
	public void setMessageRetentionPeriod(int messageRetentionPeriod) {
		this.messageRetentionPeriod = messageRetentionPeriod;
	}
	public SqsQueueConfig withMessageRetentionPeriod(final int retention){
		this.messageRetentionPeriod = retention;
		return this;
	}
	
	
	/**
	 * Default Visibility timeout. Messages will remain hidden from all consumers for a period of time after being consumed. 
	 * If not deleted within the visibilityTimeout window the message will appear as available again, so long as it's still under the retention period
	 * @return - visibilityTimeout
	 */
	@JsonProperty("visibilityTimeout")
	public int getVisibilityTimeout() {
		return visibilityTimeout;
	}
	@JsonProperty("visibilityTimeout")
	public void setVisibilityTimeout(int visibilityTimeout) {
		this.visibilityTimeout = visibilityTimeout;
	}
	public SqsQueueConfig withVisibilityTimeout(final int timeout){
		this.visibilityTimeout = timeout;
		return this;
	}
	
	@JsonProperty("queueCreationTimeoutMS")
	public int getQueueCreationTimeoutMS() {
		return queueCreationTimeoutMS;
	}
	
	@JsonProperty("queueCreationTimeoutMS")
	public void setQueueCreationTimeoutMS(int queueCreationTimeoutMS) {
		this.queueCreationTimeoutMS = queueCreationTimeoutMS;
	}
	public SqsQueueConfig withQueueCreationTimeout(final int queueCreationTimeoutMS){
		this.queueCreationTimeoutMS = queueCreationTimeoutMS;
		return this;
	}
	
	
	
	protected Map<String, String> getAttributes() {
		return this.createAttributes();
	}
	
	/***
	 * Create SQS Attribute Map from configuration settings
	 */
	public Map<String, String> createAttributes() {
		final Map<String, String> attributes = new HashMap<String, String>();
		if(this.maximumMessageSize>0){
			attributes.put(MAXIMUM_MESSAGE_SIZE, String.valueOf(this.maximumMessageSize));	
		}
		if(this.messageRetentionPeriod>0){
			attributes.put(MESSAGE_RETENTION_PERIOD, String.valueOf(this.messageRetentionPeriod));	
		}
		if(this.visibilityTimeout>0){
			attributes.put(VISIBILITY_TIMEOUT, String.valueOf(this.visibilityTimeout));	
		}
		if(this.delaySeconds>0){
			attributes.put(DELAY_SECONDS, String.valueOf(this.delaySeconds));	
		}
		return attributes;
	}
} public class EventHandlers {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(EventHandlers.class);
	
	
	public static final String DEFAULT_CLIENT_ID = "default-example-client";
	
	

	public static final String TOPIC_CHECKOUT_BOOK_REQUEST = "CheckoutBookRequest";
	public static final int TOPIC_CHECKOUT_BOOK_REQUEST_SHARDS = 1;
	
	public static final String TOPIC_BOOK_CHANGE_EVENT = "BookChangeEvent";
	public static final int TOPIC_BOOK_CHANGE_EVENT_SHARDS = 1;
	
	/***
	 * Event Handler that receives Book Check out submissions
	 * @author shawn
	 *
	 */
	static class CheckOutRequestEventHandler implements EventHandler{
		private final BookRepository repository;
		public CheckOutRequestEventHandler(final BookRepository repository){
			this.repository = repository;
		}
		@Override
		public void handle(Event event) {
			
			final EventData eventData = event.getEventData();
			
			try {
				
				final String bookId = new String(eventData.getData().get("bookId").toString());
				final String userId = new String(eventData.getData().get("userId").toString());
				
				final Book book = this.repository.get(Key.create(bookId));
				if(Book.CHECKED_OUT_BY_NOBODY.equals(book.getCheckedOutBy())){
					book.setCheckedOutBy(userId);
					this.repository.saveIf(book,ComparisonOperator.EQ, "checkedOutBy", Book.CHECKED_OUT_BY_NOBODY);
					LOGGER.info("=========================== SAVING BOOK CHECKOUT REQUEST ===========================");
				}
			} catch (Exception e) {
				LOGGER.error(e.getMessage(), e);
			}
			
			
			
		}
		@Override
		public void handle(List<Event> events) {
			for(Event event : events){
				this.handle(event);
			}
		}
	}
	
	
	
	/****
	 * Event Handler that receives change events for books
	 * @author shawn
	 *
	 */
	static class BookChangeEventHandler implements EventHandler{
		public BookChangeEventHandler(){}
		@Override
		public void handle(Event event) {
			
			final EventData eventData = event.getEventData();
			
			try {

				final Object bookId = eventData.getData().get("bookId");
				final Object checkedOutBy = eventData.getData().get("checkedOutBy");
				
				if(!Book.CHECKED_OUT_BY_NOBODY.equals(checkedOutBy)){
					LOGGER.info("========================= Book has successfully been checked out ========================");
					LOGGER.info("**** received book change event: eventId: {}, eventCorrelationId:{},  bookId: {} checkedOutBy:{}", 
							event.getId(), event.getCorrelationId(), bookId, checkedOutBy);
				}
			} catch (Exception e) {
				LOGGER.error(e.getMessage(), e);
			}
		}
		
		@Override
		public void handle(List<Event> events) {
			for(Event event : events){
				this.handle(event);
			}
		}
	}
	
	
	
	
	/*** Events **/
	static class BookChangeEvent {
		private Book book;
		public BookChangeEvent(){}
		public BookChangeEvent(final Book book){
			this.book = book;
		}
		public Book getBook() {
			return book;
		}
		public void setBook(Book book) {
			this.book = book;
		}	
	}
	
	
	
	/*** Events **/
	 static class CheckoutBookRequest {
		public CheckoutBookRequest(){}
		public CheckoutBookRequest(final String bookId, final String userId){
			this.bookId = bookId;
			this.userId = userId;
		}
		private String userId;
		private String bookId;
		public String getUserId() {
			return userId;
		}
		public void setUserId(String userId) {
			this.userId = userId;
		}
		public String getBookId() {
			return bookId;
		}
		public void setBookId(String bookId) {
			this.bookId = bookId;
		}
	}
} public class ExampleProperties {
	
	
	
	/***
	 * Initialize Properties
	 * @param propertyFileName
	 * @return properties
	 * @throws IOException
	 */
	public Properties initalize(final String propertyFileName) throws IOException{
		final Properties props = this.getProperties(propertyFileName);
		return props;
	}
	
	
	/***
	 * Get example properties for file name
	 * @param propertiesFileName
	 * @return properties
	 * @throws IOException
	 */
	public Properties getProperties(final String propertiesFileName) throws IOException{
		final Properties properties = new Properties();
		try(InputStream stream = getClass().getClassLoader().getResourceAsStream(propertiesFileName)){
			if(stream!=null){
				properties.load(stream);
			}else {
				throw new FileNotFoundException(String.format("Property file %s was not found", propertiesFileName));
			}
			return properties;	
		}
	}
}
import io.microgenie.application.queue.Message;
import io.microgenie.application.queue.MessageHandler;
import io.microgenie.aws.config.AwsConfig;
import io.microgenie.aws.config.DynamoDbConfig;
import io.microgenie.aws.config.KinesisConfig;
import io.microgenie.aws.config.S3Config;
import io.microgenie.aws.config.SqsConfig;
import io.microgenie.aws.config.SqsConsumerConfig;
import io.microgenie.aws.config.SqsQueueConfig;
import io.microgenie.examples.application.EventHandlers;
import io.microgenie.examples.util.ExampleProperties;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
package io.microgenie.examples;






public class ExampleConfig {
	
	private static final Logger LOGGER = LoggerFactory.getLogger(ExampleConfig.class);
	
	public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
	
	public static final ExampleProperties appProperties = new ExampleProperties();
	
	/** Command Example Properties **/
	public static final String PROPERTY_FILE_NAME = "application.properties";
	public static final String EVENT_PROPERTY_FILE_NAME = "event.examples.properties";
	public static final String DATABASE_PROPERTY_FILE_NAME = "database.examples.properties";
	
	/** Property Keys **/
	public  static final String BUCKET_PROPERTY = "s3.buckets.default";
	public  static final String FILE_SAVED_QUEUE_PROPERTY = "sqs.queues.filesaved";
	
	
	
	/*** Event Example Properties **/
	public static final String TOPIC_1_NAME = "events.topics.topic1";
	public static final String TOPIC_1_SHARDS = "events.topics.topic1.shards";
	
	public static final String TOPIC_2_NAME = "events.topics.topic2";
	public static final String TOPIC_2_SHARDS = "events.topics.topic2.shards";
	
	
	
	
	/*** Database Examples **/
	public static final String DATABASE_SCAN_PACKAGE = "database.scanPackagePrefix";

	
	public static Properties getProperties(final String propertyFileName) throws IOException{
		return appProperties.getProperties(propertyFileName);
	}
	
	
	/***
	 * Create the aws Configuration
	 * @param properties 
	 * @return awsConfig
	 * @throws IOException 
	 */
	public static AwsConfig createConfig(final String propertyFileName) throws IOException {
		
		final Properties props = appProperties.getProperties(propertyFileName);
		final String defaultBucket = props.getProperty(ExampleConfig.BUCKET_PROPERTY);
		final String fileSavedQueueName = props.getProperty(ExampleConfig.FILE_SAVED_QUEUE_PROPERTY);
		final S3Config s3 = new S3Config();
		
		s3.setDefaultDrive(defaultBucket);
		
		SqsQueueConfig queue = new SqsQueueConfig();
		queue.setName(fileSavedQueueName);
		
		final SqsConsumerConfig consumerConfig = new SqsConsumerConfig();
		consumerConfig.setQueue(fileSavedQueueName);
		consumerConfig.setThreads(3);
		consumerConfig.setHandlerInstance(new MessageHandler(){