feat: add support for Future and IDeref impl ReadPort

CHANGELOG.md

@@ -1,5 +1,8 @@
 This is a history of changes to k13labs/futurama
 
+# 0.3.8-SNAPSHOT
+* Add support for `Future` and `IDeref`
+
 # 0.3.7
 * Add collection helpers for: `async-reduce`, `async-some`, `async-every?`, `async-walk/prewalk/postwalk`
 

README.md

@@ -4,7 +4,7 @@
 
 Futurama is a Clojure library for more deeply integrating async abstractions in the Clojure and JVM ecosystem with Clojure [core.async](https://github.com/clojure/core.async).
 
-It adds support for [CompletableFuture](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html) and [IDeferred](https://github.com/clj-commons/manifold/blob/master/src/manifold/deferred.clj) to be used in the same fashion as Clojure [core.async](https://github.com/clojure/core.async) channels.
+It adds support for [CompletableFuture](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html) and [IDeferred](https://github.com/clj-commons/manifold/blob/master/src/manifold/deferred.clj) to be used in approximately the same fashion as Clojure [core.async](https://github.com/clojure/core.async) channels, and for `Future` and `IDeref` to be read in the same manner as well.
 
 # _Usage_
 

pom.xml

@@ -5,9 +5,9 @@
   <groupId>com.github.k13labs</groupId>
   <artifactId>futurama</artifactId>
   <name>futurama</name>
-  <version>0.3.7</version>
+  <version>0.3.8-SNAPSHOT</version>
   <scm>
-    <tag>0.3.7</tag>
+    <tag>0.3.8-SNAPSHOT</tag>
     <url>https://github.com/k13labs/futurama</url>
     <connection>scm:git:git://github.com/k13labs/futurama.git</connection>
     <developerConnection>scm:git:ssh://git@github.com/k13labs/futurama.git</developerConnection>

src/futurama/core.clj

@@ -1,12 +1,12 @@
 (ns futurama.core
-  (:require [clojure.core.async :refer [<! <!! put! take! close!] :as async]
+  (:require [clojure.core.async :refer [<! <!! put! take! close! thread] :as async]
             [clojure.core.async.impl.protocols :as impl]
             [clojure.core.async.impl.channels :refer [box]]
             [clojure.core.async.impl.ioc-macros :as ioc]
             [clojure.core.reducers :as r]
             [futurama.util :as u]
             [futurama.deferred])
-  (:import [clojure.lang Var]
+  (:import [clojure.lang Var IDeref IFn]
            [java.util.concurrent
             CompletableFuture
             CompletionException
@@ -29,7 +29,8 @@
   "unwraps an ExecutionException or CompletionException via ex-cause until the root exception is returned"
   ^Throwable [^Throwable ex]
   (if-let [ce (and (or (instance? ExecutionException ex)
-                       (instance? CompletionException ex))
+                       (instance? CompletionException ex)
+                       (instance? InterruptedException ex))
                    (ex-cause ex))]
     ce
     ex))
@@ -68,6 +69,126 @@
      (.exceptionally res-fut# cancel#)
      res-fut#))
 
+(extend-type Future
+  impl/ReadPort
+  (take! [fut handler]
+    (let [^Future fut fut
+          ^Lock handler handler
+          commit-handler (fn do-commit []
+                           (.lock handler)
+                           (let [take-cb (and (impl/active? handler) (impl/commit handler))]
+                             (.unlock handler)
+                             take-cb))]
+      (when-let [cb (commit-handler)]
+        (cond
+          (realized? fut)
+          (let [val (try
+                      (.get ^Future fut)
+                      (catch Throwable e
+                        (unwrap-exception e)))]
+            (if (u/satisfies? impl/ReadPort val)
+              (do
+                (take! val (fn do-read
+                             [val]
+                             (if (u/satisfies? impl/ReadPort val)
+                               (take! val do-read)
+                               (cb val))))
+                nil)
+              (box val)))
+
+          :else
+          (do
+            (thread
+              (let [[val ex]
+                    (try
+                      [(.get ^Future fut) nil]
+                      (catch Throwable e
+                        [nil e]))]
+                (cond
+                  (u/satisfies? impl/ReadPort val)
+                  (take! val (fn do-read
+                               [val]
+                               (if (u/satisfies? impl/ReadPort val)
+                                 (take! val do-read)
+                                 (cb val))))
+
+                  (some? val)
+                  (cb val)
+
+                  (some? ex)
+                  (cb ex)
+
+                  :else
+                  (cb nil))))
+            nil)))))
+
+  impl/Channel
+  (close! [fut]
+    (when-not (realized? fut)
+      (future-cancel ^Future fut)))
+  (closed? [fut]
+    (realized? ^Future fut)))
+
+(extend-type IDeref
+  impl/ReadPort
+  (take! [ref handler]
+    (let [^IDeref ref ref
+          ^Lock handler handler
+          commit-handler (fn do-commit []
+                           (.lock handler)
+                           (let [take-cb (and (impl/active? handler) (impl/commit handler))]
+                             (.unlock handler)
+                             take-cb))]
+      (when-let [cb (commit-handler)]
+        (cond
+          (realized? ref)
+          (let [val (try
+                      (deref ref)
+                      (catch Throwable e
+                        (unwrap-exception e)))]
+            (if (u/satisfies? impl/ReadPort val)
+              (do
+                (take! val (fn do-read
+                             [val]
+                             (if (u/satisfies? impl/ReadPort val)
+                               (take! val do-read)
+                               (cb val))))
+                nil)
+              (box val)))
+
+          :else
+          (do
+            (thread
+              (let [[val ex]
+                    (try
+                      [(deref ref) nil]
+                      (catch Throwable e
+                        [nil e]))]
+                (cond
+                  (u/satisfies? impl/ReadPort val)
+                  (take! val (fn do-read
+                               [val]
+                               (if (u/satisfies? impl/ReadPort val)
+                                 (take! val do-read)
+                                 (cb val))))
+
+                  (some? val)
+                  (cb val)
+
+                  (some? ex)
+                  (cb ex)
+
+                  :else
+                  (cb nil))))
+            nil)))))
+
+  impl/Channel
+  (close! [ref]
+    (if (instance? IFn ref)
+      (ref nil)))
+  (closed? [ref]
+    (realized? ref)))
+
 (extend-type CompletableFuture
   impl/ReadPort
   (take! [fut handler]

test/futurama/core_test.clj

@@ -217,7 +217,11 @@
                            (<! (timeout 50))
                            (let [c (CompletableFuture.)]
                              (>! c {:foo "bar"})
-                             c))))))))))
+                             (delay
+                               (future
+                                 (let [p (promise)]
+                                   (deliver p c)
+                                   p)))))))))))))
   (testing "nested non-blocking take - !<!"
     (<!!
       (async
@@ -228,7 +232,12 @@
                           (completable-future
                             (go
                               (<! (timeout 50))
-                              (CompletableFuture/completedFuture {:foo "bar"})))))))))))))
+                              (delay
+                               (future
+                                 (let [p (promise)]
+                                   (deliver p
+                                            (CompletableFuture/completedFuture {:foo "bar"}))
+                                   p)))))))))))))))
 
 (deftest error-handling
   (testing "throws async exception on blocking deref - @"