make sure put! returns a boolean value, mark 0.1.3

Author: ztellman

README.md

@@ -5,7 +5,7 @@ This library implements a disk-backed task queue, allowing for queues that can s
 ### usage
 
 ```clj
-[factual/durable-queue "0.1.2"]
+[factual/durable-queue "0.1.3"]
 ```
 
 To interact with queues, first create a `queues` object by specifying a directory in the filesystem and an options map:

project.clj

@@ -1,10 +1,11 @@
-(defproject factual/durable-queue "0.1.2"
+(defproject factual/durable-queue "0.1.3"
   :description "a in-process task-queue that is backed by disk."
   :license {:name "Eclipse Public License"
             :url "http://www.eclipse.org/legal/epl-v10.html"}
   :dependencies [[com.taoensso/nippy "2.5.2"]
                  [primitive-math "0.1.3"]
-                 [byte-streams "0.1.9"]]
+                 [byte-streams "0.1.9"]
+                 [manifold "0.1.0-alpha3"]]
   :profiles {:dev {:dependencies [[org.clojure/clojure "1.5.1"]
                                   [criterium "0.4.3"]
                                   [codox-md "0.2.0" :exclusions [org.clojure/clojure]]]}}

src/durable_queue.clj

@@ -4,6 +4,7 @@
     [byte-streams :as bs]
     [clojure.string :as str]
     [primitive-math :as p]
+    [manifold.stream :as s]
     [taoensso.nippy :as nippy])
   (:import
     [java.lang.reflect
@@ -726,9 +727,7 @@
                    (let [^AtomicLong counter (get-in @queue-name->stats [q-name :enqueued])]
                      (.incrementAndGet counter))
                    true)
-                 false)
-
-               nil))
+                 false)))
 
            (put! [this q-name task-descriptor]
              (put! this q-name task-descriptor Long/MAX_VALUE))))
@@ -774,21 +773,27 @@
 (defn complete!
   "Marks a task as complete."
   [task]
-  (status! task :complete)
-  (when (-> task meta ::fsync?)
-    (sync! (:slab task)))
-  (mark-complete! @(-> task meta ::this) (-> task meta ::queue-name))
-  true)
+  (if (identical? :complete (status task))
+    false
+    (do
+      (status! task :complete)
+      (when (-> task meta ::fsync?)
+        (sync! (:slab task)))
+      (mark-complete! @(-> task meta ::this) (-> task meta ::queue-name))
+      true)))
 
 (defn retry!
   "Marks a task as available for retry."
   [task]
-  (status! task :incomplete)
-  (when (-> task meta ::fsync?)
-    (sync! (:slab task)))
-  (mark-retry! @(-> task meta ::this) (-> task meta ::queue-name))
-  (let [^LinkedBlockingQueue q (-> task meta ::queue)]
-    (.offer q
-      task
-      Long/MAX_VALUE
-      TimeUnit/MILLISECONDS)))
+  (if (or
+        (identical? :complete (status task))
+        (identical? :incomplete (status task)))
+    false
+    (do
+      (status! task :incomplete)
+      (when (-> task meta ::fsync?)
+        (sync! (:slab task)))
+      (mark-retry! @(-> task meta ::this) (-> task meta ::queue-name))
+      (let [^LinkedBlockingQueue q (-> task meta ::queue)]
+        (.put q task))
+      true)))