package org.atlasapi.system.bootstrap.workers;

import static com.google.common.base.Preconditions.checkNotNull;

import java.util.Set;

import org.atlasapi.media.channel.Channel;
import org.atlasapi.media.channel.ChannelResolver;
import org.atlasapi.media.entity.Publisher;
import org.atlasapi.messaging.v3.ScheduleUpdateMessage;
import org.atlasapi.system.bootstrap.ChannelIntervalScheduleBootstrapTask;
import org.atlasapi.system.bootstrap.SourceChannelIntervalFactory;
import org.joda.time.Interval;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ImmutableSet;
import com.metabroadcast.common.base.Maybe;
import com.metabroadcast.common.ids.SubstitutionTableNumberCodec;
import com.metabroadcast.common.queue.Worker;
import com.metabroadcast.common.scheduling.UpdateProgress;

public class ScheduleReadWriteWorker implements Worker<ScheduleUpdateMessage> {

    private final Logger log = LoggerFactory.getLogger(ScheduleReadWriteWorker.class);

    private final SubstitutionTableNumberCodec idCodec = SubstitutionTableNumberCodec.lowerCaseOnly();

    private final SourceChannelIntervalFactory<ChannelIntervalScheduleBootstrapTask> taskFactory;
    private final ChannelResolver channelResolver;
    private final Set<Publisher> ignoredSources;

    public ScheduleReadWriteWorker(SourceChannelIntervalFactory<ChannelIntervalScheduleBootstrapTask> taskFactory, 
            ChannelResolver channelResolver, Iterable<Publisher> ignoredSources) {
        this.channelResolver = checkNotNull(channelResolver);
        this.taskFactory = checkNotNull(taskFactory);
        this.ignoredSources = ImmutableSet.copyOf(ignoredSources);
    }
    
    @Override
    public void process(ScheduleUpdateMessage msg) {
        String updateMsg = String.format("update %s %s %s-%s", 
                msg.getSource(), msg.getChannel(), msg.getUpdateStart(), msg.getUpdateEnd());
        
        Maybe<Publisher> source = Publisher.fromKey(msg.getSource());
        if (!source.hasValue()) {
            log.warn(updateMsg + ": unknown source %s", msg.getSource());
            return;
        }
        Publisher src = source.requireValue();
        if (ignoredSources.contains(src)) {
            log.debug(updateMsg + ": ignoring source %s", src.key());
            return;
        }
        
        long cid = idCodec.decode(msg.getChannel()).longValue();
        Maybe<Channel> channel = channelResolver.fromId(cid);
        if (!channel.hasValue()) {
            log.warn(updateMsg + ": unknown channel %s (%s)", msg.getChannel(), cid);
        }
        
        Interval interval = new Interval(msg.getUpdateStart(), msg.getUpdateEnd());
        
        log.debug(updateMsg + ": processing");
        try {
            UpdateProgress result = taskFactory.create(src, channel.requireValue(), interval).call();
            log.debug(updateMsg + ": processed: " + result);
        } catch (Exception e) {
            log.error("failed " + updateMsg, e);
        }
    }

}

@@ -2038,37 +2038,27 @@
    log.warn(
-updateMsg + %22
+%22%7B%7D
 : unknown so
@@ -2062,19 +2062,30 @@
  source 
-%25s%22
+%7B%7D%22, updateMsg
 , msg.ge
@@ -2239,29 +2239,19 @@
 g.debug(
-updateMsg + %22
+%22%7B%7D
 : ignori
@@ -2260,19 +2260,30 @@
  source 
-%25s%22
+%7B%7D%22, updateMsg
 , src.ke
@@ -2510,29 +2510,19 @@
 og.warn(
-updateMsg + %22
+%22%7B%7D
 : unknow
@@ -2535,16 +2535,27 @@
 nel 
-%25s (%25s)%22
+%7B%7D (%7B%7D)%22, updateMsg
 , ms
@@ -2702,37 +2702,27 @@
   log.debug(
-updateMsg + %22
+%22%7B%7D
 : processing
@@ -2722,16 +2722,27 @@
 cessing%22
+, updateMsg
 );%0A     
@@ -2878,37 +2878,39 @@
 bug(
-updateMsg + %22: processed: %22 +
+%22%7B%7D: processed: %7B%7D%22, updateMsg,
  res

improve schedule bootstrap logging