protocol.clj

(ns ckampfe.protocol
  (:require [clojure.java.io :as io]))

(defrecord BulkString [s])
(defrecord SimpleError [s])

(defmacro as-byte [s]
  `(byte (first ~s)))

(declare decode)

(defmulti decode-payload (fn [reader] (.read reader)))

;; TODO handle disconnection, which returns -1 on the socket
;; (defmethod decode-payload -1 [reader]
;;   :disconnect)

;; array
(defmethod decode-payload (as-byte "*") [reader]
  ;; (println "decoding array")
  (let [number-of-elements (Integer/parseInt (.readLine reader))]
    (loop [n number-of-elements
           elements []]
      (if (> n 0)
        (recur (- n 1)
               (let [el (decode reader)]
                 ;; (println "array el:" el)
                 (conj elements el)))
        elements))))

;; simple string
(defmethod decode-payload (as-byte "+") [reader]
  (.readLine reader))

;; simple error
(defmethod decode-payload (as-byte "-") [reader]
  (.readLine reader))

;; integer
(defmethod decode-payload (as-byte ":") [reader]
  (let [i (.readLine reader)]
    (Integer/parseInt i)))

;; bulk string
(defmethod decode-payload (as-byte "$") [reader]
  (let [_length (.readLine reader)
        body (.readLine reader)]
    body))

;; boolean
(defmethod decode-payload (as-byte "#") [reader]
  (let [v (.readLine reader)]
    (case v
      "t" true
      "f" false)))

;; null
(defmethod decode-payload (as-byte "_") [reader]
  ;; (println "decode null")
  (.readLine reader)
  nil)

(defn decode [^java.io.Reader reader]
  (decode-payload reader))

(declare encode-expr)

(defn encode [expr ^java.io.Writer writer]
  (encode-expr expr writer))

(defmulti encode-expr (fn [expr _writer] (type expr)))

(defmethod encode-expr java.lang.String [expr ^java.io.Writer writer]
  (.write writer "+")
  (.write writer expr)
  (.write writer "\r\n")
  writer)

(defmethod encode-expr SimpleError [expr ^java.io.Writer writer]
  (.write writer "-")
  (.write writer (:s expr))
  (.write writer "\r\n")
  writer)

(defmethod encode-expr java.lang.Integer [expr writer]
  (.write writer ":")
  (.write writer (.toString expr))
  (.write writer "\r\n")
  writer)

(defmethod encode-expr java.lang.Long [expr writer]
  (.write writer ":")
  (.write writer (.toString expr))
  (.write writer "\r\n")
  writer)

(defmethod encode-expr BulkString [expr writer]
  (.write writer "$")
  (.write writer (.toString (.length (:s expr))))
  (.write writer "\r\n")
  (.write writer (:s expr))
  (.write writer "\r\n")
  writer)

(defmethod encode-expr nil [_expr writer]
  (.write writer "_")
  (.write writer "\r\n")
  writer)

(defmethod encode-expr java.lang.Boolean [expr writer]
  (if expr
    (.write writer "#t\r\n")
    (.write writer "#f\r\n"))
  writer)

(defmethod encode-expr clojure.lang.PersistentArrayMap [expr writer]
  (.write writer "%")
  (.write writer (.toString (count expr)))
  (.write writer "\r\n")
  (doseq [[k v] expr]
    (encode-expr k writer)
    (encode-expr v writer))
  writer)

(defmethod encode-expr clojure.lang.PersistentHashMap [expr writer]
  (.write writer "%")
  (.write writer (.toString (count expr)))
  (.write writer "\r\n")
  (doseq [[k v] expr]
    (encode-expr k writer)
    (encode-expr v writer))
  writer)

(defmethod encode-expr clojure.lang.PersistentVector [expr writer]
  (.write writer "*")
  (.write writer (.toString (count expr)))
  (.write writer "\r\n")
  (doseq [el expr]
    (encode-expr el writer))
  writer)

(comment
  (ns-unmap *ns* 'encode-expr)

  (.toString (encode-expr "foo" (java.io.StringWriter.)))

  (.toString (encode-expr 7 (java.io.StringWriter.)))

  (.toString (encode-expr -7 (java.io.StringWriter.)))

  (.toString (encode-expr (BulkString. "hello") (java.io.StringWriter.)))

  (.toString (encode-expr (BulkString. "") (java.io.StringWriter.)))

  (.toString (encode-expr nil (java.io.StringWriter.)))

  (.toString (encode-expr true (java.io.StringWriter.)))

  (.toString (encode-expr false (java.io.StringWriter.)))

  (=
   (.toString (encode-expr [[1 2 3] ["Hello" (SimpleError. "World")]] (java.io.StringWriter.)))
   "*2\r\n*3\r\n:1\r\n:2\r\n:3\r\n*2\r\n+Hello\r\n-World\r\n")

  (.toString (encode-expr [[1 2 3] ["Hello" (SimpleError. "World")]] (java.io.StringWriter.)))

  (ns-unmap *ns* 'decode-payload)

  ;; strings
  (decode-payload (io/reader (java.io.StringReader. "+OK\r\n")))

  ;; errors
  (decode-payload (io/reader (java.io.StringReader. "-Error message\r\n")))

  ;; integers
  (decode-payload (io/reader (java.io.StringReader. ":42\r\n")))

  (decode-payload (io/reader (java.io.StringReader. ":-42\r\n")))

  (decode-payload (io/reader (java.io.StringReader. ":+42\r\n")))

  ;; bulk strings
  (decode-payload (io/reader (java.io.StringReader. "$5\r\nhello\r\n")))

  (decode-payload (io/reader (java.io.StringReader. "$0\r\n\r\n")))

  ;; null bulk string for null
  (decode-payload (io/reader (java.io.StringReader. "$-1\r\n")))

  ;; array
  (decode-payload (io/reader (java.io.StringReader. "*0\r\n")))

  (decode-payload (io/reader (java.io.StringReader. "*2\r\n$5\r\nhello\r\n$5\r\nworld\r\n")))

  (decode-payload (io/reader (java.io.StringReader. "*3\r\n:1\r\n:2\r\n:3\r\n")))

  (decode-payload (io/reader (java.io.StringReader. "*5\r\n:1\r\n:2\r\n:3\r\n:4\r\n$5\r\nhello\r\n")))

  (decode-payload (io/reader (java.io.StringReader. "*2\r\n*3\r\n:1\r\n:2\r\n:3\r\n*2\r\n+Hello\r\n-World\r\n")))

  ;; null array
  (decode-payload (io/reader (java.io.StringReader. "*-1\r\n")))

  ;; null
  (decode-payload (io/reader (java.io.StringReader. "_\r\n")))

  ;; booleans
  (decode-payload (io/reader (java.io.StringReader. "#t\r\n")))

  (decode-payload (io/reader (java.io.StringReader. "#f\r\n"))))

redjus.clj

(ns ckampfe.redjus
  (:require [ckampfe.protocol :as proto]
            [clojure.java.io :as io]
            [clojure.core.match :refer [match]])
  (:gen-class))

(defmacro bs [s]
  `(proto/->BulkString ~s))

(defn interpret
  "main action loop"
  [state command]
  (match command
    ["PING"] ["PONG" state]
    ["GET" k] [(get-in state [:keys k]) state]
    ["SET" k v] ["OK" (assoc-in state [:keys k] v)]
    ["COMMAND" "DOCS"]
    [[(bs "ping") [(bs "summary") (bs "Returns the server's liveliness response.")
                   (bs "since") (bs "1.0.0")
                   (bs "group") (bs "connection")
                   (bs "complexity") (bs "O(1)")
                   (bs "arguments") [[(bs "name") (bs "message")
                                      (bs "type") (bs "string")
                                      (bs "display_text") (bs "message")
                                      (bs "flags") ["optional"]]]]]
     state]))

(defn start-server
  "start the server, listening on `port` and with a default `state`"
  [port state]
  (let [running (atom true)]
    (future
      (try
        (with-open [listener (java.net.ServerSocket. port)
                    socket (.accept listener)]
          (println "got conn")
          (loop [state state]
            (let [decoded (proto/decode (io/reader socket))
                  _ (println "after decode")
                  [res state] (try (interpret state decoded)
                                   (catch Exception e
                                     (println (.toString e))
                                     (throw e)))
                  _ (println "after interpret")
                  _ (println res)
                  writer (proto/encode res (io/writer socket))]
              (.flush writer)
              (println "DECODED" decoded)
              (println "RESPONSE" res)
              (println "STATE" state)

              (when @running
                (recur state))))
          (println "shutting down..."))

        (catch Exception e (println e))))

    running))

(defn -main
  "I don't do a whole lot ... yet."
  [& args]
  (start-server 6379 {}))

(comment
  (def server (start-server 6379 {}))

  (reset! server false)

  )