allow to set a max number of failed items to keep via FailQueueStrategy

src/main/java/net/greghaines/jesque/worker/DefaultFailQueueStrategy.java

@@ -21,7 +21,8 @@
 import net.greghaines.jesque.utils.JesqueUtils;
 
 /**
- * DefaultFailQueueStrategy puts all jobs in the standard Redis failure queue.
+ * DefaultFailQueueStrategy puts all jobs in the standard Redis failure queue
+ * and imposes no limits on the max number of failed items to keep.
  */
 public class DefaultFailQueueStrategy implements FailQueueStrategy {
 
@@ -42,4 +43,12 @@ public DefaultFailQueueStrategy(final String namespace) {
     public String getFailQueueKey(final Throwable thrwbl, final Job job, final String curQueue) {
         return JesqueUtils.createKey(this.namespace, FAILED);
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public int getFailQueueMaxItems(String curQueue) {
+        return 0;
+    }
 }

src/main/java/net/greghaines/jesque/worker/FailQueueStrategy.java

@@ -32,4 +32,12 @@ public interface FailQueueStrategy {
      * @return the key of the failure queue to put the job into or null
      */
     String getFailQueueKey(Throwable thrwbl, Job job, String curQueue);
+
+    /**
+     * Determine the max number of items to keep in the failure queue.
+     * Returning a value <1 means that there is no limit.
+     * @param curQueue the queue the Job came from
+     * @return the max number of items to keep for the failure queue
+     */
+    int getFailQueueMaxItems(String curQueue);
 }

src/main/java/net/greghaines/jesque/worker/WorkerImpl.java

@@ -51,6 +51,7 @@
 import net.greghaines.jesque.utils.ScriptUtils;
 import net.greghaines.jesque.utils.VersionUtils;
 import redis.clients.jedis.Jedis;
+import redis.clients.jedis.Transaction;
 import redis.clients.jedis.exceptions.JedisException;
 
 /**
@@ -677,9 +678,21 @@ protected void failure(final Throwable thrwbl, final Job job, final String curQu
         try {
             this.jedis.incr(key(STAT, FAILED));
             this.jedis.incr(key(STAT, FAILED, this.name));
-            final String failQueueKey = this.failQueueStrategyRef.get().getFailQueueKey(thrwbl, job, curQueue);
+            final FailQueueStrategy strategy = this.failQueueStrategyRef.get();
+            final String failQueueKey = strategy.getFailQueueKey(thrwbl, job, curQueue);
             if (failQueueKey != null) {
-                this.jedis.rpush(failQueueKey, failMsg(thrwbl, curQueue, job));
+                final int failQueueMaxItems = strategy.getFailQueueMaxItems(curQueue);
+                if (failQueueMaxItems > 0) {
+                    Long currentItems = this.jedis.llen(failQueueKey);
+                    if (currentItems >= failQueueMaxItems) {
+                        Transaction tx = this.jedis.multi();
+                        tx.ltrim(failQueueKey, 1, -1);
+                        tx.rpush(failQueueKey, failMsg(thrwbl, curQueue, job));
+                        tx.exec();
+                    }
+                } else {
+                    this.jedis.rpush(failQueueKey, failMsg(thrwbl, curQueue, job));
+                }
             }
         } catch (JedisException je) {
             LOG.warn("Error updating failure stats for throwable=" + thrwbl + " job=" + job, je);

src/main/java/net/greghaines/jesque/worker/WorkerPoolImpl.java

@@ -49,6 +49,7 @@
 import org.slf4j.LoggerFactory;
 
 import redis.clients.jedis.Jedis;
+import redis.clients.jedis.Transaction;
 import redis.clients.jedis.exceptions.JedisException;
 import redis.clients.util.Pool;
 
@@ -675,9 +676,21 @@ protected void failure(final Throwable thrwbl, final Job job, final String curQu
                 public Void doWork(final Jedis jedis) throws IOException {
                     jedis.incr(key(STAT, FAILED));
                     jedis.incr(key(STAT, FAILED, name));
-                    final String failQueueKey = failQueueStrategyRef.get().getFailQueueKey(thrwbl, job, curQueue);
+                    final FailQueueStrategy strategy = failQueueStrategyRef.get();
+                    final String failQueueKey = strategy.getFailQueueKey(thrwbl, job, curQueue);
                     if (failQueueKey != null) {
-                        jedis.rpush(failQueueKey, failMsg(thrwbl, curQueue, job));
+                        final int failQueueMaxItems = strategy.getFailQueueMaxItems(curQueue);
+                        if (failQueueMaxItems > 0) {
+                            Long currentItems = jedis.llen(failQueueKey);
+                            if (currentItems >= failQueueMaxItems) {
+                                Transaction tx = jedis.multi();
+                                tx.ltrim(failQueueKey, 1, -1);
+                                tx.rpush(failQueueKey, failMsg(thrwbl, curQueue, job));
+                                tx.exec();
+                            }
+                        } else {
+                            jedis.rpush(failQueueKey, failMsg(thrwbl, curQueue, job));
+                        }
                     }
                     return null;
                 }