package org.atlasapi.system.bootstrap;

import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.metabroadcast.common.properties.Configurer;

import org.atlasapi.AtlasPersistenceModule;
import org.atlasapi.ElasticSearchContentIndexModule;
import org.atlasapi.SchedulerModule;
import org.atlasapi.content.Content;
import org.atlasapi.media.entity.Publisher;
import org.atlasapi.system.bootstrap.workers.BootstrapWorkersModule;
import org.atlasapi.system.bootstrap.workers.DelegatingContentStore;
import org.atlasapi.system.legacy.LegacyPersistenceModule;
import org.atlasapi.topic.Topic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;

import com.google.common.base.Supplier;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.metabroadcast.common.scheduling.RepetitionRules;
import com.metabroadcast.common.scheduling.UpdateProgress;
import com.metabroadcast.common.time.DayRangeGenerator;
import com.metabroadcast.common.time.SystemClock;

@Configuration
@Import({AtlasPersistenceModule.class, BootstrapWorkersModule.class, LegacyPersistenceModule.class,
    SchedulerModule.class})
public class BootstrapModule {

    //we only need 2 here, on to run the bootstrap and one to be able to return quickly when it's running
    private static final Integer NUMBER_OF_SCHECHULE_CONTROLLER_THREADS = 2;
    private static final Integer NUMBER_OF_SCHEDULE_BOOTSTRAP_THREADS = Configurer.get("boootstrap.schedule.numThreads").toInt();

    @Autowired private AtlasPersistenceModule persistence;
    @Autowired private LegacyPersistenceModule legacy;
    @Autowired private BootstrapWorkersModule workers;
    @Autowired private SchedulerModule scheduler;
    @Autowired private ElasticSearchContentIndexModule search;

    @Bean
    BootstrapController bootstrapController() {
        BootstrapController bootstrapController = new BootstrapController();
        
        bootstrapController.addBootstrapPair("legacy-content", new ResourceBootstrapper<Content>(legacy.legacyContentLister()),
                concurrencyLevel -> new ContentWritingBootstrapListener(concurrencyLevel, persistence.contentStore())
        );
        bootstrapController.addBootstrapPair("legacy-topics", new ResourceBootstrapper<Topic>(legacy.legacyTopicLister()),
                concurrencyLevel -> new TopicWritingBootstrapListener(concurrencyLevel, persistence.topicStore())
        );
        return bootstrapController;
    }

    @Bean
    IndividualContentBootstrapController contentBootstrapController() {
        return new IndividualContentBootstrapController(legacy.legacyContentResolver(), 
                legacy.legacyContentLister(),
                persistence.contentStore(), search.contentIndex());
    }

    @Bean
    IndividualTopicBootstrapController topicBootstrapController() {
        return new IndividualTopicBootstrapController(legacy.legacyTopicResolver(), persistence.topicStore());
    }
    
    @Bean
    ExecutorServiceScheduledTask<UpdateProgress> scheduleBootstrapTask() {
        ChannelIntervalScheduleBootstrapTaskFactory taskFactory = workers.scheduleBootstrapTaskFactory();
        DayRangeGenerator dayRangeGenerator = new DayRangeGenerator().withLookAhead(7).withLookBack(7);
        Set<Publisher> sources = ImmutableSet.of(Publisher.PA);
        Supplier<Iterable<ChannelIntervalScheduleBootstrapTask>> supplier =
                new SourceChannelIntervalTaskSupplier<>(taskFactory, persistence.channelResolver(), dayRangeGenerator, sources, new SystemClock());
        ExecutorService executor = Executors.newFixedThreadPool(10, 
            new ThreadFactoryBuilder().setDaemon(true).setNameFormat("schedule-bootstrap-%d").build());
        return new ExecutorServiceScheduledTask<>(executor, supplier, 10, 1, TimeUnit.MINUTES);
    }
    
    @Bean
    ChannelIntervalScheduleBootstrapTaskFactory scheduleBootstrapTaskFactory() {
        return new ChannelIntervalScheduleBootstrapTaskFactory(legacy.legacyScheduleStore(), persistence.scheduleStore(), 
            new DelegatingContentStore(legacy.legacyContentResolver(), persistence.contentStore()));
    }
    
    @PostConstruct
    public void schedule() {
        scheduler.scheduler().schedule(scheduleBootstrapTask(), RepetitionRules.NEVER);
    }
    
    @Bean
    public ScheduleBootstrapController scheduleBootstrapController() {
        return new ScheduleBootstrapController(
                workers.scheduleBootstrapTaskFactory(),
                persistence.channelResolver(),
                executorService(NUMBER_OF_SCHECHULE_CONTROLLER_THREADS, "ScheduleBootstrapController"),
                scheduleBootstrapper()
        );
    }

    @Bean
    public ScheduleBootstrapper scheduleBootstrapper() {
        return new ScheduleBootstrapper(
                executorService(NUMBER_OF_SCHEDULE_BOOTSTRAP_THREADS, "ScheduleBootstrapper"),
                workers.scheduleBootstrapTaskFactory()
        );
    }


    private ListeningExecutorService executorService(Integer concurrencyLevel, String namePrefix) {
        return MoreExecutors.listeningDecorator(
                new ThreadPoolExecutor(
                        concurrencyLevel,
                        concurrencyLevel,
                        0, TimeUnit.MICROSECONDS,
                        new ArrayBlockingQueue<>(100 * Runtime.getRuntime().availableProcessors()),
                        new ThreadFactoryBuilder().setNameFormat(namePrefix + " Thread %d").build(),
                        new ThreadPoolExecutor.CallerRunsPolicy()
                )
        );
    }
}

@@ -2601,39 +2601,205 @@
         
-    concurrencyLevel -%3E
+new BootstrapListenerFactory%3CContent%3E() %7B%0A                @Override%0A                public BootstrapListener%3CContent%3E buildWithConcurrency(int concurrencyLevel) %7B%0A                    return
  new Con
@@ -2864,32 +2864,65 @@
 .contentStore())
+;%0A                %7D%0A            %7D
 %0A        );%0A    
@@ -3056,31 +3056,193 @@
     
-    concurrencyLevel -%3E
+new BootstrapListenerFactory%3CTopic%3E() %7B%0A                @Override%0A                public BootstrapListener%3CTopic%3E buildWithConcurrency(int concurrencyLevel) %7B%0A                    return
  new
@@ -3315,16 +3315,49 @@
 Store())
+;%0A                %7D%0A            %7D
 %0A       
@@ -4338,28 +4338,24 @@
             
-    
 new SourceCh
@@ -4380,16 +4380,52 @@
 upplier%3C
+ChannelIntervalScheduleBootstrapTask
 %3E(taskFa
@@ -4731,16 +4731,30 @@
 ledTask%3C
+UpdateProgress
 %3E(execut
@@ -6317,16 +6317,24 @@
 ngQueue%3C
+Runnable
 %3E(100 * 

Revert lambdafication because Java