public abstract class ApplicationFactory implements Closeable{

	public ApplicationFactory(){}

	public abstract  EventFactory events();
	public abstract  QueueFactory queues();
	public abstract  FileStoreFactory blobs();
	public abstract <T extends DatabaseFactory>  T database();
} public abstract class FileStoreFactory implements Closeable {

	public FileStoreFactory(){}
	public abstract FileContent read(final FilePath path) throws IOException;
	public abstract FileContentStream readStream(final FilePath path);
	public abstract FilePath delete(final FilePath path);
	public abstract FilePath save(final FileContent file); 
	public abstract String getDefaultDrive();
} public abstract class DatabaseFactory implements Closeable {

	
	private final Map<Class<?>, EntityRepository<?>> repositories = Maps.newHashMap();
	
	public DatabaseFactory(){}

	
	@SuppressWarnings("unchecked")
	public <T,R extends EntityRepository<T>> R repos(Class<T> clazz) {
		final EntityRepository<?> repo =  repositories.get(clazz);
		if(repo!=null){
			return (R) repo;	
		}
		return null;
	}
	
	
	
	
	/***
	 *  Register a repository class
	 * @param clazz
	 * @param repo
	 */
	public <T> EntityRepository<T> registerRepo(Class<T> clazz, EntityRepository<T> repo){
		this.repositories.put(clazz, repo);
		return repo;
	}
} public abstract class EventFactory implements Closeable {


	/***
	 * Constructor
	 */
	public EventFactory(){}
	

	public abstract void publish(final Event event);
	public abstract void publish(final List<Event> events);
	
	public abstract void publish(final String clientId, final Event event);
	public abstract void publish(final String clientId, final List<Event> events);

	public abstract Publisher createPublisher(final String clientId);
	public abstract Subscriber createSubscriber(final String clientId, final String topic);
	
	public abstract StateChangePublisher createChangePublisher(final String clientId, final StateChangeConfiguration stateChangeConfig);
	
	
	/***
	 * Subscribe to the given topic, providing an event handler that will process all
	 * consumed events for this Subscriber
	 * 
	 * @param topic
	 * @param clientId
	 * @param handler
	 */
	public abstract void subcribe(final String clientId, final String topic, final EventHandler handler);
	

	
	/**
	 * Creates a publisher
	 * @param clientId - The clientId of the publisher to get
	 * @return publisher
	 */
	public synchronized Publisher getPublisher(final String clientId) {
		return this.createPublisher(clientId);
	}
} public abstract class QueueFactory implements Closeable {

	/***
	 * Creates the default ThreadGroupFactory
	 */
	public QueueFactory(){}
	
	
	public abstract Producer getProducer();
	public abstract Consumer getConsumer(final String queue);
	
	public abstract void produce(final Message message);
	public abstract void produceBatch(final List<Message> messages);
	
	public abstract void consume(String queue, int threadCount, MessageHandler handler);

} public class CloseableUtil {

	private static final Logger LOGGER = LoggerFactory.getLogger(CloseableUtil.class); 
	
	/** The Guava version throws an IOException even if swallow exception is set to true, based on guava version **/
	public static void closeQuietly(Closeable closeable){
		try{
			Closeables.close(closeable, true);
		}catch(Exception ex){
			LOGGER.debug(ex.getMessage(),ex);
		}
	}
} 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 DynamoDbMapperFactory extends DatabaseFactory {

	
	private final DynamoMapperRepository mapperRepo;
	
	
	/***
	 * Create a {@link DatabaseFactory} that implements behavior of the {@link DynamoDBMapper} but also
	 * includes initialization of any scanned models that are annotated with {@link DynamoDBTable}.
	 * 
	 * Any models that are found from the scan will be further scanned for fields and indexes to create

	 * @param client - The AmazonDynamoDBClient {@link AmazonDynamoDBClient}
	 */
	public DynamoDbMapperFactory(final AmazonDynamoDBClient client) {
		this.mapperRepo = DynamoMapperRepository.create(client);
	}

	
	
	public DynamoMapperRepository getMapperRepository(){
		return this.mapperRepo;
	}


	@Override
	public void close() throws IOException {

	}
} public class KinesisEventFactory extends EventFactory {

	private static final Logger LOGGER = LoggerFactory.getLogger(KinesisEventFactory.class);

	private static final String WORKER_ID_TEMPLATE = "kinesis-%s-%s";
	private static final String DEFAULT_CLIENT_ID = "default-client";
	
	private final ObjectMapper mapper;
	
	private final AmazonKinesisClient kinesisClient;	
	private final AmazonDynamoDBClient dynamoDbClient;
	private final AmazonCloudWatchClient cloudwatchClient;

	
	private final Map<String, Publisher> publishers = Maps.newHashMap();
	private final Map<String, Subscriber> subscribers = Maps.newHashMap();

	
	public KinesisEventFactory(final AmazonKinesisClient kinesisClient, final AmazonDynamoDBClient dynamoDbClient, final AmazonCloudWatchClient cloudwatchClient, final ObjectMapper mapper){
		this.mapper = mapper;
		this.kinesisClient = kinesisClient;
		this.dynamoDbClient = dynamoDbClient;
		this.cloudwatchClient = cloudwatchClient;
	}
	
	@Override
	public void publish(final Event event) {
		this.publish(DEFAULT_CLIENT_ID, event);
	}
	@Override
	public void publish(final List<Event> events) {
		this.publish(DEFAULT_CLIENT_ID, events);
	}
	@Override
	public void publish(final String clientId, final Event event) {
		final Publisher publisher = this.createPublisher(clientId);
		publisher.submit(event);
	}
	@Override
	public void publish(final String clientId, final List<Event> events) {
		Publisher publisher = this.createPublisher(clientId);
		publisher.submitBatch(events);
	}
	
	
	
	
	/****
	 * Create subscriber if it has not already been created
	 */
	@Override
	public synchronized  Subscriber createSubscriber(final String clientId, final String topic) {
		
		final String clientIdToUse = Strings.isNullOrEmpty(clientId)? DEFAULT_CLIENT_ID : clientId;
		Subscriber subscriber = this.subscribers.get(topic);
		if(subscriber == null){
			LOGGER.debug("creating kinsis subscriber for topic {} - clientId: {}", topic, clientIdToUse);
			final KinesisClientLibConfiguration config = createConsumerConfig(clientIdToUse, topic);
			subscriber = new KinesisConsumer(topic, config, this.kinesisClient, this.dynamoDbClient, this.cloudwatchClient, this.mapper);
			this.subscribers.put(topic, subscriber);
		}
		return subscriber;
	}

	

	
	/***
	 * Create publisher, if it has not already been created
	 */
	@Override
	public synchronized Publisher createPublisher(final String clientId) {
		
		final String clientIdToUse = Strings.isNullOrEmpty(clientId) ? DEFAULT_CLIENT_ID : clientId;
		
		Publisher publisher = publishers.get(clientIdToUse);
		if(publisher==null){
			LOGGER.debug("creating kinsis publisher");
			publisher = new KinesisProducer(clientIdToUse, this.mapper, this.kinesisClient);
			publishers.put(clientIdToUse, publisher);
		}
		return publisher;
	}




	@Override
	public StateChangePublisher createChangePublisher(final String clientId, StateChangeConfiguration stateChangeConfig) {
		final Publisher publisher = this.createPublisher(clientId);
		final StateChangePublisher changePublisher = new StateChangePublisher(stateChangeConfig, publisher);
		return changePublisher;
	}
	
	

	@Override
	public void subcribe(final String clientId, final String topic, final EventHandler handler) {
		final Subscriber subscriber = this.createSubscriber(clientId, topic);
		subscriber.subscribe(handler);
	}
	
	
	
	
	/***
	 * Create consumer configurations
	 */
	private KinesisClientLibConfiguration createConsumerConfig(final String clientId, final String topic) {
		
		final String kinesisApplication = String.format(WORKER_ID_TEMPLATE, topic, clientId);
		final KinesisClientLibConfiguration clientConfig = 
				new KinesisClientLibConfiguration(kinesisApplication, topic, new DefaultAWSCredentialsProviderChain(), UUID.randomUUID().toString());
		return clientConfig;		
	}



	
	/***
	 * Close all publishers and subscribers
	 */
	@Override
	public void close() throws IOException{
		
		for(java.util.Map.Entry<String, Subscriber> subscriptionEntry : this.subscribers.entrySet()){
			subscriptionEntry.getValue().stop();
		}
		this.subscribers.clear();
		for(java.util.Map.Entry<String, Publisher> publisherEntry : this.publishers.entrySet()){
			publisherEntry.getValue().close();
		}
		this.publishers.clear();
	}
} public class S3BlobFactory extends FileStoreFactory{

	
	private final String defaultBucket;
	private FileStore fileStore;
	private AmazonS3Client s3; 

	
	/**
	 * Constructs an S3 File Store implementation of CommandFactory
	 * @param s3Client
	 * @param defaultBucket - default bucket
	 */
	public S3BlobFactory(final AmazonS3Client s3Client, final String defaultBucket){
		this.s3 = s3Client;
		this.defaultBucket = defaultBucket;
		this.fileStore = new S3FileStore(defaultBucket, s3);
	}
	

	@Override
	public FileContent read(final FilePath path) throws IOException {
		return this.fileStore.read(path.getDrive(),path.getPath());
	}
	@Override
	public FileContentStream readStream(final FilePath path) {
		return this.fileStore.readStream(path.getDrive(), path.getPath());
	}
	@Override
	public FilePath delete(final FilePath path) {
		return this.fileStore.delete(path.getDrive(), path.getPath());
	}
	@Override
	public FilePath save(final FileContent file) {
		return this.fileStore.save(file);
	}
	

	/***
	 * Get the default bucket
	 */
	@Override
	public String getDefaultDrive() {
		return this.defaultBucket;
	}
	
	
	
	/***
	 * Close Thread Group Factory
	 * @throws IOException 
	 */
	@Override
	public void close() throws IOException{
		CloseableUtil.closeQuietly(this.fileStore);
	}
} public class SqsFactory extends QueueFactory{

	private static final Logger LOGGER = LoggerFactory.getLogger(SqsFactory.class);
	
	private final AmazonSQSClient sqs;
	private final SqsQueueAdmin admin;
	private final SqsConfig config;
	
	
	/** internal maps to store configuration for queues, consumer config and consumers **/
	private final Map<String, SqsQueueConfig> queueConfigMap = Maps.newHashMap();
	private final Map<String, SqsConsumerConfig> consumerConfigMap = Maps.newHashMap();
	private final Map<String, Consumer> consumers = Maps.newHashMap();
	
	private Producer producer;
	
	
	/**
	 * @param sqsClient
	 * @param config
	 */
	public SqsFactory(final AmazonSQSClient sqsClient, final SqsConfig config){
		
		this.sqs = sqsClient;
		this.config = config;
		this.admin = new SqsQueueAdmin(this.sqs);
		this.mapQueueConfig(config.getQueues());
		this.createConsumers(config.getConsumers());
		
		if(config.isProduces()){
			this.producer = new SqsProducer(this.sqs, admin);
		}
	}



	@Override
	public Producer getProducer() {
		return this.producer;
	}
	@Override
	public Consumer getConsumer(String queue) {
		return this.consumers.get(queue);
	}
	@Override
	public void produce(Message message) {
		this.producer.submit(message);
	}
	@Override
	public void produceBatch(List<Message> messages) {
		this.producer.submitBatch(messages);
	}
	
	
	
	/***
	 * Creates a consumer with the default SqsConfig. 
	 * <p>
	 * This method ensures that the queue exists. If it does not exist it will be created
	 * After the queue is created Consumers will be started. The number of threads determines how
	 * many consuming threads are started with the given queue
	 */
	@Override
	public synchronized void consume(final String queue, final int threads, final MessageHandler handler) {
		
		Consumer consumer = this.consumers.get(queue);
		if(consumer==null){
			this.createQueueAndConfigIfNotExists(queue, handler);
			final SqsConsumerConfig consumerConfig = new SqsConsumerConfig();
			consumerConfig.setQueue(queue);
			consumerConfig.setHandlerInstance(handler);
			consumerConfig.setThreads(threads);
			consumer = this.createAndSetConsumer(consumerConfig);
		}
		consumer.start();
	}


	


	/***
	 * Stop all consumers and close the SQS Client
	 */
	@Override
	public synchronized void close(){
		LOGGER.info("shutting down admin client");
		
		if(this.admin!=null){
			this.admin.shutdown();	
		}
		
		if(this.consumers !=null && this.consumers.size()>0){
			int i = 0;
			for(Entry<String, Consumer> c : consumers.entrySet()){
				LOGGER.info("stopping consumer: {}", ++i);
				c.getValue().stop();
				LOGGER.info("consumer[{}].isRunning() is: {}", i, c.getValue().isRunning());
			}
			this.consumers.clear();
		}
		if(this.consumerConfigMap !=null && !this.consumerConfigMap.isEmpty()){
			this.consumerConfigMap.clear();
		}
	}
	
	
	
	
	
	/***
	 * Initialize Queue Consumers, this also starts the consumers
	 * @param consumers
	 * @throws ClassNotFoundException 
	 */
	private void createConsumers(final List<SqsConsumerConfig> consumers)  {
		if(CollectionUtil.hasElements(consumers)){
			for(SqsConsumerConfig config: consumers){
				this.createAndSetConsumer(config);
			}
		}
	}

	

	/** 
	 * Create the {@link SqsConsumer} and map the consumerConfiguration
	 * @param consumerConfig
	 */
	private Consumer createAndSetConsumer(final SqsConsumerConfig consumerConfig) {
		final Consumer consumer = new SqsConsumer(this.sqs, this.admin, consumerConfig);
		this.consumers.put(consumerConfig.getQueue(), consumer);
		this.consumerConfigMap.put(consumerConfig.getQueue(), consumerConfig);
		return consumer;
	}
	



	
	
	/***
	 * Create a mapping of queue to queue configuration for lookups
	 *  
	 * @param queues
	 */
	private void mapQueueConfig(final List<SqsQueueConfig> queues) {
		if(CollectionUtil.hasElements(queues)){
			for(SqsQueueConfig config : queues){
				this.queueConfigMap.put(config.getName(), config);
			}			
		}
	}

	

	private void createQueueAndConfigIfNotExists(final String queue, final MessageHandler handler) {
		
		/** determine if the queue exists, if not, create it with the default settings **/
		try{
			final String url = this.admin.getQueueUrl(queue);
			if(Strings.isNullOrEmpty(url)){
				throw new QueueDoesNotExistException(String.format("The queue: %s was not found", queue));
			}
		}catch(QueueDoesNotExistException qneException){
			/** determine if the queue configuration exists **/
			SqsQueueConfig queueConfig = this.queueConfigMap.get(queue);
			if(queueConfig==null){
				/** create default config if we don't know about it **/
				queueConfig = new SqsQueueConfig();
				queueConfig.setName(queue);
				this.queueConfigMap.put(queue, queueConfig);
			}
			LOGGER.info("Queue: {} does not exist - creating the queue now", queue);
			this.admin.initializeQueue(queueConfig, this.config.isBlockUntilReady());
		}
	}
}
import io.microgenie.application.ApplicationFactory;
import io.microgenie.application.blob.FileStoreFactory;
import io.microgenie.application.database.DatabaseFactory;
import io.microgenie.application.events.EventFactory;
import io.microgenie.application.queue.QueueFactory;
import io.microgenie.application.util.CloseableUtil;
import io.microgenie.aws.config.AwsConfig;
import io.microgenie.aws.dynamodb.DynamoDbMapperFactory;
import io.microgenie.aws.kinesis.KinesisEventFactory;
import io.microgenie.aws.s3.S3BlobFactory;
import io.microgenie.aws.sqs.SqsFactory;
import com.amazonaws.services.cloudwatch.AmazonCloudWatchClient;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.amazonaws.services.kinesis.AmazonKinesisClient;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.sqs.AmazonSQSClient;
import com.fasterxml.jackson.databind.ObjectMapper;
package io.microgenie.aws;



/***
 * AWS Application factory
 * 
 * @author shawn
 */
public class AwsApplicationFactory extends ApplicationFactory {


	private final ObjectMapper mapper;
	private final AwsConfig config;

	private S3BlobFactory files;
	private KinesisEventFactory events;
	private SqsFactory queues;
	private DynamoDbMapperFactory databases;

	/** Amazon Clients **/
	private AmazonDynamoDBClient dynamoClient;
	private AmazonCloudWatchClient cloudwatchClient;
	private AmazonKinesisClient kinesisClient;
	private AmazonSQSClient sqsClient;
	private AmazonS3Client s3Client;

	/**
	 * Default Constructor to close thread group factory
	 * 
	 * @param config
	 *            - AWS configuration
	 */
	public AwsApplicationFactory(final AwsConfig config,final ObjectMapper mapper) {
		this.config = config;
		this.mapper = mapper;
		this.createConfiguredFactories(config);
	}

	/***
	 * Implementations that subclass {@link AwsApplicationFactory} should
	 * override this method to create implementation specific factories for
	 * events, http, queues, etc.. where an AWS service is not being used.
	 * 
	 * @param config
	 */
	protected void createConfiguredFactories(final AwsConfig config) {

		
		/*** Create any clients that have been configured **/
		if (config != null) {
			
			if (config.getKinesis() != null || config.getDynamo() != null) {

				/** both kinesis and dynamodb rely on the AmazonDynamoDbClient **/
				this.dynamoClient = new AmazonDynamoDBClient();
				
				/** Kinesis KCL uses the cloudwatchClient **/
				if (this.config.getKinesis() != null) {
					this.kinesisClient = new AmazonKinesisClient();
					this.cloudwatchClient = new AmazonCloudWatchClient();
					events = new KinesisEventFactory(kinesisClient, this.dynamoClient, this.cloudwatchClient, this.mapper);
				}
			}
			
			if (config.getS3() != null) {
				this.s3Client = new AmazonS3Client();
				files = new S3BlobFactory(this.s3Client, config.getS3().getDefaultDrive());
			}
			if (config.getDynamo() != null) {
				databases = new DynamoDbMapperFactory(this.dynamoClient);
			}
			if (config.getSqs() != null) {
				this.sqsClient = new AmazonSQSClient();
				queues = new SqsFactory(this.sqsClient, config.getSqs());
			}
		}
	}

	@Override
	public EventFactory events() {
		return events;
	}

	@Override
	public QueueFactory queues() {
		return queues;
	}

	@Override
	public FileStoreFactory blobs() {
		return files;
	}

	
	@SuppressWarnings("unchecked")
	@Override
	public <T extends DatabaseFactory> T database() {