<commit_before>package com.s24.redjob.worker;

import java.lang.management.ManagementFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.s24.redjob.worker.events.JobExecute;
import com.s24.redjob.worker.events.JobFailure;
import com.s24.redjob.worker.events.JobProcess;
import com.s24.redjob.worker.events.JobSkipped;
import com.s24.redjob.worker.events.JobStart;
import com.s24.redjob.worker.events.JobSuccess;

/**
 * Base implementation of {@link Worker}.
 */
public abstract class AbstractWorker<S extends WorkerState> implements Worker, ApplicationEventPublisherAware, DisposableBean {
   /**
    * Log.
    */
   protected final Logger log = LoggerFactory.getLogger(getClass());

   /**
    * Placeholder for the worker id.
    */
   public static final String ID = "[id]";

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

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

   /**
    * Name of this worker. Defaults to a generated unique name.
    */
   protected String name;

   /**
    * Worker dao.
    */
   protected WorkerDao workerDao;

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

   /**
    * Default: Number of milliseconds the worker pauses, if none of the queues contained a job.
    */
   public static final int DEFAULT_EMPTY_QUEUE_SLEEP_MILLIS = 500;

   /**
    * Number of milliseconds the worker pauses, if none of the queues contained a job. Defaults to {@value
    * #DEFAULT_EMPTY_QUEUE_SLEEP_MILLIS}.
    */
   protected long emptyQueuesSleepMillis = DEFAULT_EMPTY_QUEUE_SLEEP_MILLIS;

   /**
    * Should worker run?.
    */
   protected final AtomicBoolean run = new AtomicBoolean(true);

   /**
    * Worker state.
    */
   protected S state;

   /**
    * Event bus.
    */
   protected ApplicationEventPublisher eventBus;

   /**
    * Init.
    */
   @PostConstruct
   public void afterPropertiesSet() throws Exception {
      Assert.notNull(jobRunnerFactory, "Precondition violated: jobRunnerFactory != null.");
      Assert.isTrue(emptyQueuesSleepMillis > 0, "Precondition violated: emptyQueuesSleepMillis > 0.");
      Assert.notNull(eventBus, "Precondition violated: eventBus != null.");

      id = IDS.incrementAndGet();
      if (StringUtils.hasLength(name)) {
         name = resolvePlaceholders(name);
      } else {
         name = createName();
      }
   }

   @Override
   public void destroy() {
      // TODO markus 2016-06-15: Interrupt too?
      stop();
   }

   /**
    * Resolve placeholder in custom worker names.
    *
    * Supported placeholders:
    * <ul>
    *    <li>id: Worker id.</li>
    *    <li>hostname: Hostname without domain.</li>
    *    <li>full-hostname: Hostname with domain.</li>
    * </ul>
    */
   protected String resolvePlaceholders(String name) throws Exception {
      name = name.replaceAll(Pattern.quote(ID), Long.toString(id));

      return HostnameResolver.resolve(name);
   }

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

   @Override
   public String getNamespace() {
      return workerDao.getNamespace();
   }

   @Override
   public int getId() {
      return id;
   }

   @Override
   public String getName() {
      return name;
   }

   /**
    * Set worker state to the given values.
    */
   protected void setWorkerState(String state) {
      this.state.setState(state);
      saveWorkerState();
   }

   /**
    * Save worker state.
    */
   protected void saveWorkerState() {
      try {
         workerDao.state(name, this.state);
      } catch (Exception e) {
         log.error("Failed to set worker state to {}.", state);
      }
   }

   @Override
   public void stop() {
      log.info("Stopping worker {}.", getName());
      run.set(false);
      setWorkerState(WorkerState.STOPPING);
   }

   /**
    * Process job. Sends {@link JobProcess} event.<commit_msg>Fix NPE<commit_after>package com.s24.redjob.worker;

import com.s24.redjob.worker.events.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.lang.management.ManagementFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;

/**
 * Base implementation of {@link Worker}.
 */
public abstract class AbstractWorker<S extends WorkerState> implements Worker, ApplicationEventPublisherAware, DisposableBean {
   /**
    * Log.
    */
   protected final Logger log = LoggerFactory.getLogger(getClass());

   /**
    * Placeholder for the worker id.
    */
   public static final String ID = "[id]";

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

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

   /**
    * Name of this worker. Defaults to a generated unique name.
    */
   protected String name;

   /**
    * Worker dao.
    */
   protected WorkerDao workerDao;

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

   /**
    * Default: Number of milliseconds the worker pauses, if none of the queues contained a job.
    */
   public static final int DEFAULT_EMPTY_QUEUE_SLEEP_MILLIS = 500;

   /**
    * Number of milliseconds the worker pauses, if none of the queues contained a job. Defaults to {@value
    * #DEFAULT_EMPTY_QUEUE_SLEEP_MILLIS}.
    */
   protected long emptyQueuesSleepMillis = DEFAULT_EMPTY_QUEUE_SLEEP_MILLIS;

   /**
    * Should worker run?.
    */
   protected final AtomicBoolean run = new AtomicBoolean(true);

   /**
    * Worker state.
    */
   protected S state;

   /**
    * Event bus.
    */
   protected ApplicationEventPublisher eventBus;

   /**
    * Init.
    */
   @PostConstruct
   public void afterPropertiesSet() throws Exception {
      Assert.notNull(jobRunnerFactory, "Precondition violated: jobRunnerFactory != null.");
      Assert.isTrue(emptyQueuesSleepMillis > 0, "Precondition violated: emptyQueuesSleepMillis > 0.");
      Assert.notNull(eventBus, "Precondition violated: eventBus != null.");

      id = IDS.incrementAndGet();
      if (StringUtils.hasLength(name)) {
         name = resolvePlaceholders(name);
      } else {
         name = createName();
      }
   }

   @Override
   public void destroy() {
      // TODO markus 2016-06-15: Interrupt too?
      stop();
   }

   /**
    * Resolve placeholder in custom worker names.
    *
    * Supported placeholders:
    * <ul>
    *    <li>id: Worker id.</li>
    *    <li>hostname: Hostname without domain.</li>
    *    <li>full-hostname: Hostname with domain.</li>
    * </ul>
    */
   protected String resolvePlaceholders(String name) throws Exception {
      name = name.replaceAll(Pattern.quote(ID), Long.toString(id));

      return HostnameResolver.resolve(name);
   }

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

   @Override
   public String getNamespace() {
      return workerDao.getNamespace();
   }

   @Override
   public int getId() {
      return id;
   }

   @Override
   public String getName() {
      return name;
   }

   /**
    * Set worker state to the given values.
    */
   protected void setWorkerState(String state) {
      if (this.state != null) {
         this.state.setState(state);
         saveWorkerState();
      }
   }

   /**
    * Save worker state.
    */
   protected void saveWorkerState() {
      try {
         workerDao.state(name, this.state);
      } catch (Exception e) {
         log.error("Failed to set worker state to {}.", state);
      }
   }

   @Override
   public void stop() {
      log.info("Stopping worker {}.", getName());
      run.set(false);
      setWorkerState(WorkerState.STOPPING);
   }

   /**
    * Process job. Sends {@link JobProcess} event.
{'fraction_non_alphanumeric': 0.11284980011422044, 'fraction_numerical': 0.00548258138206739, 'mean_word_length': 2.8036490008688095, 'pattern_counts': {'":': 2, '<': 21, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1213779', 'n_tokens_mistral': 2652, 'n_tokens_neox': 2545, 'n_words': 641}