Implement updated pubsub API

Author: ianopolous

lib/cid.jar

@@ -34,7 +34,7 @@
 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
         <version.junit>4.13.2</version.junit>
         <version.hamcrest>2.2</version.hamcrest>
-        <version.multiaddr>v1.4.3</version.multiaddr>
+        <version.multiaddr>v1.4.6</version.multiaddr>
 	</properties>
 
 	<repositories>

lib/multiaddr.jar

@@ -1,6 +1,7 @@
 package io.ipfs.api;
 
 import io.ipfs.cid.*;
+import io.ipfs.multibase.*;
 import io.ipfs.multihash.Multihash;
 import io.ipfs.multiaddr.MultiAddress;
 
@@ -276,16 +277,24 @@ public Object peers(String topic) throws IOException {
          * @return
          * @throws IOException
          */
-        public Object pub(String topic, String data) throws Exception {
-            return retrieveAndParse("pubsub/pub?arg="+topic + "&arg=" + data);
+        public void pub(String topic, String data) {
+            String encodedTopic = Multibase.encode(Multibase.Base.Base64Url, topic.getBytes());
+            Multipart m = new Multipart(protocol +"://" + host + ":" + port + version+"pubsub/pub?arg=" + encodedTopic, "UTF-8");
+            try {
+                m.addFilePart("file", Paths.get(""), new NamedStreamable.ByteArrayWrapper(data.getBytes()));
+                String res = m.finish();
+            } catch (IOException e) {
+                throw new RuntimeException(e.getMessage(), e);
+            }
         }
 
         public Stream<Map<String, Object>> sub(String topic) throws Exception {
             return sub(topic, ForkJoinPool.commonPool());
         }
 
         public Stream<Map<String, Object>> sub(String topic, ForkJoinPool threadSupplier) throws Exception {
-            return retrieveAndParseStream("pubsub/sub?arg=" + topic, threadSupplier).map(obj -> (Map)obj);
+            String encodedTopic = Multibase.encode(Multibase.Base.Base64Url, topic.getBytes());
+            return retrieveAndParseStream("pubsub/sub?arg=" + encodedTopic, threadSupplier).map(obj -> (Map)obj);
         }
 
         /**
@@ -295,10 +304,9 @@ public Stream<Map<String, Object>> sub(String topic, ForkJoinPool threadSupplier
          * @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);
+            String encodedTopic = Multibase.encode(Multibase.Base.Base64Url, topic.getBytes());
+            retrieveAndParseStream("pubsub/sub?arg="+encodedTopic, res -> results.accept((Map)res), error);
         }
-
-
     }
 
     /* 'ipfs block' is a plumbing command used to manipulate raw ipfs blocks.
@@ -778,7 +786,15 @@ private InputStream retrieveStream(String path) throws IOException {
 
     private static InputStream getStream(URL target, int connectTimeoutMillis, int readTimeoutMillis) throws IOException {
         HttpURLConnection conn = configureConnection(target, "POST", connectTimeoutMillis, readTimeoutMillis);
-        return conn.getInputStream();
+        try {
+            return conn.getInputStream();
+        } catch (IOException e) {
+            e.printStackTrace();
+            InputStream errorStream = conn.getErrorStream();
+            String err = errorStream == null ? e.getMessage() : new String(readFully(errorStream));
+            List<String> trailer = conn.getHeaderFields().get("Trailer");
+            throw new RuntimeException("IOException contacting IPFS daemon.\n"+err+"\nTrailer: " + trailer, e);
+        }
     }
 
     private Map postMap(String path, byte[] body, Map<String, String> headers) throws IOException {

lib/multibase.jar

@@ -9,7 +9,6 @@
 import java.io.*;
 import java.nio.file.*;
 import java.util.*;
-import java.util.concurrent.*;
 import java.util.function.*;
 import java.util.stream.*;
 
@@ -408,7 +407,7 @@ public void publish() throws Exception {
     }
 
     @Test
-    public void pubsubSynchronous() throws Exception {
+    public void pubsubSynchronous() {
         String topic = "topic" + System.nanoTime();
         List<Map<String, Object>> res = Collections.synchronizedList(new ArrayList<>());
         new Thread(() -> {
@@ -433,8 +432,8 @@ public void pubsub() throws Exception {
         String topic = "topic" + System.nanoTime();
         Stream<Map<String, Object>> sub = ipfs.pubsub.sub(topic);
         String data = "Hello!";
-        Object pub = ipfs.pubsub.pub(topic, data);
-        Object pub2 = ipfs.pubsub.pub(topic, "G'day");
+        ipfs.pubsub.pub(topic, data);
+        ipfs.pubsub.pub(topic, "G'day");
         List<Map> results = sub.limit(2).collect(Collectors.toList());
         Assert.assertTrue( ! results.get(0).equals(Collections.emptyMap()));
     }