package com.s24.redjob.channel.worker;

import java.lang.management.ManagementFactory;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.s24.redjob.channel.ChannelDao;
import com.s24.redjob.queue.worker.JobRunnerFactory;

/**
 * Default implementation of {@link AdminWorker}.
 */
public class AdminWorkerImpl {
   /**
    * Log.
    */
   private static final Logger log = LoggerFactory.getLogger(AdminWorkerImpl.class);

   /**
    * Channels to listen to.
    */
   private List<Topic> channels;

   /**
    * Channel dao.
    */
   private ChannelDao channelDao;

   /**
    * Message listener container.
    */
   private RedisMessageListenerContainer listenerContainer;

   /**
    * Sequence for worker ids.
    */
   private static final AtomicInteger IDS = new AtomicInteger();

   /**
    * Worker id.
    */
   private int id;

   /**
    * Name of this worker.
    */
   private String name;

   /**
    * Factory for creating job runners.
    */
   private JobRunnerFactory jobRunnerFactory;

   @PostConstruct
   public void afterPropertiesSet() {
      Assert.notNull(channels, "Precondition violated: channels != null.");
      Assert.notNull(channelDao, "Precondition violated: channelDao != null.");
      Assert.notNull(listenerContainer, "Precondition violated: listenerContainer != null.");

      id = IDS.incrementAndGet();
      name = createName();

      listenerContainer.addMessageListener((message, pattern) -> {
         // Deserialize job.
            try {
               execute(null);
            } catch (Throwable throwable) {
               throwable.printStackTrace();
            }
         }, channels);
   }

   /**
    * Create name for this worker.
    */
   protected String createName() {
      return ManagementFactory.getRuntimeMXBean().getName() + ":" + id + ":" +
            StringUtils.collectionToCommaDelimitedString(channels);
   }

   /**
    * Execute job.
    *
    * @param job
    *           Job.
    * @throws Throwable
    *            In case of errors.
    */
   protected void execute(Object job) throws Throwable {
      if (job == null) {
         log.error("AdminWorker {}: Missing job.", name);
         throw new IllegalArgumentException("Missing job.");
      }

      Runnable runner = jobRunnerFactory.runnerFor(job);
      if (runner == null) {
         log.error("AdminWorker {}: Job {}: No runner found.", name, job.getClass());
         throw new IllegalArgumentException("No runner found.");
      }

      log.info("AdminWorker {}: Job {}: Start.", name, job.getClass());
      try {
         runner.run();
         log.info("AdminWorker {}: Job {}: Success.", name, job.getClass());
      } catch (Throwable t) {
         log.error("AdminWorker {}: Job {}: Failed to execute.", name, job.getClass(), t);
         throw new IllegalArgumentException("Failed to execute.", t);
      } finally {
         log.info("AdminWorker {}: Job {}: End.", name, job.getClass());
      }
   }

   //
   // Injections.
   //

   /**
    * Channels to listen to.
    */
   public void setChannels(List<String> channels) {
      this.channels = channels.stream().map(ChannelTopic::new).collect(Collectors.toList());
   }

   /**
    * Channel dao.
    */
   public ChannelDao getChannelDao() {
      return channelDao;
   }

   /**
    * Channel dao.
    */
   public void setChannelDao(ChannelDao channelDao) {
      this.channelDao = channelDao;
   }
}

@@ -283,24 +283,46 @@
 gerFactory;%0A
+import org.slf4j.MDC;%0A
 import org.s
@@ -630,16 +630,55 @@
 nelDao;%0A
+import com.s24.redjob.queue.Execution;%0A
 import c
@@ -1955,49 +1955,275 @@
     
-// Deserialize job.%0A            try %7B%0A   
+MDC.put(%22worker%22, getName());%0A         // Deserialize job.%0A         Execution execution = null;%0A         Object job = null;%0A         try %7B%0A            MDC.put(%22execution%22, Long.toString(execution.getId()));%0A            MDC.put(%22job%22, job.getClass().getSimpleName());%0A
     
@@ -2245,19 +2245,16 @@
 (null);%0A
-   
         
@@ -2294,27 +2294,24 @@
             
-  
- 
 throwable.pr
@@ -2336,24 +2336,141 @@
         
+%7D finally %7B%0A            MDC.remove(%22job%22);%0A            MDC.remove(%22execution%22);%0A            MDC.remove(%22worker%22);%0A      
    %7D%0A
-   
       %7D,
@@ -2716,32 +2716,216 @@
 hannels);%0A   %7D%0A%0A
+   /**%0A    * Local unique id of worker.%0A    */%0A   public int getId() %7B%0A      return id;%0A   %7D%0A%0A   /**%0A    * Name of worker.%0A    */%0A   public String getName() %7B%0A      return name;%0A   %7D%0A%0A
    /**%0A    * Exe
@@ -3147,32 +3147,16 @@
 .error(%22
-AdminWorker %7B%7D: 
 Missing 
@@ -3160,22 +3160,16 @@
 ng job.%22
-, name
 );%0A     
@@ -3342,34 +3342,14 @@
 or(%22
-AdminWorker %7B%7D: Job %7B%7D: No
+No job
  run
@@ -3432,16 +3432,20 @@
 ion(%22No 
+job 
 runner f
@@ -3482,61 +3482,22 @@
 fo(%22
-AdminWorker %7B%7D: Job %7B%7D: Start.%22, name, job.getClass()
+Starting job.%22
 );%0A 
@@ -3553,181 +3553,88 @@
 fo(%22
-AdminWorker %7B%7D: Job %7B%7D: Success.%22, name, job.getClass());%0A      %7D catch (Throwable t) %7B%0A         log.error(%22AdminWorker %7B%7D: Job %7B%7D: Failed to execute.%22, name, job.getClass()
+Job succeeded.%22);%0A      %7D catch (Throwable t) %7B%0A         log.error(%22Job failed.%22
 , t)
@@ -3684,25 +3684,18 @@
 on(%22
-Failed to execute
+Job failed
 .%22, 
@@ -3739,34 +3739,19 @@
 fo(%22
-AdminWorker %7B%7D: Job %7B%7D: En
+Job finishe
 d.%22,

Rework logging