Fix pubsub, and switch to a Supplier rather than a stream

Author: ianopolous

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

@@ -234,14 +234,14 @@ public Object peers(String topic) throws IOException {
         }
 
         public Object pub(String topic, String data) throws IOException {
-            return retrieveAndParse("pubsub/peers?arg="+topic + "&data=" + data);
+            return retrieveAndParse("pubsub/peers?arg="+topic + "&arg=" + data);
         }
 
-        public Stream<Object> sub(String topic) throws IOException {
+        public Supplier<Object> sub(String topic) throws IOException {
             return sub(topic, ForkJoinPool.commonPool());
         }
 
-        public Stream<Object> sub(String topic, ForkJoinPool threadSupplier) throws IOException {
+        public Supplier<Object> sub(String topic, ForkJoinPool threadSupplier) throws IOException {
             return retrieveAndParseStream("pubsub/sub?arg="+topic, threadSupplier);
         }
     }
@@ -556,10 +556,16 @@ private Object retrieveAndParse(String path) throws IOException {
         return JSONParser.parse(new String(res));
     }
 
-    private Stream<Object> retrieveAndParseStream(String path, ForkJoinPool executor) throws IOException {
-        Queue<byte[]> objects = new LinkedBlockingQueue<>();
+    private Supplier<Object> retrieveAndParseStream(String path, ForkJoinPool executor) throws IOException {
+        BlockingQueue<byte[]> objects = new LinkedBlockingQueue<>();
         executor.submit(() -> getObjectStream(path, objects::add));
-        return Stream.generate(() -> JSONParser.parse(new String(objects.poll())));
+        return () -> {
+            try {
+                return JSONParser.parse(new String(objects.take()));
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        };
     }
 
     private byte[] retrieve(String path) throws IOException {

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

@@ -401,11 +401,12 @@ public void bulkBlockTest() {
     public void pubsub() throws IOException {
         Object ls = ipfs.pubsub.ls();
         Object peers = ipfs.pubsub.peers();
-        Stream<Object> sub = ipfs.pubsub.sub("orbit" + System.nanoTime());
-        String data = "Gday All!";
-        Object pub = ipfs.pubsub.pub("orbit", data);
-        Optional<Object> first = sub.findFirst();
-        Assert.assertTrue(first.isPresent() && first.get().equals(data));
+        String topic = "topic" + System.nanoTime();
+        Supplier<Object> sub = ipfs.pubsub.sub(topic);
+        Object first = sub.get();
+        String data = "Hello!";
+        Object pub = ipfs.pubsub.pub(topic, data);
+        Assert.assertTrue(first.equals(Collections.emptyMap()));
     }
 
     private static String toEscapedHex(byte[] in) throws IOException {