package info.glennengstrand;

import com.google.inject.Binder;
import com.google.inject.Module;  
import com.google.inject.Provides;

import io.dropwizard.jdbi.DBIFactory;
import io.dropwizard.setup.Environment;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Cluster;

import info.glennengstrand.db.ElasticSearchDAO;
import info.glennengstrand.db.RedisCache;
import info.glennengstrand.db.SearchDAO.DoNothingSearchDAO;
import info.glennengstrand.db.SearchDAO;
import info.glennengstrand.core.ParticipantApiServiceImpl;
import info.glennengstrand.core.MessageLogger;
import info.glennengstrand.api.Friend;
import info.glennengstrand.api.Participant;
import info.glennengstrand.core.FriendApiServiceImpl;
import info.glennengstrand.core.InboundApiServiceImpl;
import info.glennengstrand.core.KafkaPerformanceLogger;
import info.glennengstrand.core.OutboundApiServiceImpl;
import info.glennengstrand.resources.ParticipantApi.ParticipantApiService;
import info.glennengstrand.resources.InboundApi.InboundApiService;
import info.glennengstrand.resources.FriendApi.FriendApiService;
import info.glennengstrand.resources.OutboundApi.OutboundApiService;

import java.util.List;
import java.util.function.Supplier;

import org.apache.kafka.clients.producer.Producer;
import org.skife.jdbi.v2.DBI;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class NewsFeedModule implements Module {

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

	private final DBIFactory factory = new DBIFactory();
	private DBI dbi = null;
	private Session session = null;
	private SearchDAO esdao = null;
	private MessageLogger<Long> logger = null;
	private FriendCache friendCache = null;
	private ParticipantCache participantCache = null;

	@Override
	public void configure(Binder binder) {
		binder.bind(ParticipantApiService.class).to(ParticipantApiServiceImpl.class);
		binder.bind(FriendApiService.class).to(FriendApiServiceImpl.class);
		binder.bind(InboundApiService.class).to(InboundApiServiceImpl.class);
		binder.bind(OutboundApiService.class).to(OutboundApiServiceImpl.class);
	}
	
    @Provides
	public DBI getDbi(NewsFeedConfiguration config, Environment environment) {
    	if (dbi == null) {
    		synchronized(LOGGER) {
    			if (dbi == null) {
    				dbi = factory.build(environment, config.getDataSourceFactory(), "mysql");
    			}
    		}
    	}
		return dbi;
	}
    
    @Provides
    public Session getNoSqlSession(NewsFeedConfiguration config) {
    	if (session == null) {
    		synchronized(LOGGER) {
    			if (session == null) {
    		    	Cluster cluster = Cluster.builder().addContactPoint(config.getNosqlHost()).build();
    		    	session = cluster.connect(config.getNosqlKeyspace());
    			}
    		}
    	}
    	return session;
    }
    
    @Provides
    public SearchDAO getElasticSearch(NewsFeedConfiguration config) {
    	if (esdao == null) {
    		synchronized(LOGGER) {
    			if (esdao == null) {
    		    	try {
    		    		esdao = new ElasticSearchDAO(config.getSearchHost(), config.getSearchPort(), config.getSearchIndex(), config.getSearchMapping());
    				} catch (Exception e) {
    					LOGGER.error("Cannot connect to elastic search: ", e);
    					esdao = new DoNothingSearchDAO();
    				}
    			}
    		}
    	}
    	return esdao;
    }
    
    @Provides
    public ParticipantCache getParticipantCache(NewsFeedConfiguration config) {
    	if (participantCache == null) {
    		synchronized(LOGGER) {
    			if (participantCache == null) {
    				participantCache = new ParticipantCache(Participant.class, config);
    			}
    		}
    	}
    	return participantCache;
    }
    
    @Provides
    public FriendCache getFriendCache(NewsFeedConfiguration config) {
    	if (friendCache == null) {
    		synchronized(LOGGER) {
    			if (friendCache == null) {
    				friendCache = new FriendCache(Friend.class, config);
    			}
    		}
    	}
    	return friendCache;
    }
    
    @Provides
    public MessageLogger<Long> getPerformanceLogger(NewsFeedConfiguration config) {
    	if (logger == null) {
    		synchronized(LOGGER) {
    			if (logger == null) {
    		    	try {
    		    		logger = new KafkaPerformanceLogger(config.getMessageBroker(), config.getMessageTopic());
    		    	} catch (Exception e) {
    		    		LOGGER.error("Cannot connect to Kafka: ", e);
    		    		logger = new MessageLogger.DoNothingMessageLogger();
    		    	}
    			}
    		}
    	}
    	return logger;
    }
    
    public NewsFeedModule() {
    	
    }
    
    public static class ParticipantCache extends RedisCache<Participant> {

		public ParticipantCache(Class<Participant> serializationType, NewsFeedConfiguration config) {
			super(serializationType, config);
		}
    	
    }
    
    public static class FriendCache extends RedisCache<Friend> {

		public FriendCache(Class<Friend> serializationType, NewsFeedConfiguration config) {
			super(serializationType, config);
		}
    	
    }

}

@@ -1580,16 +1580,78 @@
 .class);
+%0A%09private static final String KAFKA_ENABLED = %22KAFKA_ENABLED%22;
 %0A%0A%09priva
@@ -4303,32 +4303,177 @@
 gger == null) %7B%0A
+    %09%09%09%09String ke = System.getenv(KAFKA_ENABLED);%0A    %09%09%09%09boolean useKafka = ke == null %7C%7C %22true%22.equalsIgnoreCase(ke);%0A    %09%09%09%09if (useKafka) %7B%0A%09
     %09%09    %09try %7B
@@ -4465,32 +4465,33 @@
    %09%09    %09try %7B%0A
+%09
     %09%09    %09%09logg
@@ -4572,24 +4572,25 @@
 geTopic());%0A
+%09
     %09%09    %09%7D
@@ -4608,24 +4608,25 @@
 eption e) %7B%0A
+%09
     %09%09    %09%09
@@ -4667,24 +4667,25 @@
 fka: %22, e);%0A
+%09
     %09%09    %09%09
@@ -4733,24 +4733,25 @@
 geLogger();%0A
+%09
     %09%09    %09%7D
@@ -4747,24 +4747,113 @@
    %09%09    %09%7D%0A
+    %09%09%09%09%7D else %7B%0A    %09%09%09%09%09logger = new MessageLogger.DoNothingMessageLogger();%0A    %09%09%09%09%7D%0A
     %09%09%09%7D%0A   

enable or disable kafka integration with an environment variable switch