Implement pubsub, though pub currently doesn't work

Author: ianopolous

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

@@ -8,6 +8,8 @@
 import java.net.*;
 import java.nio.file.*;
 import java.util.*;
+import java.util.concurrent.*;
+import java.util.function.*;
 import java.util.stream.*;
 
 public class IPFS {
@@ -36,6 +38,7 @@ public enum PinType {all, direct, indirect, recursive}
     public final File file = new File();
     public final Stats stats = new Stats();
     public final Name name = new Name();
+    public final Pubsub pubsub = new Pubsub();
 
     public IPFS(String host, int port) {
         this(host, port, "/api/v0/");
@@ -217,6 +220,32 @@ public Object gc() throws IOException {
         }
     }
 
+    public class Pubsub {
+        public Object ls() throws IOException {
+            return retrieveAndParse("pubsub/ls");
+        }
+
+        public Object peers() throws IOException {
+            return retrieveAndParse("pubsub/peers");
+        }
+
+        public Object peers(String topic) throws IOException {
+            return retrieveAndParse("pubsub/peers?topic="+topic);
+        }
+
+        public Object pub(String topic, String data) throws IOException {
+            return retrieveAndParse("pubsub/peers?arg="+topic + "&data=" + data);
+        }
+
+        public Stream<Object> sub(String topic) throws IOException {
+            return sub(topic, ForkJoinPool.commonPool());
+        }
+
+        public Stream<Object> sub(String topic, ForkJoinPool threadSupplier) throws IOException {
+            return retrieveAndParseStream("pubsub/sub?arg="+topic, threadSupplier);
+        }
+    }
+
     /* 'ipfs block' is a plumbing command used to manipulate raw ipfs blocks.
      */
     public class Block {
@@ -527,6 +556,12 @@ 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<>();
+        executor.submit(() -> getObjectStream(path, objects::add));
+        return Stream.generate(() -> JSONParser.parse(new String(objects.poll())));
+    }
+
     private byte[] retrieve(String path) throws IOException {
         URL target = new URL("http", host, port, version + path);
         return IPFS.get(target);
@@ -554,6 +589,27 @@ private static byte[] get(URL target) throws IOException {
         }
     }
 
+    private void getObjectStream(String path, Consumer<byte[]> processor) {
+        byte LINE_FEED = (byte)10;
+
+        try {
+            InputStream in = retrieveStream(path);
+            ByteArrayOutputStream resp = new ByteArrayOutputStream();
+
+            byte[] buf = new byte[4096];
+            int r;
+            while ((r = in.read(buf)) >= 0) {
+                resp.write(buf, 0, r);
+                if (buf[r - 1] == LINE_FEED) {
+                    processor.accept(resp.toByteArray());
+                    resp.reset();
+                }
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
     private InputStream retrieveStream(String path) throws IOException {
         URL target = new URL("http", host, port, version + path);
         return IPFS.getStream(target);

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

@@ -7,6 +7,7 @@
 import org.junit.*;
 
 import java.io.*;
+import java.net.*;
 import java.nio.file.*;
 import java.util.*;
 import java.util.function.*;
@@ -396,6 +397,17 @@ public void bulkBlockTest() {
         }
     }
 
+    @Test
+    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));
+    }
+
     private static String toEscapedHex(byte[] in) throws IOException {
         StringBuilder res = new StringBuilder();
         for (byte b : in) {