Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Merge branch 'master' of github.com:sunng87/clojalk

  • Loading branch information...
commit b7d6e37739e22055ff91fa9cf02a71b96e03bb54 2 parents 30086d7 + 1ad2579
Ning Sun authored
98 README.md
View
@@ -1,13 +1,103 @@
# clojalk [![Build Status](https://secure.travis-ci.org/sunng87/clojalk.png)](http://travis-ci.org/) #
-A distributed task queue in clojure. A Beanstalkd clone.
+A distributed task queue written pure in clojure. A Beanstalkd clone.
## Usage ##
-FIXME: write
+### Installation ###
+
+Clojalk is still in development so we don't have a packaged release.
+To use clojalk, you should checkout the code base and build
+it by yourself. This is not difficult task but be sure you have
+leiningen installed.
+
+ git clone git@github.com:sunng87/clojalk.git
+ cd clojalk
+ lein uberjar clojalk.jar
+
+To start a clojalk server:
+
+ java -jar clojalk.jar [clojalk.properties]
+
+Clojalk will load a property file "clojalk.properties" from current
+working directory if you don't specify a custom file path from command
+line.
+
+Also you can start clojalk from code base with lein. This is only for
+test purpose:
+
+ lein run
+
+Try out your installation:
+
+ telnet 127.0.0.1 12026
+
+You should be familiar with beanstalkd's memcached-styled protocol.
+
+### Protocol ###
+
+Clojalk is almost fully compatible with Beanstalkd's protocol. So you
+can refer to the [protocol
+document](https://github.com/kr/beanstalkd/blob/master/doc/protocol.txt
+"Beanstalkd Protocol") of Beanstalkd which also works with clojalk.
+
+Commands supported by clojalk are listed here.
+
+Tube operations:
+
+* watch
+* use
+* ignore
+* pause-tube
+
+Job life-cycle operations:
+
+* put
+* reserve
+* reserve-with-timeout
+* delete
+* release
+* bury
+* kick
+* touch
+
+Monitoring commands:
+
+* stats
+* stats-job
+* stats-tube
+* list-tubes-watched
+* list-tube-used
+* list-tubes
+* peek
+* peek-ready
+* peek-delayed
+* peek-buried
+
+### Clients ###
+
+The clojure client [beanstalk](https://github.com/sunng87/beanstalk
+"beanstalk") is forked and maintained by me, which works with clojalk
+and beanstalkd.
+
+More clients to be tested against clojalk.
+
+## Thanks ##
+
+I should thanks [Keith Rarick](https://github.com/kr "Keith Rarick")
+who designed beanstalkd and its protocol.
+
+And also I received great help from [Zach
+Tellman](https://github.com/ztellman "Zach Tellman") on implementing
+the protocol with gloss.
+
+### Contributors ###
+
+* [xiaonaitong](https://github.com/xiaonaitong "xiaonaitong")
## License ###
-Copyright (C) 2011 Sun Ning
+Copyright (C) 2011 [Sun Ning](http://sunng.info/ "Sun Ning")
+
+Distributed under the Eclipse Public License, the same as Clojure uses.
-TODO
4 scripts/bench.clj
View
@@ -70,8 +70,8 @@
(.setDaemon t false)
(.start t)))
-(doall (map run-in-thread (take 5 (repeat producer))))
-(doall (map run-in-thread (take 10 (repeat worker))))
+(dorun (map run-in-thread (take 5 (repeat producer))))
+(dorun (map run-in-thread (take 10 (repeat worker))))
(run-in-thread monitor)
(println "benchmark started")
2  scripts/put.clj
View
@@ -41,6 +41,6 @@
(time
(do
- (doall (map run-in-thread (take total-clients (repeat do-put-jobs))))
+ (dorun (map run-in-thread (take total-clients (repeat do-put-jobs))))
(.await latch)))
24 src/clojalk/core.clj
View
@@ -105,8 +105,7 @@
(alter jobs assoc (:id job) updated-top-job)
(dequeue-waiting-session session)
(alter session assoc :incoming_job updated-top-job)
- (alter session assoc :reserved_jobs
- (conj (:reserved_jobs @session) (:id updated-top-job)))
+ (alter session update-in [:reserved_jobs] conj (:id updated-top-job))
(clojalk.wal/write-job updated-top-job false)
(schedule #(update-expired-job (:id updated-top-job)) (:ttr job))
updated-top-job)))
@@ -127,7 +126,7 @@
(defn- set-job-as-ready [job]
(let [tube ((:tube job) @tubes)]
(do
- (alter jobs assoc (:id job) (assoc job :state :ready))
+ (alter jobs update-in [(:id job)] (fnil assoc job) :state :ready)
(alter (:ready_set tube) conj job)
(if-let [s (first @(:waiting_list tube))]
(reserve-job s job)))))
@@ -157,7 +156,7 @@
(let [session (@sessions id)]
(dosync
(dequeue-waiting-session session)
- (doall (map #(set-job-as-ready (@jobs %)) (:reserved_jobs @session)))
+ (dorun (map #(set-job-as-ready (@jobs %)) (:reserved_jobs @session)))
(alter sessions dissoc id))))
;; ## Macros for convenience of creating and executing commands
@@ -181,7 +180,7 @@
;; `put` is a producer task. It will create a new job according to information passed in.
;; When server is in drain mode, it does not store the job and return nil.
-;; If dealy is not zero, the job will be created as a delayed job. Delayed
+;; If delay is not zero, the job will be created as a delayed job. Delayed
;; job could not be reserved until it's timeout and ready.
(defcommand "put" [session priority delay ttr body]
(if-not @drain
@@ -279,8 +278,7 @@
(if (= (:state job) :ready)
(alter (:ready_set tube) disj job))
(alter session assoc :incoming_job nil)
- (alter session assoc :reserved_jobs
- (disj (:reserved_jobs @session) (:id job)))
+ (alter session update-in [:reserved_jobs] disj (:id job))
(if (empty? (:reserved_jobs @session))
(alter session assoc :state :idle)))
(clojalk.wal/write-job (assoc job :state :invalid) false)
@@ -311,8 +309,7 @@
(schedule #(update-delayed-job (:id updated-job)) delay))
(set-job-as-ready (assoc updated-job :state :ready)))
(alter session assoc :incoming_job nil)
- (alter session assoc :reserved_jobs
- (disj (:reserved_jobs @session) (:id updated-job)))
+ (alter session update-in [:reserved_jobs] disj (:id job))
(if (empty? (:reserved_jobs @session))
(alter session assoc :state :idle)))
(clojalk.wal/write-job updated-job false)
@@ -335,8 +332,7 @@
(alter (:buried_list tube) conj updated-job)
(alter jobs assoc (:id updated-job) updated-job)
(alter session assoc :incoming_job nil)
- (alter session assoc :reserved_jobs
- (disj (:reserved_jobs @session) (:id updated-job)))
+ (alter session update-in [:reserved_jobs] disj (:id job))
(if (empty? (:reserved_jobs @session))
(alter session assoc :state :idle)))
(clojalk.wal/write-job updated-job false)
@@ -396,7 +392,7 @@
(dosync
(if-not (contains? @tubes tube-name-kw)
(alter tubes assoc tube-name-kw (make-tube tube-name)))
- (alter session assoc :watch (conj (:watch @session) tube-name-kw))
+ (alter session update-in [:watch] conj tube-name-kw)
session)))
;; `ignore` is a worker command to remove tube from watching list.
@@ -405,7 +401,7 @@
(let [tube-name-kw (keyword tube-name)]
(dosync
(if (> (count (:watch @session)) 1)
- (alter session assoc :watch (disj (:watch @session) tube-name-kw))))
+ (alter session update-in [:watch] disj tube-name-kw)))
session))
;; stats command. list tubes names.
@@ -547,7 +543,7 @@
:timeouts (inc (:timeouts job)))]
(clojalk.wal/write-job updated-job false)
(dosync
- (alter session assoc :reserved_jobs (disj (:reserved_jobs @session) (:id updated-job)))
+ (alter session update-in [:reserved_jobs] disj (:id updated-job))
(alter job-timeouts inc)
(set-job-as-ready updated-job))))))
2  src/clojalk/wal.clj
View
@@ -213,7 +213,7 @@
(if-let [bin-log-files (scan-dir *clojalk-log-dir*)]
(do
(dosync
- (doall (map #(read-file % replay-handler) bin-log-files))
+ (dorun (map #(read-file % replay-handler) bin-log-files))
(replay-tubes))
(println (str (count @clojalk.data/jobs) " jobs loaded from write-ahead logs."))
(update-id-counter)))
Please sign in to comment.
Something went wrong with that request. Please try again.