Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Implement delete-q! function to delete a single queue #17

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Implement delete-q! function to delete a single queue #17

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3be4c89
Implement delete-q! function to delete a single queue
sundbry Apr 24, 2018
80346da
Rm dead code
sundbry Apr 24, 2018
7e9701d
Rm unnecessary lock
sundbry Apr 24, 2018
f0f3c90
Drop slab data on initialization that does not fit when slab size > m…
sundbry Apr 25, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion project.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
(defproject factual/durable-queue "0.1.6"
(defproject factual/durable-queue "0.1.7-SNAPSHOT"
: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"}
Expand Down
29 changes: 20 additions & 9 deletions src/durable_queue.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -432,6 +432,8 @@
(^:private mark-retry! [_ q-name])
(delete! [_]
"Deletes all files associated with the queues.")
(delete-q! [_ q-name]
"Deletes all files associated with a queue.")
(stats [_]
"Returns a map of queue names onto information about the immediate state of the queue.")
(fsync [_]
Expand Down Expand Up @@ -489,9 +491,7 @@

(.mkdirs (io/file directory))

(let [

queue (memoize (fn [_] (LinkedBlockingQueue. (int max-queue-size))))
(let [queue (memoize (fn [_] (LinkedBlockingQueue. (int max-queue-size))))
queue-name->files (directory->queue-name->slab-files directory)

;; core state stores
Expand Down Expand Up @@ -590,12 +590,13 @@
(swap! empty-slabs conj slab))

(do
(doseq [task tasks]
(status! task :incomplete)
(when-not (.offer q' task)
(throw
(IllegalArgumentException.
"'max-queue-size' insufficient to hold existing tasks."))))
(loop [tasks tasks]
(when-let [task (first tasks)]
(status! task :incomplete)
; Continue loading until data doesn't fit into the queue.
; Drop data in slabs that overflows queue size.
(when (.offer q' task)
(recur (rest tasks)))))
(unmap slab)))))

(let [^AtomicLong counter (get-in @queue-name->stats [q :enqueued])]
Expand Down Expand Up @@ -624,6 +625,16 @@
(unmap s)
(delete-slab s)))

(delete-q! [this q-name]
(let [q-name (munge (name q-name))]
(doseq [s (get @queue-name->slabs q-name)]
(unmap s)
(delete-slab s))
(.clear (queue q-name))
(swap! queue-name->stats assoc q-name nil)
(swap! queue-name->slabs assoc q-name nil)
(swap! queue-name->current-slab assoc q-name nil)))

(fsync [_]
(doseq [slab (->> @queue-name->slabs vals (apply concat))]
(sync! slab)))
Expand Down