Core.async Support

benchmarks/things_bench.clj

@@ -0,0 +1,85 @@
+(ns things-bench
+  (:require [liberator.async :refer (go?)]
+            [liberator.core :refer (resource request-method-in)]
+            [liberator.representation :refer (ring-response)]
+            [perforate.core :refer :all]
+            [ring.mock.request :refer (request header)])
+  (:import [java.security MessageDigest]))
+
+(defn things-resource
+  [base things]
+  (resource
+   base
+   ;; early lookup   
+   :service-available? (fn [ctx] {::r (get @things (get-in ctx [:request :uri]))})
+   :method-allowed? (request-method-in :get :put :delete)
+   ;; lookup media types of the requested resource
+   :available-media-types #(if-let [m (get-in % [::r :media-type])] [m])
+   ;; the resource exists if a value is stored in @things at the uri
+   ;; store the looked up value at key ::r in the context
+   :exists? #(get % ::r)
+   ;; ...it existed if the stored value is nil (and not some random
+   ;; Objeced we use as a setinel)
+   :existed? #(nil? (get @things (get-in % [:request :uri]) (Object.)))
+   ;; use the previously stored value at ::r
+   :handle-ok #(get-in % [::r :content])
+   ;; update the representation
+   :put! #(dosync
+           (alter things assoc-in
+                  [(get-in % [:request :uri])]
+                  {:content (get-in % [:request :body])
+                   :media-type (get-in % [:request :headers "content-type"]
+                                       "application/octet-stream")
+                   :last-modified (java.util.Date.)}))
+   ;; ...store a nil value to marke the resource as gone
+   :delete! #(dosync (alter things assoc (get-in % [:request :uri]) nil))
+   :last-modified #(get-in % [::r :last-modified])))
+
+(defgoal things-bench "'Things' Resource benchmarks")
+
+(defn run-things
+  [{:keys [read-response resource-fn]}]
+  (let [t-r (comp read-response resource-fn)]
+    (let [resp (t-r (request :get "/r1"))]
+      (-> resp :status (= 404) assert))
+    (let [resp (t-r (-> (request :put "/r1")
+                        (assoc :body "r1")
+                        (header "content-type" "text/plain")))]
+      (-> resp :status (= 201) assert))
+    (let [resp (t-r (-> (request :get "/r1")))]
+      (-> resp :status (= 200) assert)
+      (-> resp :body (= "r1") assert)
+      (-> resp (get-in [:headers "Content-Type"])
+          (= "text/plain;charset=UTF-8")
+          assert))
+    (let [resp (t-r (-> (request :delete "/r1")))]
+      (-> resp :status (= 204) assert)
+      (-> resp :body nil? assert))
+    (let [resp (t-r (request :get "/r1"))]
+      (-> resp :status (= 410) assert))))
+
+(defcase things-bench "Things Sync"
+  []
+  (run-things {:read-response identity
+               :resource-fn
+               (things-resource
+                {:authorized? 
+                 (request-method-in :get :put :delete)}
+                (ref nil))}))
+
+(defcase things-bench "Things Async"
+  []
+  (let [things (ref nil)]
+    (run-things {:read-response #(-> % :body clojure.core.async/<!!)
+                 :resource-fn
+                 (things-resource
+                  {:async? true
+                   :authorized? 
+                   (fn [ctx]
+                     (go?
+                      ;; this is just to ensure that _something_ happens
+                      ;; in the go block
+                      #(some #{(:request-method (:request %))}
+                             [:get :put :delete])))}
+                  things)})))
+

project.clj

@@ -1,6 +1,7 @@
 (defproject liberator "0.12.0-SNAPSHOT"
   :description "Liberator - A REST library for Clojure."
-  :dependencies [[org.clojure/clojure "1.4.0"]
+  :dependencies [[org.clojure/clojure "1.5.1"]
+                 [org.clojure/core.async "0.1.267.0-0d7780-alpha"]
                  [org.clojure/tools.trace "0.7.3"]
                  [org.clojure/tools.logging "0.2.3"]
                  [org.clojure/data.json "0.2.1"]
@@ -16,7 +17,8 @@
         :url "https://github.com/clojure-liberator/liberator"}
 
   :plugins [[lein-midje "3.1.3" :exclusions [leiningen-core]]
-            [lein-ring "0.8.10" :exclusions [org.clojure/clojure]]]
+            [lein-ring "0.8.10" :exclusions [org.clojure/clojure]]
+            [perforate "0.3.3"]]
 
   :profiles {:dev {:dependencies [[ring/ring-jetty-adapter "1.2.1" :exclusions [joda-time]]
                                   [ring-mock "0.1.2"]
@@ -26,15 +28,19 @@
                                   [compojure "1.0.2" :exclusions [org.clojure/tools.macro]]
                                   [org.clojure/clojurescript "0.0-1450"]]
                    :source-paths [ "src" "examples/clj"]}
-             :1.4 {:dependencies [[org.clojure/clojure "1.4.0"]]}
              :1.5 {:dependencies [[org.clojure/clojure "1.5.1"]]}
              :1.6 {:dependencies [[org.clojure/clojure "1.6.0-beta1"]]}}
 
   :source-paths ["src"]
   :test-paths ["test"]
 
+  :perforate
+  {:environments [{:name :core
+                   :profiles [:dev :1.5]
+                   :namespaces [things-bench]}]}
+
   :ring {:handler examples.server/handler
          :adapter {:port 8000}}
 
   :aliases {"examples" ["run" "-m" "examples.server"]
-            "test-all" ["with-profile" "+1.4:+1.5:+1.6" "test"]})
+            "test-all" ["with-profile" "+1.5:+1.6" "test"]})

src/liberator/async.clj

@@ -0,0 +1,100 @@
+;; Copyright (c) Philipp Meier ([email protected]). All rights reserved.
+;; The use and distribution terms for this software are covered by the Eclipse
+;; Public License 1.0 (http://opensource.org/licenses/eclipse-1.0.php) which
+;; can be found in the file epl-v10.html at the root of this distribution. By
+;; using this software in any fashion, you are agreeing to be bound by the
+;; terms of this license. You must not remove this notice, or any other, from
+;; this software.
+
+(ns liberator.async
+  (:require [clojure.core.async :as async :refer [go <!]]
+            [clojure.core.async.impl.protocols :as async-p]))
+
+(def channel?
+  "Returns true if argument is a core.async channel"
+  (partial satisfies? async-p/Channel))
+
+(defmacro go?
+  "A `go` block which will catch any thrown `Throwable` instances
+and yield them as the value of the block channel."
+  [& body]
+  `(async/go
+     (try
+       ~@body
+       (catch Throwable e# e#))))
+
+(defn rethrow-exceptions
+  [maybe-e]
+  (if (instance? Throwable maybe-e)
+    (throw maybe-e)
+    maybe-e))
+
+(defmacro <?
+  "Read a value from the channel found by evaluating `expr` and, if it is
+a `Throwable` instance, then throw it, otherwise return it. Must be used
+inside a `go` block."
+  [expr]
+  `(rethrow-exceptions (async/<! ~expr)))
+
+(defn async-response
+  "If `r` has a `:body` which is a core.async channel, then returns that
+channel, else returns `nil`."
+  [r]
+  (let [b (and r (:body r))]
+    (when (channel? b)
+      b)))
+
+(defn async-middleware
+  "Utility for constructing ring middleware out of a pair of `:request` and
+`:response` handling functions (such as those found in ring core). Returns
+a function which can be used to wrap handlers in the resulting middleware.
+Note that neither `:request` and `:response` can itself return a core.async
+channel; if you need to write async aware middleware then you don't need this
+function!"
+  [& {:keys [request response]
+      :or {request identity response nil}}]
+  (fn [handler]
+    (fn [req]
+      (let [resp (-> req request handler)]
+        (if (and response (async-response resp))
+          (update-in resp [:body] (partial async/map< response))
+          resp)))))
+
+(defmacro <val?
+  "Evaluate `expr` and, if it is a channel, read a value from it with `<?`
+and yield it as the result, otherwise return the value of `expr`.
+Must be used inside a `go` block (at least when the result of `expr` is a
+channel)."
+  [expr]
+  `(let [v# ~expr]
+     (if (channel? v#)
+       (<? v#)
+       v#)))
+
+(defmacro <let?
+  "Optimistic mixed blocking and non-blocking version of `let`. Used to write
+code which receives functions which may be either blocking or non blocking,
+and needs to itself become either blocking or non blocking. This allows
+liberator to optionally work with core.async without forcing it on clients,
+and without requiring two sets of parallel APIs etc. (at the price of a
+rather ugly macro).
+
+Establishes `bindings` as if by `let`. If any of the resulting bindings
+are a channel, then a `go` block is initiated, the results of any channel
+bindings are read inside the block and rebound, and `body` is executed
+in the scope of the resulting bindings.
+
+If none of the resulting bindings from the first step are channels
+then `body` is executed in the scope of these bindings (no go block)."
+  [bindings & body]
+  (let [bind-map (apply hash-map bindings)
+        bind-keys (-> bind-map keys vec)
+        bind-gs (vec (for [k bind-keys] (gensym)))]
+    `(let [do-body# (fn ~bind-keys ~@body)
+            ~@(mapcat (fn [gs k] [gs (get bind-map k)])
+                      bind-gs bind-keys)]
+       (if (or ~@(for [gs bind-gs] `(channel? ~gs)))
+         (go?
+          (<val?
+           (do-body# ~@(for [gs bind-gs] `(<val? ~gs)))))
+         (do-body# ~@bind-gs)))))