Merge pull request #439 from mpenet/feature/csp-timeout

add csp/timeout channel

Author: heyLu

pixie/channels.pxi

@@ -1,5 +1,7 @@
 (ns pixie.channels
   (:require [pixie.stacklets :as st]
+            [pixie.uv :as uv]
+            [pixie.ffi :as ffi]
             [pixie.buffers :as b]))
 
 (defprotocol ICancelable
@@ -143,6 +145,24 @@
                                     false
                                     0)))))
 
+(defn timeout
+  "Returns a channel that will close after given delay in ms"
+  [ms]
+  (let [ch (chan)
+        timer (uv/uv_timer_t)
+        cb (atom nil)]
+    (reset! cb (ffi/ffi-prep-callback uv/uv_timer_cb
+                                      (fn [handle]
+                                        (try
+                                          (-close! ch)
+                                          (uv/uv_timer_stop timer)
+                                          (-dispose! @cb)
+                                          (catch ex
+                                              (println ex))))))
+    (uv/uv_timer_init (uv/uv_default_loop) timer)
+    (uv/uv_timer_start timer @cb ms 0)
+    ch))
+
 (deftype AltHandler [atm f]
   ICancelable
   (-canceled? [this]

pixie/csp.pxi

@@ -4,6 +4,7 @@
             [pixie.channels :as chans]))
 
 (def chan chans/chan)
+(def timeout chans/timeout)
 
 (defn close!
   "Closes the channel, future writes will be rejected, future reads will

tests/pixie/tests/test-csp.pxi

@@ -31,3 +31,18 @@
     (assert= (alts! [c] :default 42) [:default 42])
     (>! c 1)
     (assert= (alts! [c] :default 42) [c 1])))
+
+(deftest test-timeout-channel
+  (let [ts [(timeout 300)
+            (timeout 200)
+            (timeout 100)]]
+    (-> (go
+          (loop [ts (set ts)
+                 res []]
+            (if (empty? ts)
+              res
+              (let [[p _] (alts! ts)]
+                (recur (set (remove #{p} ts))
+                       (conj res p))))))
+        <!
+        (assert= (reverse ts)))))