Add something

pom.xml

@@ -20,7 +20,7 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <jedis.version>2.9.0</jedis.version>
-        <jackson.version>2.8.5</jackson.version>
+        <jackson.version>2.9.5</jackson.version>
         <slf4j.version>1.7.21</slf4j.version>
         <logback.version>1.1.7</logback.version>
         <junit.version>4.12</junit.version>
@@ -137,8 +137,8 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>3.5.1</version>
                 <configuration>
-                    <source>1.7</source>
-                    <target>1.7</target>
+                    <source>1.8</source>
+                    <target>1.8</target>
                 </configuration>
             </plugin>
             <plugin>

src/main/java/net/greghaines/jesque/DuplicateJobException.java

@@ -0,0 +1,13 @@
+package net.greghaines.jesque;
+
+/**
+ * Created by dimav
+ * on 14/09/17 22:23.
+ */
+public class DuplicateJobException extends RuntimeException{
+
+    public DuplicateJobException(final String message) {
+
+        super(message);
+    }
+}

src/main/java/net/greghaines/jesque/RetryJobException.java

@@ -0,0 +1,18 @@
+package net.greghaines.jesque;
+
+/**
+ * Created by dimav
+ * on 14/09/17 22:23.
+ */
+public class RetryJobException extends RuntimeException{
+    private final long delay;
+
+    public RetryJobException(final String message, long delay) {
+        super(message);
+        this.delay = delay;
+    }
+
+    public long getDelay() {
+        return delay;
+    }
+}

src/main/java/net/greghaines/jesque/client/AbstractClient.java

@@ -19,26 +19,42 @@
 import static net.greghaines.jesque.utils.ResqueConstants.QUEUES;
 
 import net.greghaines.jesque.Config;
+import net.greghaines.jesque.DuplicateJobException;
 import net.greghaines.jesque.Job;
 import net.greghaines.jesque.json.ObjectMapperFactory;
 import net.greghaines.jesque.utils.JedisUtils;
 import net.greghaines.jesque.utils.JesqueUtils;
+import net.greghaines.jesque.utils.ScriptUtils;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.Transaction;
+import redis.clients.jedis.exceptions.JedisException;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Common logic for Client implementations.
- * 
+ *
  * @author Greg Haines
  * @author Animesh Kumar
  */
 public abstract class AbstractClient implements Client {
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractClient.class);
 
+    private static final String ENQUEUE = "enqueue";
+    private static final String PRIORITY_ENQUEUE = "priorityEnqueue";
+    private static final String DELAYED_ENQUEUE = "delayedEnqueue";
+    private static final String DUPLICATED = "duplicated";
     private final String namespace;
+    protected boolean jobUniquenessValidation;
+    private static final String PUSH_LUA = "/clientScripts/jesque_push.lua";
+    private static final AtomicReference<String> pushScriptHash = new AtomicReference<>(null);
 
     /**
      * Constructor.
-     * 
+     *
      * @param config
      *            used to get the namespace for key creation
      */
@@ -56,9 +72,34 @@ protected String getNamespace() {
         return this.namespace;
     }
 
+    protected boolean getJobUniquenessValidation() {
+        return jobUniquenessValidation;
+    }
+
+    private static boolean isScriptLoaded = false;
+
+    static void loadScript(Jedis jedis) {
+        if(!isScriptLoaded) {
+            synchronized (AbstractClient.class) {
+                if(!isScriptLoaded) {
+                    doLoadScript(jedis);
+                    isScriptLoaded = true;
+                }
+            }
+        }
+    }
+
+    private static void doLoadScript(final Jedis jedis) {
+        try {
+            pushScriptHash.set(jedis.scriptLoad(ScriptUtils.readScript(PUSH_LUA)));
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     /**
      * Builds a namespaced Redis key with the given arguments.
-     * 
+     *
      * @param parts
      *            the key parts to be joined
      * @return an assembled String key
@@ -67,6 +108,8 @@ protected String key(final String... parts) {
         return JesqueUtils.createKey(this.namespace, parts);
     }
 
+    abstract Jedis getJedis();
+
     /**
      * {@inheritDoc}
      */
@@ -75,6 +118,9 @@ public void enqueue(final String queue, final Job job) {
         validateArguments(queue, job);
         try {
             doEnqueue(queue, ObjectMapperFactory.get().writeValueAsString(job));
+        } catch (JedisException re) {
+            doLoadScript(getJedis());
+            throw re;
         } catch (RuntimeException re) {
             throw re;
         } catch (Exception e) {
@@ -90,6 +136,9 @@ public void priorityEnqueue(final String queue, final Job job) {
         validateArguments(queue, job);
         try {
             doPriorityEnqueue(queue, ObjectMapperFactory.get().writeValueAsString(job));
+        } catch (JedisException re) {
+            doLoadScript(getJedis());
+            throw re;
         } catch (RuntimeException re) {
             throw re;
         } catch (Exception e) {
@@ -122,7 +171,7 @@ public boolean acquireLock(final String lockName, final String lockHolder, final
 
     /**
      * Actually enqueue the serialized job.
-     * 
+     *
      * @param queue
      *            the queue to add the Job to
      * @param msg
@@ -134,7 +183,7 @@ public boolean acquireLock(final String lockName, final String lockHolder, final
 
     /**
      * Actually enqueue the serialized job with high priority.
-     * 
+     *
      * @param queue
      *            the queue to add the Job to
      * @param msg
@@ -146,7 +195,7 @@ public boolean acquireLock(final String lockName, final String lockHolder, final
 
     /**
      * Actually acquire the lock based upon the client acquisition model.
-     * 
+     *
      * @param lockName
      *            the name of the lock to acquire
      * @param timeout
@@ -157,13 +206,13 @@ public boolean acquireLock(final String lockName, final String lockHolder, final
      * @throws Exception
      *             in case something goes wrong
      */
-    protected abstract boolean doAcquireLock(final String lockName, final String lockHolder, 
+    protected abstract boolean doAcquireLock(final String lockName, final String lockHolder,
             final int timeout) throws Exception;
 
     /**
      * Helper method that encapsulates the minimum logic for adding a job to a
      * queue.
-     * 
+     *
      * @param jedis
      *            the connection to Redis
      * @param namespace
@@ -174,14 +223,20 @@ protected abstract boolean doAcquireLock(final String lockName, final String loc
      *            the job serialized as JSON
      */
     public static void doEnqueue(final Jedis jedis, final String namespace, final String queue, final String jobJson) {
-        jedis.sadd(JesqueUtils.createKey(namespace, QUEUES), queue);
-        jedis.rpush(JesqueUtils.createKey(namespace, QUEUE, queue), jobJson);
+        doEnqueue(jedis,namespace,queue, jobJson , false);
+    }
+
+    static void doEnqueue(final Jedis jedis, final String namespace, final String queue, final String jobJson, boolean jobUniquenessValidation) {
+        final String queuesKey = JesqueUtils.createKey(namespace, QUEUES);
+        final String queueKey = JesqueUtils.createKey(namespace, QUEUE, queue);
+
+        doEnqueue(jedis, ENQUEUE, queue, jobJson, queueKey, queuesKey, jobUniquenessValidation);
     }
 
     /**
      * Helper method that encapsulates the minimum logic for adding a high
      * priority job to a queue.
-     * 
+     *
      * @param jedis
      *            the connection to Redis
      * @param namespace
@@ -192,13 +247,19 @@ public static void doEnqueue(final Jedis jedis, final String namespace, final St
      *            the job serialized as JSON
      */
     public static void doPriorityEnqueue(final Jedis jedis, final String namespace, final String queue, final String jobJson) {
-        jedis.sadd(JesqueUtils.createKey(namespace, QUEUES), queue);
-        jedis.lpush(JesqueUtils.createKey(namespace, QUEUE, queue), jobJson);
+        doPriorityEnqueue(jedis,namespace, queue, jobJson,false);
+    }
+
+    static void doPriorityEnqueue(final Jedis jedis, final String namespace, final String queue, final String jobJson , boolean jobUniquenessValidation) {
+        final String queuesKey = JesqueUtils.createKey(namespace, QUEUES);
+        final String queueKey = JesqueUtils.createKey(namespace, QUEUE, queue);
+
+        doEnqueue(jedis, PRIORITY_ENQUEUE, queue, jobJson, queueKey, queuesKey, jobUniquenessValidation);
     }
 
     /**
      * Helper method that encapsulates the logic to acquire a lock.
-     * 
+     *
      * @param jedis
      *            the connection to Redis
      * @param namespace
@@ -269,16 +330,49 @@ public static boolean doAcquireLock(final Jedis jedis, final String namespace, f
     }
 
     public static void doDelayedEnqueue(final Jedis jedis, final String namespace, final String queue, final String jobJson, final long future) {
-        final String key = JesqueUtils.createKey(namespace, QUEUE, queue);
+        doDelayedEnqueue(jedis,namespace,queue,jobJson,future, false);
+    }
+
+    static void doDelayedEnqueue(final Jedis jedis, final String namespace, final String queue, final String jobJson, final long future, boolean jobUniquenessValidation) {
+        final String queueKey = JesqueUtils.createKey(namespace, QUEUE, queue);
         // Add task only if this queue is either delayed or unused
-        if (JedisUtils.canUseAsDelayedQueue(jedis, key)) {
-            jedis.zadd(key, future, jobJson);
-            jedis.sadd(JesqueUtils.createKey(namespace, QUEUES), queue);
+        if (JedisUtils.canUseAsDelayedQueue(jedis, queueKey)) {
+            final String queuesKey = JesqueUtils.createKey(namespace, QUEUES);
+            doEnqueue(jedis, DELAYED_ENQUEUE, queue, jobJson, queueKey, queuesKey, jobUniquenessValidation, future);
         } else {
             throw new IllegalArgumentException(queue + " cannot be used as a delayed queue");
         }
     }
 
+    private static void doEnqueue(final Jedis jedis, final String enqueueType, final String queue, final String jobJson, final String queueKey, final String queuesKey, final boolean jobUniquenessValidation) {
+        doEnqueue(jedis, enqueueType, queue, jobJson, queueKey, queuesKey, jobUniquenessValidation, 0L);
+    }
+
+    private static void doEnqueue(final Jedis jedis, final String enqueueType, final String queue, final String jobJson, final String queueKey, final String queuesKey, final boolean jobUniquenessValidation, final long future) {
+        final String pushStatus;
+        LOG.info("Jesque starting {} job {} to the queue {} with delay {} ", enqueueType, jobJson, queue, future);
+        try {
+            final String uniquenessValidation = String.valueOf(jobUniquenessValidation);
+
+            if(future==0) {
+                pushStatus = (String) jedis.evalsha(pushScriptHash.get(), 2, queuesKey, queueKey, enqueueType, getCurrTime(), queue, jobJson, uniquenessValidation);
+            } else{
+                pushStatus = (String) jedis.evalsha(pushScriptHash.get(), 2, queuesKey, queueKey, enqueueType, getCurrTime(), queue, jobJson, uniquenessValidation, String.valueOf(future));
+            }
+        } catch (Exception e) {
+            LOG.error("Jesque {} job {} to the queue {} has failed", enqueueType, jobJson, queue, e);
+            throw e;
+        }
+
+        if(DUPLICATED.equals(pushStatus)){
+            final String duplicatedJobMessage = String.format("Jesque duplicated job %s has been found", jobJson);
+            LOG.warn(duplicatedJobMessage,jobJson);
+            throw new DuplicateJobException(duplicatedJobMessage);
+        }
+
+        LOG.info("Jesque {} job {} to the queue {} has finished successfully", enqueueType, jobJson, queue);
+    }
+
     protected abstract void doDelayedEnqueue(String queue, String msg, long future) throws Exception;
 
     /**
@@ -289,6 +383,9 @@ public void delayedEnqueue(final String queue, final Job job, final long future)
         validateArguments(queue, job, future);
         try {
             doDelayedEnqueue(queue, ObjectMapperFactory.get().writeValueAsString(job), future);
+        } catch (JedisException re) {
+            doLoadScript(getJedis());
+            throw re;
         } catch (RuntimeException re) {
             throw re;
         } catch (Exception e) {
@@ -411,4 +508,8 @@ private static void validateArguments(final String queue, final Job job, final l
             throw new IllegalArgumentException("frequency must be greater than one second");
         }
     }
+
+    private static String getCurrTime() {
+        return String.valueOf(System.currentTimeMillis());
+    }
 }

src/main/java/net/greghaines/jesque/client/ClientImpl.java

@@ -62,12 +62,35 @@ public ClientImpl(final Config config) {
      *             if the config is null
      */
     public ClientImpl(final Config config, final boolean checkConnectionBeforeUse) {
+        this(config,checkConnectionBeforeUse,false);
+    }
+
+    /**
+     * Create a new ClientImpl, which creates it's own connection to Redis using
+     * values from the config.
+     *
+     * @param config
+     *            used to create a connection to Redis
+     * @param checkConnectionBeforeUse
+     *            check to make sure the connection is alive before using it
+     *  @param jobUniquenessValidation avoid duplicate jobs submission
+     * @throws IllegalArgumentException
+     *             if the config is null
+     */
+    public ClientImpl(final Config config, final boolean checkConnectionBeforeUse, boolean jobUniquenessValidation) {
         super(config);
         this.config = config;
         this.jedis = new Jedis(config.getHost(), config.getPort(), config.getTimeout());
         authenticateAndSelectDB();
         this.checkConnectionBeforeUse = checkConnectionBeforeUse;
         this.keepAliveService = null;
+        this.jobUniquenessValidation = jobUniquenessValidation;
+        loadScript(jedis);
+    }
+
+    @Override
+    Jedis getJedis() {
+        return jedis;
     }
 
     /**
@@ -106,7 +129,7 @@ public void run() {
     @Override
     protected void doEnqueue(final String queue, final String jobJson) {
         ensureJedisConnection();
-        doEnqueue(this.jedis, getNamespace(), queue, jobJson);
+        doEnqueue(this.jedis, getNamespace(), queue, jobJson, getJobUniquenessValidation());
     }
 
     /**
@@ -115,7 +138,7 @@ protected void doEnqueue(final String queue, final String jobJson) {
     @Override
     protected void doPriorityEnqueue(final String queue, final String jobJson) {
         ensureJedisConnection();
-        doPriorityEnqueue(this.jedis, getNamespace(), queue, jobJson);
+        doPriorityEnqueue(this.jedis, getNamespace(), queue, jobJson, getJobUniquenessValidation());
     }
 
     /**
@@ -145,7 +168,7 @@ public void end() {
     @Override
     protected void doDelayedEnqueue(final String queue, final String msg, final long future) throws Exception {
         ensureJedisConnection();
-        doDelayedEnqueue(this.jedis, getNamespace(), queue, msg, future);
+        doDelayedEnqueue(this.jedis, getNamespace(), queue, msg, future, getJobUniquenessValidation());
     }
 
     /**