Add a synchronous version of subscribe which consumes the calling thread. This allows much lower latency (1ms)

Author: ianopolous

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

@@ -253,6 +253,18 @@ public Supplier<Object> sub(String topic) throws IOException {
         public Supplier<Object> sub(String topic, ForkJoinPool threadSupplier) throws IOException {
             return retrieveAndParseStream("pubsub/sub?arg="+topic, threadSupplier);
         }
+
+        /**
+         * A synchronous method to subscribe which consumes the calling thread
+         * @param topic
+         * @param results
+         * @throws IOException
+         */
+        public void sub(String topic, Consumer<Object> results) throws IOException {
+            retrieveAndParseStream("pubsub/sub?arg="+topic, results);
+        }
+
+
     }
 
     /* 'ipfs block' is a plumbing command used to manipulate raw ipfs blocks.
@@ -581,6 +593,16 @@ private Supplier<Object> retrieveAndParseStream(String path, ForkJoinPool execut
         };
     }
 
+    /**
+     * A synchronous stream retriever that consumes the calling thread
+     * @param path
+     * @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 byte[] retrieve(String path) throws IOException {
         URL target = new URL("http", host, port, version + path);
         return IPFS.get(target);

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

@@ -380,6 +380,31 @@ public void bulkBlockTest() throws IOException {
         System.out.println();
     }
 
+    @Test
+    public void pubsubSynchronous() throws IOException {
+        String topic = "topic" + System.nanoTime();
+        List<Object> res = Collections.synchronizedList(new ArrayList<>());
+        new Thread(() -> {
+            try {
+                ipfs.pubsub.sub(topic, res::add);
+            } catch (IOException e) {
+                throw new RuntimeException(e);}
+        }).start();
+
+        long start = System.currentTimeMillis();
+        for (int i=1; i < 100; ) {
+            long t1 = System.currentTimeMillis();
+            ipfs.pubsub.pub(topic, "Hello!");
+            if (res.size() >= i) {
+                long t2 = System.currentTimeMillis();
+                System.out.println("pub => sub took " + (t2 - t1));
+                i++;
+            }
+        }
+        long duration = System.currentTimeMillis() - start;
+        Assert.assertTrue("Fast synchronous pub-sub", duration < 1000);
+    }
+
     @Test
     public void pubsub() throws IOException {
         Object ls = ipfs.pubsub.ls();