unbreak backpressure in stream -> input-stream conversion, mark 0.2.2-alpha1

Author: ztellman

project.clj

@@ -1,10 +1,10 @@
-(defproject byte-streams "0.2.1"
+(defproject byte-streams "0.2.2-alpha1"
   :description "A simple way to handle the menagerie of Java byte represenations."
   :license {:name "MIT License"
             :url "http://opensource.org/licenses/MIT"}
   :dependencies [[primitive-math "0.1.5"]
                  [clj-tuple "0.2.2"]
-                 [manifold "0.1.2"]]
+                 [manifold "0.1.3"]]
   :profiles {:dev {:dependencies [[org.clojure/clojure "1.8.0"]
                                   [org.clojure/test.check "0.9.0"]
                                   [codox-md "0.2.0" :exclusions [org.clojure/clojure]]]}}

src/byte_streams.clj

@@ -340,11 +340,13 @@
       (d/chain (s/take! s ::none)
         (fn [^bytes msg]
           (if (identical? ::none msg)
-            (ps/close ps)
-            (
-             do
-              (ps/put-array ps msg 0 (alength msg))
-              (d/recur))))))
+            (do
+              (ps/close ps)
+              false)
+            (ps/put-array ps msg 0 (alength msg))))
+        (fn [result]
+          (when result
+            (d/recur)))))
     (ps/->input-stream ps)))
 
 (def-conversion ^{:cost 0} [(stream-of ByteBuffer) InputStream]
@@ -354,10 +356,13 @@
       (d/chain (s/take! s ::none)
         (fn [^ByteBuffer msg]
           (if (identical? ::none msg)
-            (ps/close ps)
             (do
-              (ps/put-buffer ps (.duplicate msg))
-              (d/recur))))))
+              (ps/close ps)
+              false)
+            (ps/put-buffer ps (.duplicate msg))))
+        (fn [result]
+          (when result
+            (d/recur)))))
     (ps/->input-stream ps)))
 
 ;; byte-array => byte-buffer