ASYNC-258 Datafy buffers and report capacity

Author: puredanger

src/main/clojure/clojure/core/async/impl/buffers.clj

@@ -14,6 +14,7 @@
 
 (set! *warn-on-reflection* true)
 
+
 (deftype FixedBuffer [^LinkedList buf ^long n]
   impl/Buffer
   (full? [_this]
@@ -26,12 +27,14 @@
   (close-buf! [_this])
   Counted
   (count [_this]
-    (.size buf)))
+    (.size buf))
+  impl/Capacity
+  (capacity [_this]
+    n))
 
 (defn fixed-buffer [^long n]
   (FixedBuffer. (LinkedList.) n))
 
-
 (deftype DroppingBuffer [^LinkedList buf ^long n]
   impl/UnblockingBuffer
   impl/Buffer
@@ -46,7 +49,10 @@
   (close-buf! [_this])
   Counted
   (count [_this]
-    (.size buf)))
+    (.size buf))
+  impl/Capacity
+  (capacity [_this]
+    n))
 
 (defn dropping-buffer [n]
   (DroppingBuffer. (LinkedList.) n))
@@ -66,7 +72,10 @@
   (close-buf! [_this])
   Counted
   (count [_this]
-    (.size buf)))
+    (.size buf))
+  impl/Capacity
+  (capacity [_this]
+    n))
 
 (defn sliding-buffer [n]
   (SlidingBuffer. (LinkedList.) n))
@@ -91,7 +100,27 @@
       (set! val nil)))
   Counted
   (count [_]
-    (if (undelivered? val) 0 1)))
+    (if (undelivered? val) 0 1))
+  impl/Capacity
+  (capacity [_this]
+    1))
 
 (defn promise-buffer []
-  (PromiseBuffer. NO-VAL))
+  (PromiseBuffer. NO-VAL))
+
+(defn datafy-buffer [buffer]
+  {:type (-> buffer class .getSimpleName symbol)
+   :count (count buffer)
+   :capacity (impl/capacity buffer)})
+
+(extend-protocol clojure.core.protocols/Datafiable
+  FixedBuffer
+  (datafy [b] (datafy-buffer b))
+  DroppingBuffer
+  (datafy [b] (datafy-buffer b))
+  SlidingBuffer
+  (datafy [b] (datafy-buffer b))
+  PromiseBuffer
+  (datafy [b] (datafy-buffer b))
+  Object
+  (datafy [b] nil))

src/main/clojure/clojure/core/async/impl/channels.clj

@@ -8,7 +8,8 @@
 
 (ns ^{:skip-wiki true}
   clojure.core.async.impl.channels
-  (:require [clojure.core.async.impl.protocols :as impl]
+  (:require [clojure.datafy :as datafy]
+            [clojure.core.async.impl.protocols :as impl]
             [clojure.core.async.impl.dispatch :as dispatch]
             [clojure.core.async.impl.mutex :as mutex])
   (:import [java.util LinkedList Queue]
@@ -308,10 +309,8 @@
   ManyToManyChannel
   (datafy [c]
     (let [b (.buf c)]
-      {:buffer-type (if b
-                      (-> b class .getSimpleName symbol)
-                      :none)
-       :buffer-count (count b)
-       :put-count (count (.puts c))
-       :take-count (count (.takes c))
-       :closed? (impl/closed? c)})))
+      (cond->
+        {:put-count (count (.puts c))
+         :take-count (count (.takes c))
+         :closed? (impl/closed? c)}
+        b (assoc :buffer (clojure.datafy/datafy b))))))

src/main/clojure/clojure/core/async/impl/protocols.clj

@@ -34,6 +34,9 @@
   (add!* [b itm] "if room, add item to the buffer, returns b, called under chan mutex")
   (close-buf! [b] "called on chan closed under chan mutex, return ignored"))
 
+(defprotocol Capacity
+  (capacity [b] "The max capacity of the buffer, nil if unknown"))
+
 (defn add!
   ([b] b)
   ([b itm]