per bfs, preemptively answer the 'why not kafka?' question

Author: ztellman

README.md

@@ -1,6 +1,6 @@
 ![](docs/EasterIsland.jpg)
 
-This library implements a disk-backed task queue, allowing for queues that can survive processes dying, and whose size is bounded by available disk rather than memory.
+This library implements a disk-backed task queue, allowing for queues that can survive processes dying, and whose size is bounded by available disk rather than memory.  It is a small, purely-Clojure implementation focused entirely on the in-process use case, meaning that it is both simpler and more easily embedded than network-aware queue implementations such as Kafka and ActiveMQ.
 
 ### usage
 
@@ -77,7 +77,7 @@ A complete list of options is as follows:
 | name | description |
 |------|-------------|
 | `:complete?` | a predicate for identifying already completed tasks, defaults to always returning false |
-| `:max-queue-size` | the maximum number of elements that can be in the queue before `put!` blocks, defaults to `Integer/MAX_VALUE` |
+| `:max-queue-size` | the maximum number of elements that can be in the queue before `put!` blocks, defaults to `Integer/MAX_VALUE`	 |
 | `:slab-size` | The size, in bytes, of the backing files for the queue.  Defaults to 16mb. |
 | `:fsync-put?` | Whether an fsync should be performed for each `put!`.  Defaults to true. |
 | `:fsync-take?` | Whether an fsync should be performed for each `take!`.  Defaults to false. |

src/durable_queue.clj

@@ -131,41 +131,42 @@
   ([slab]
      (slab->task-seq slab 0))
   ([slab pos]
-     (lazy-seq
-       (try
-         (let [^ByteBuffer
-               buf' (-> (buffer slab)
-                      .duplicate
-                      (.position pos))]
-           
-           ;; is there a next task, and is there space left in the buffer?
-           (when (and
-                   (pos? (.remaining buf'))
-                   (== 1 (.get buf')))
-             
-             (let [status (.get buf')
-                   checksum (.getLong buf')
-                   size (.getInt buf')]
-               (cons
-                 
-                 (vary-meta
-                   (task
-                     #(-> (buffer slab)
+     (seq
+       (lazy-seq
+         (try
+           (let [^ByteBuffer
+                 buf' (-> (buffer slab)
                         .duplicate
-                        (.position pos)
-                        ^ByteBuffer
-                        (.limit (+ pos header-size size))
-                        .slice)
-                     (read-write-lock slab))
-                   assoc ::slab slab)
-                 
-                 (slab->task-seq
-                   slab
-                   (+ pos header-size size))))))
-         (catch Throwable e
-           ;; this implies unrecoverable corruption
-           nil
-           )))))
+                        (.position pos))]
+             
+             ;; is there a next task, and is there space left in the buffer?
+             (when (and
+                     (pos? (.remaining buf'))
+                     (== 1 (.get buf')))
+               
+               (let [status (.get buf')
+                     checksum (.getLong buf')
+                     size (.getInt buf')]
+                 (cons
+                   
+                   (vary-meta
+                     (task
+                       #(-> (buffer slab)
+                          .duplicate
+                          (.position pos)
+                          ^ByteBuffer
+                          (.limit (+ pos header-size size))
+                          .slice)
+                       (read-write-lock slab))
+                     assoc ::slab slab)
+                   
+                   (slab->task-seq
+                     slab
+                     (+ pos header-size size))))))
+           (catch Throwable e
+             ;; this implies unrecoverable corruption
+             nil
+             ))))))
 
 (deftype TaskSlab
   [filename