Update pubsub api to enable handling errors e.g. ipfs daemon terminating

Author: ianopolous

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

@@ -242,17 +242,18 @@ public Object peers(String topic) throws IOException {
          * @return
          * @throws IOException
          */
-        public Object pub(String topic, String data) throws IOException {
+        public Object pub(String topic, String data) throws Exception {
             return retrieveAndParse("pubsub/pub?arg="+topic + "&arg=" + data);
         }
 
-        public Supplier<Map<String, Object>> sub(String topic) throws IOException {
+        public Supplier<CompletableFuture<Map<String, Object>>> sub(String topic) throws Exception {
             return sub(topic, ForkJoinPool.commonPool());
         }
 
-        public Supplier<Map<String, Object>> sub(String topic, ForkJoinPool threadSupplier) throws IOException {
-            Supplier<Object> sup = retrieveAndParseStream("pubsub/sub?arg=" + topic, threadSupplier);
-            return () -> (Map) sup.get();
+        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);
         }
 
         /**
@@ -261,8 +262,8 @@ public Supplier<Map<String, Object>> sub(String topic, ForkJoinPool threadSuppli
          * @param results
          * @throws IOException
          */
-        public void sub(String topic, Consumer<Map<String, Object>> results) throws IOException {
-            retrieveAndParseStream("pubsub/sub?arg="+topic, res -> results.accept((Map)res));
+        public void sub(String topic, Consumer<Map<String, Object>> results, Consumer<Throwable> error) throws IOException {
+            retrieveAndParseStream("pubsub/sub?arg="+topic, res -> results.accept((Map)res), error);
         }
 
 
@@ -582,12 +583,23 @@ private Object retrieveAndParse(String path) throws IOException {
         return JSONParser.parse(new String(res));
     }
 
-    private Supplier<Object> retrieveAndParseStream(String path, ForkJoinPool executor) throws IOException {
-        BlockingQueue<byte[]> objects = new LinkedBlockingQueue<>();
-        executor.submit(() -> getObjectStream(path, objects::add));
+    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);
+        executor.submit(() -> getObjectStream(path,
+                res -> {
+                    futures.peek().complete(res);
+                    futures.add(new CompletableFuture<>());
+                },
+                err -> {
+                    futures.peek().completeExceptionally(err);
+                    futures.add(new CompletableFuture<>());
+                })
+        );
         return () -> {
             try {
-                return JSONParser.parse(new String(objects.take()));
+                return futures.take().thenApply(arr -> JSONParser.parse(new String(arr)));
             } catch (InterruptedException e) {
                 throw new RuntimeException(e);
             }
@@ -600,8 +612,8 @@ private Supplier<Object> retrieveAndParseStream(String path, ForkJoinPool execut
      * @param results
      * @throws IOException
      */
-    private void retrieveAndParseStream(String path, Consumer<Object> results) throws IOException {
-        getObjectStream(path, d -> results.accept(JSONParser.parse(new String(d))));
+    private void retrieveAndParseStream(String path, Consumer<Object> results, Consumer<Throwable> err) throws IOException {
+        getObjectStream(path, d -> results.accept(JSONParser.parse(new String(d))), err);
     }
 
     private byte[] retrieve(String path) throws IOException {
@@ -631,7 +643,7 @@ private static byte[] get(URL target) throws IOException {
         }
     }
 
-    private void getObjectStream(String path, Consumer<byte[]> processor) {
+    private void getObjectStream(String path, Consumer<byte[]> processor, Consumer<Throwable> error) {
         byte LINE_FEED = (byte)10;
 
         try {
@@ -648,7 +660,7 @@ private void getObjectStream(String path, Consumer<byte[]> processor) {
                 }
             }
         } catch (IOException e) {
-            e.printStackTrace();
+            error.accept(e);
         }
     }
 

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

@@ -9,6 +9,7 @@
 import java.io.*;
 import java.nio.file.*;
 import java.util.*;
+import java.util.concurrent.*;
 import java.util.function.*;
 import java.util.stream.*;
 
@@ -381,12 +382,12 @@ public void bulkBlockTest() throws IOException {
     }
 
     @Test
-    public void pubsubSynchronous() throws IOException {
+    public void pubsubSynchronous() throws Exception {
         String topic = "topic" + System.nanoTime();
         List<Map<String, Object>> res = Collections.synchronizedList(new ArrayList<>());
         new Thread(() -> {
             try {
-                ipfs.pubsub.sub(topic, res::add);
+                ipfs.pubsub.sub(topic, res::add, t -> t.printStackTrace());
             } catch (IOException e) {
                 throw new RuntimeException(e);}
         }).start();
@@ -406,17 +407,16 @@ public void pubsubSynchronous() throws IOException {
     }
 
     @Test
-    public void pubsub() throws IOException {
+    public void pubsub() throws Exception {
         Object ls = ipfs.pubsub.ls();
         Object peers = ipfs.pubsub.peers();
         String topic = "topic" + System.nanoTime();
-        Supplier<Map<String, Object>> sub = ipfs.pubsub.sub(topic);
-        Map<String, Object> first = sub.get();
-        Assert.assertTrue(first.equals(Collections.emptyMap()));
+        Supplier<CompletableFuture<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 second = sub.get();
-        Assert.assertTrue( ! second.equals(Collections.emptyMap()));
+        Map result = sub.get().get();
+        Assert.assertTrue( ! result.equals(Collections.emptyMap()));
     }
 
     private static String toEscapedHex(byte[] in) throws IOException {