Update pubsub api

Author: ianopolous

src/main/java/io/ipfs/api/IPFS.java

@@ -246,14 +246,12 @@ public Object pub(String topic, String data) throws Exception {
             return retrieveAndParse("pubsub/pub?arg="+topic + "&arg=" + data);
         }
 
-        public Supplier<CompletableFuture<Map<String, Object>>> sub(String topic) throws Exception {
+        public Stream<Map<String, Object>> sub(String topic) throws Exception {
             return sub(topic, ForkJoinPool.commonPool());
         }
 
-        public Supplier<CompletableFuture<Map<String, Object>>> sub(String topic, ForkJoinPool threadSupplier) throws Exception {
-            Supplier<CompletableFuture<Object>> sup = retrieveAndParseStream("pubsub/sub?arg=" + topic, threadSupplier);
-            return () -> sup.get()
-                    .thenApply(obj -> (Map)obj);
+        public Stream<Map<String, Object>> sub(String topic, ForkJoinPool threadSupplier) throws Exception {
+            return retrieveAndParseStream("pubsub/sub?arg=" + topic, threadSupplier).map(obj -> (Map)obj);
         }
 
         /**
@@ -262,7 +260,7 @@ public Supplier<CompletableFuture<Map<String, Object>>> sub(String topic, ForkJo
          * @param results
          * @throws IOException
          */
-        public void sub(String topic, Consumer<Map<String, Object>> results, Consumer<Throwable> error) throws IOException {
+        public void sub(String topic, Consumer<Map<String, Object>> results, Consumer<IOException> error) throws IOException {
             retrieveAndParseStream("pubsub/sub?arg="+topic, res -> results.accept((Map)res), error);
         }
 
@@ -598,27 +596,25 @@ private Object retrieveAndParse(String path) throws IOException {
         return JSONParser.parse(new String(res));
     }
 
-    private Supplier<CompletableFuture<Object>> retrieveAndParseStream(String path, ForkJoinPool executor) throws IOException {
-        BlockingQueue<CompletableFuture<byte[]>> futures = new LinkedBlockingQueue<>();
-        CompletableFuture<byte[]> first = new CompletableFuture<>();
-        futures.add(first);
+    private Stream<Object> retrieveAndParseStream(String path, ForkJoinPool executor) {
+        BlockingQueue<CompletableFuture<byte[]>> results = new LinkedBlockingQueue<>();
         executor.submit(() -> getObjectStream(path,
                 res -> {
-                    futures.peek().complete(res);
-                    futures.add(new CompletableFuture<>());
+                    results.add(CompletableFuture.completedFuture(res));
                 },
                 err -> {
-                    futures.peek().completeExceptionally(err);
-                    futures.add(new CompletableFuture<>());
+                    CompletableFuture<byte[]> fut = new CompletableFuture<>();
+                    fut.completeExceptionally(err);
+                    results.add(fut);
                 })
         );
-        return () -> {
+        return Stream.generate(() -> {
             try {
-                return futures.take().thenApply(arr -> JSONParser.parse(new String(arr)));
-            } catch (InterruptedException e) {
+                return JSONParser.parse(new String(results.take().get()));
+            } catch (Exception e) {
                 throw new RuntimeException(e);
             }
-        };
+        });
     }
 
     /**
@@ -627,7 +623,7 @@ private Supplier<CompletableFuture<Object>> retrieveAndParseStream(String path,
      * @param results
      * @throws IOException
      */
-    private void retrieveAndParseStream(String path, Consumer<Object> results, Consumer<Throwable> err) throws IOException {
+    private void retrieveAndParseStream(String path, Consumer<Object> results, Consumer<IOException> err) throws IOException {
         getObjectStream(path, d -> results.accept(JSONParser.parse(new String(d))), err);
     }
 
@@ -658,7 +654,7 @@ private static byte[] get(URL target) throws IOException {
         }
     }
 
-    private void getObjectStream(String path, Consumer<byte[]> processor, Consumer<Throwable> error) {
+    private void getObjectStream(String path, Consumer<byte[]> processor, Consumer<IOException> error) {
         byte LINE_FEED = (byte)10;
 
         try {

src/test/java/io/ipfs/api/APITest.java

@@ -408,15 +408,14 @@ public void pubsubSynchronous() throws Exception {
 
     @Test
     public void pubsub() throws Exception {
-        Object ls = ipfs.pubsub.ls();
-        Object peers = ipfs.pubsub.peers();
         String topic = "topic" + System.nanoTime();
-        Supplier<CompletableFuture<Map<String, Object>>> sub = ipfs.pubsub.sub(topic);
+        Stream<Map<String, Object>> sub = ipfs.pubsub.sub(topic);
         Thread.sleep(100); // There's a race condition in ipfs
         String data = "Hello!";
         Object pub = ipfs.pubsub.pub(topic, data);
-        Map result = sub.get().get();
-        Assert.assertTrue( ! result.equals(Collections.emptyMap()));
+        Object pub2 = ipfs.pubsub.pub(topic, "G'day");
+        List<Map> results = sub.limit(2).collect(Collectors.toList());
+        Assert.assertTrue( ! results.get(0).equals(Collections.emptyMap()));
     }
 
     private static String toEscapedHex(byte[] in) throws IOException {