A Clojure library for Redis



dev dependencies


(this space intentionally left almost blank)
(ns accession.core
  (:refer-clojure :exclude [get set keys type sync sort])
  (:require [ :as io]
            [clojure.string :as str])
  (:import ( Socket)
           ( BufferedInputStream DataInputStream)))

The new unified protocol was introduced in Redis 1.2, but it became the standard way for talking with the Redis server in Redis 2.0. In the unified protocol all the arguments sent to the Redis server are binary safe. This is the general form:

  *<number of arguments> CR LF
  $<number of bytes of argument 1> CR LF
  <argument data> CR LF
  $<number of bytes of argument N> CR LF
  <argument data> CR LF

See the following example:


This is how the above command looks as a quoted string, so that it is possible to see the exact value of every byte in the query:

(defn query
  [name & args]
  (str "*"
       (+ 1 (count args)) "\r\n"
       "$" (count name) "\r\n"
       (str/upper-case name) "\r\n"
       (str/join "\r\n"
                 (map (fn [a] (str "$" (count (.getBytes (str a))) "\r\n" a))

These protocols were created to support pub/sub, in particular, subscribe. Subscribe is called in terms of both a connection-map and an open channel. This is not implemented via the macro because the semantics are so much different from that of the standard Redis calls. Since channels and subscriptions live longer than a single query, we needed a way to keep a connection open and close it later when we are finished.

(defprotocol ISubscribable
  (subscribe [this channels])
  #_(psubscribe [this channels]))
(defprotocol IUnsubscribable
  (unsubscribe [this channels])
  #_(punsubscribe [this channels]))
(defprotocol IRedisChannel
  (close [this]))

Redis will reply to commands with different kinds of replies. It is possible to check the kind of reply from the first byte sent by the server:

  • With a single line reply the first byte of the reply will be +
  • With an error message the first byte of the reply will be -
  • With an integer number the first byte of the reply will be :
  • With bulk reply the first byte of the reply will be $
  • With multi-bulk reply the first byte of the reply will be *
(defmulti response
  (fn [in] (char (.readByte in))))
(defmethod response \- [in]
  (.readLine in))
(defmethod response \+ [in]
  (.readLine in))
(defmethod response \$ [in]
  (let [length (Integer/parseInt (.readLine in))]
    (when (not= length -1)
      (let [content (byte-array (+ 2 length))]
        (.read in content)
        (String. content 0 length)))))
(defmethod response \: [in]
  (Long/parseLong (.readLine in)))
(defmethod response \* [in]
  (let [length (Integer/parseInt (.readLine in))]
    (doall (repeatedly length #(response in)))))

Creates an initial socket for consumption attached to the proper host and port. This socket is subject to modification later depending on the request use case or if a timeout is set.

(defn- socket 
  (doto (Socket. (:host spec) (:port spec))
    (.setTcpNoDelay true)
    (.setKeepAlive true)))

Responsible for actually making the request to the Redis server. Sets the timeout on the socket if one was specified.

(defn request
  [conn & query]
  (with-open [socket (doto (socket conn)
                       (.setSoTimeout (:timeout conn)))
              in (DataInputStream. (BufferedInputStream. (.getInputStream socket)))
              out (.getOutputStream socket)]
    (.write out (.getBytes (apply str query)))
    (if (next query)
      (doall (repeatedly (count query) #(response in)))
      (response in))))

Used in conjunction with an open channel to handle messages that arrive. Takes a channel spec and a message

(defn receive-message
  [channel-spec in]
  (let [next-message (response in)
        channel-name (second next-message)]
    (if-let [f (clojure.core/get @channel-spec channel-name)]
      (f next-message))))

Sends commands out to the specified channels

(defn write-commands
  [out command channels]
  (.write out (.getBytes (apply query command (clojure.core/keys channels)))))

This record implements the protocols defined above to provide the pub/sub infrastructure

(defrecord RedisChannel [channel-fns socket out]
  (subscribe [this channels]
    (do (swap! channel-fns merge channels)
        (write-commands out "subscribe" channels)))
  (unsubscribe [this channel]
    (do (swap! channel-fns dissoc channel)
        (.write out (.getBytes (query "unsubscribe" channel)))))
  (close [this]
    (.close socket)))

Takes a connection and sets up a connection to a Redis channel with the provided functions as callbacks to invoke when messages are received

(defn open-channel
  [conn command channels]
  (let [socket (socket conn)
        in (.getInputStream socket)
        out (.getOutputStream socket)
        in (DataInputStream. (BufferedInputStream. in))
        channel-fns (atom channels)]
    (write-commands out command channels)
    (future (doall (repeatedly #(receive-message channel-fns in))))
    (RedisChannel. channel-fns socket out)))
(extend-type clojure.lang.PersistentArrayMap
  (subscribe [this channels]
    (open-channel this "subscribe" channels)))

Creates the initial connection spec. Options can be overridden by passing in a map. The following keys are valid:

:host :port :password :socket :timeout

Although passing in your own socket is not recommended and will probably cause more problems than it solves

(defn connection-map
  ([] (connection-map {}))
     (let [default {:host ""
                    :port 6379
                    :password nil
                    :socket nil
                    :timeout 0}]
       (merge default spec))))
(defn with-connection [spec & body]
  (apply request spec body))

We would like to create one function for each command which Redis supports. The set function would look something like this:

(defn set [key value] (query "set" key value))

Similarly, the get function would be:

(defn get [key] (query "get" key))

Because each of these functions has the same pattern, we can use a macro to create them and save a lot of typing.

This function enables vararg style definitions in the queries. For example you can write:

   (mget [key & keys])

and the defquery macro will properly expand out into a variable argument function

(defn parameters
  (let [[args varargs] (split-with #(not= '& %)  params)]
    (conj (vec args) (last varargs))))

Given a redis command and a parameter list, create a function of the form:

   (defn <name> <parameter-list>
     (query <command> <p1> <p2> ... <pN>))

The name which is passed is a symbol and is first used as a symbol for the function name. We convert this symbol to a string and use that string as the command name.

params is a list of N symbols which represent the parameters to the function. We use this list as the parameter-list when we create the function. Each symbol in this list will be an argument to query after the command. We use unquote splicing (~@) to insert these arguments after the command string.

(defmacro defquery
  [name params]
  (let [command (str name)
        p (parameters params)]
    `(defn ~name ~params
       (apply query ~command ~@p))))

A call to:

(defqueries (set [key value]))

will expand to:

(do (defn set [key value] (query "set" key value)))

Given any number of redis commands and argument lists, convert them to function definitions.

This is an interesting use of unquote splicing. Unquote splicing works on a sequence and that sequence can be the result of a function call as it is here. The function which produces this sequence maps each argument passed to this macro over a function which takes each list like (set [key value]), binds it to q, and uses unquote splicing again to create a call to defquery which looks like (defquery set [key value]).

Finally, a macro can only return a single form, so we wrap all of the produced expressions in a do special form.

(defmacro defqueries
  [& queries]
  `(do ~@(map (fn [q] `(defquery ~@q)) queries)))
  (auth             [password])
  (quit             [])
  (select           [index])
  (bgwriteaof       [])
  (bgsave           [])
  (flushdb          [])
  (flushall         [])
  (dbsize           [])
  (info             [])
  (save             [])
  (sync             [])
  (lastsave         [])
  (shutdown         [])
  (slaveof          [host port])
  (slowlog          [command argument])
  ;; (config-get [parameter])
  ;; (config-set [parameter value])
  ;; (config-resetstat [])
  ;; (debug-object [key])
  ;; (debug-segfault)

  (echo             [message])
  (ping             [])
  (discard          [])
  (exec             [])
  (monitor          [])
  (multi            [])
  (object           [subcommand & arguments])
  (set              [key value])
  (setex            [key seconds value])
  (setnx            [key value])
  (setbit           [key offset value])
  (mset             [key value & pairs])
  (msetnx           [key value & pairs])
  (setrange         [key offset value])
  (get              [key])
  (mget             [key & keys])
  (getbit           [key offset])
  (getset           [key value])
  (getrange         [key start end])
  (append           [key value])
  (keys             [pattern])
  (exists           [key])
  (randomkey        [])
  (type             [key])
  (move             [key db])
  (rename           [key newkey])
  (renamenx         [key newkey])
  (strlen           [key])
  (watch            [key & keys])
  (unwatch          [])
  (del              [key & keys])
  (sort             [key & options])

  (incr             [key])
  (incrby           [key increment])
  (decr             [key])
  (decrby           [key increment])

  (expire           [key seconds])
  (expireat         [key seconds])
  (persist          [key])
  (ttl              [key])

  (llen             [key])
  (lpop             [key])
  (lpush            [key value])
  (lpushx           [key value & values])
  (lset             [key value index])
  (linsert          [key before-after pivot value])
  (lrem             [key count value])
  (ltrim            [key start stop])
  (lrange           [key start end])
  (lindex           [key index])
  (rpop             [key])
  (rpush            [key value & values])
  (rpushx           [key value])
  (rpoplpush        [source destination])
  (blpop            [key timeout])
  (brpop            [key timeout])
  (brpoplpush       [source destination timeout])

  (hset             [key field value])
  (hmset            [key field value & pairs])
  (hsetnx           [key field value])
  (hget             [key field])
  (hmget            [key field & fields])
  (hexists          [key field])
  (hdel             [key field & fields])
  (hgetall          [key])
  (hincrby          [key field increment])
  (hkeys            [key])
  (hvals            [key])
  (hlen             [key])
  (sadd             [key member & members])
  (srem             [key member & members])
  (sismember        [key member])
  (smembers         [key])
  (sunion           [key & keys])
  (sunionstore      [destination key & keys])
  (sdiff            [key & keys])
  (sdiffstore       [destination set1 set2])
  (sinter           [key & keys])
  (sinterstore      [destination key & keys])
  (scard            [key])
  (smove            [source desination member])
  (spop             [key])
  (srandmember      [key])

  (zadd             [key score member & more])
  (zrange           [key start end])
  (zrangebyscore    [key min max & more])
  (zrevrange        [key start stop & more])
  (zrevrangebyscore [key max min & more])
  (zcard            [key])
  (zcount           [key min max])
  (zincrby          [key increment member])
  (zinterstore      [destination numkeys set1 set2])
  (zrange           [key start stop])
  (zrank            [member])
  (zrem             [key member & members])
  (zremrangebyrank  [key start stop])
  (zremrangebyscore [key min max])
  (zrevrank         [key member])
  (zscore           [key member])
  (zunionstore      [destination numkeys set1 set2])

  (publish          [channel message]))