Fix potential race condition in sub

Author: ianopolous

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

@@ -596,9 +596,10 @@ private Object retrieveAndParse(String path) throws IOException {
         return JSONParser.parse(new String(res));
     }
 
-    private Stream<Object> retrieveAndParseStream(String path, ForkJoinPool executor) {
+    private Stream<Object> retrieveAndParseStream(String path, ForkJoinPool executor) throws IOException {
         BlockingQueue<CompletableFuture<byte[]>> results = new LinkedBlockingQueue<>();
-        executor.submit(() -> getObjectStream(path,
+        InputStream in = retrieveStream(path);
+        executor.submit(() -> getObjectStream(in,
                 res -> {
                     results.add(CompletableFuture.completedFuture(res));
                 },
@@ -624,7 +625,7 @@ private Stream<Object> retrieveAndParseStream(String path, ForkJoinPool executor
      * @throws IOException
      */
     private void retrieveAndParseStream(String path, Consumer<Object> results, Consumer<IOException> err) throws IOException {
-        getObjectStream(path, d -> results.accept(JSONParser.parse(new String(d))), err);
+        getObjectStream(retrieveStream(path), d -> results.accept(JSONParser.parse(new String(d))), err);
     }
 
     private byte[] retrieve(String path) throws IOException {
@@ -654,11 +655,10 @@ private static byte[] get(URL target) throws IOException {
         }
     }
 
-    private void getObjectStream(String path, Consumer<byte[]> processor, Consumer<IOException> error) {
+    private void getObjectStream(InputStream in, Consumer<byte[]> processor, Consumer<IOException> error) {
         byte LINE_FEED = (byte)10;
 
         try {
-            InputStream in = retrieveStream(path);
             ByteArrayOutputStream resp = new ByteArrayOutputStream();
 
             byte[] buf = new byte[4096];
@@ -684,7 +684,6 @@ private static InputStream getStream(URL target) throws IOException {
         HttpURLConnection conn = (HttpURLConnection) target.openConnection();
         conn.setRequestMethod("GET");
         conn.setRequestProperty("Content-Type", "application/json");
-
         return conn.getInputStream();
     }
 

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

@@ -410,7 +410,6 @@ public void pubsubSynchronous() throws Exception {
     public void pubsub() throws Exception {
         String topic = "topic" + System.nanoTime();
         Stream<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);
         Object pub2 = ipfs.pubsub.pub(topic, "G'day");