Merge branch 'master' of https://github.com/ipfs/java-ipfs-api

gjeanmart · gjeanmart · commit b02a58a79b8b · 2018-06-24T13:12:03.000+01:00
diff --git a/.travis.yml b/.travis.yml
@@ -2,8 +2,8 @@ language: java
 jdk:
   - oraclejdk8
 before_script:
-  - wget https://dist.ipfs.io/go-ipfs/v0.4.11/go-ipfs_v0.4.11_linux-amd64.tar.gz -O /tmp/go-ipfs_v0.4.11_linux-amd64.tar.gz
-  - tar -xvf /tmp/go-ipfs_v0.4.11_linux-amd64.tar.gz
+  - wget https://dist.ipfs.io/go-ipfs/v0.4.13/go-ipfs_v0.4.13_linux-amd64.tar.gz -O /tmp/go-ipfs_linux-amd64.tar.gz
+  - tar -xvf /tmp/go-ipfs_linux-amd64.tar.gz
   - export PATH=$PATH:$PWD/go-ipfs/
   - ipfs init
   - ipfs daemon --enable-pubsub-experiment &
diff --git a/lib/multiaddr.jar b/lib/multiaddr.jar
diff --git a/pom.xml b/pom.xml
@@ -59,7 +59,7 @@
 		<dependency>
 			<groupId>com.github.multiformats</groupId>
 			<artifactId>java-multiaddr</artifactId>
-			<version>v1.1.1</version>
+			<version>v1.3.0</version>
 			<scope>system</scope>
 			<systemPath>${project.basedir}/lib/multiaddr.jar</systemPath>
 		</dependency>
diff --git a/src/main/java/io/ipfs/api/IPFS.java b/src/main/java/io/ipfs/api/IPFS.java
@@ -114,24 +114,24 @@ public List<MerkleNode> add(List<NamedStreamable> files, boolean wrap, boolean h
     }
 
     public List<MerkleNode> ls(Multihash hash) throws IOException {
-        Map res = retrieveMap("ls/" + hash);
+        Map res = retrieveMap("ls?arg=" + hash);
         return ((List<Object>) res.get("Objects")).stream().map(x -> MerkleNode.fromJSON((Map) x)).collect(Collectors.toList());
     }
 
     public byte[] cat(Multihash hash) throws IOException {
-        return retrieve("cat/" + hash);
+        return retrieve("cat?arg=" + hash);
     }
 
     public byte[] cat(Multihash hash, String subPath) throws IOException {
         return retrieve("cat?arg=" + hash + URLEncoder.encode(subPath, "UTF-8"));
     }
 
     public byte[] get(Multihash hash) throws IOException {
-        return retrieve("get/" + hash);
+        return retrieve("get?arg=" + hash);
     }
 
     public InputStream catStream(Multihash hash) throws IOException {
-        return retrieveStream("cat/" + hash);
+        return retrieveStream("cat?arg=" + hash);
     }
 
     public List<Multihash> refs(Multihash hash, boolean recursive) throws IOException {
@@ -263,17 +263,16 @@ public Object peers(String topic) throws IOException {
          * @return
          * @throws IOException
          */
-        public Object pub(String topic, String data) throws IOException {
+        public Object pub(String topic, String data) throws Exception {
             return retrieveAndParse("pubsub/pub?arg="+topic + "&arg=" + data);
         }
 
-        public Supplier<Map<String, Object>> sub(String topic) throws IOException {
+        public Stream<Map<String, Object>> sub(String topic) throws Exception {
             return sub(topic, ForkJoinPool.commonPool());
         }
 
-        public Supplier<Map<String, Object>> sub(String topic, ForkJoinPool threadSupplier) throws IOException {
-            Supplier<Object> sup = retrieveAndParseStream("pubsub/sub?arg=" + topic, threadSupplier);
-            return () -> (Map) sup.get();
+        public Stream<Map<String, Object>> sub(String topic, ForkJoinPool threadSupplier) throws Exception {
+            return retrieveAndParseStream("pubsub/sub?arg=" + topic, threadSupplier).map(obj -> (Map)obj);
         }
 
         /**
@@ -282,8 +281,8 @@ public Supplier<Map<String, Object>> sub(String topic, ForkJoinPool threadSuppli
          * @param results
          * @throws IOException
          */
-        public void sub(String topic, Consumer<Map<String, Object>> results) throws IOException {
-            retrieveAndParseStream("pubsub/sub?arg="+topic, res -> results.accept((Map)res));
+        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);
         }
 
 
@@ -417,8 +416,8 @@ public Map findprovs(Multihash hash) throws IOException {
             return retrieveMap("dht/findprovs?arg=" + hash);
         }
 
-        public Map query(MultiAddress addr) throws IOException {
-            return retrieveMap("dht/query?arg=" + addr.toString());
+        public Map query(Multihash peerId) throws IOException {
+            return retrieveMap("dht/query?arg=" + peerId.toString());
         }
 
         public Map findpeer(Multihash id) throws IOException {
@@ -443,7 +442,15 @@ public Map ls(Multihash path) throws IOException {
     // Network commands
 
     public List<MultiAddress> bootstrap() throws IOException {
-        return ((List<String>)retrieveMap("bootstrap/").get("Peers")).stream().map(x -> new MultiAddress(x)).collect(Collectors.toList());
+        return ((List<String>)retrieveMap("bootstrap/").get("Peers"))
+                .stream()
+                .flatMap(x -> {
+                    try {
+                        return Stream.of(new MultiAddress(x));
+                    } catch (Exception e) {
+                        return Stream.empty();
+                    }
+                }).collect(Collectors.toList());
     }
 
     public class Bootstrap {
@@ -471,7 +478,14 @@ public List<MultiAddress> rm(MultiAddress addr, boolean all) throws IOException
     public class Swarm {
         public List<Peer> peers() throws IOException {
             Map m = retrieveMap("swarm/peers?stream-channels=true");
-            return ((List<Object>)m.get("Peers")).stream().map(Peer::fromJSON).collect(Collectors.toList());
+            return ((List<Object>)m.get("Peers")).stream()
+                    .flatMap(json -> {
+                        try {
+                            return Stream.of(Peer.fromJSON(json));
+                        } catch (Exception e) {
+                            return Stream.empty();
+                        }
+                    }).collect(Collectors.toList());
         }
 
         public Map addrs() throws IOException {
@@ -603,16 +617,26 @@ private Object retrieveAndParse(String path) throws IOException {
         return JSONParser.parse(new String(res));
     }
 
-    private Supplier<Object> retrieveAndParseStream(String path, ForkJoinPool executor) throws IOException {
-        BlockingQueue<byte[]> objects = new LinkedBlockingQueue<>();
-        executor.submit(() -> getObjectStream(path, objects::add));
-        return () -> {
+    private Stream<Object> retrieveAndParseStream(String path, ForkJoinPool executor) throws IOException {
+        BlockingQueue<CompletableFuture<byte[]>> results = new LinkedBlockingQueue<>();
+        InputStream in = retrieveStream(path);
+        executor.submit(() -> getObjectStream(in,
+                res -> {
+                    results.add(CompletableFuture.completedFuture(res));
+                },
+                err -> {
+                    CompletableFuture<byte[]> fut = new CompletableFuture<>();
+                    fut.completeExceptionally(err);
+                    results.add(fut);
+                })
+        );
+        return Stream.generate(() -> {
             try {
-                return JSONParser.parse(new String(objects.take()));
-            } catch (InterruptedException e) {
+                return JSONParser.parse(new String(results.take().get()));
+            } catch (Exception e) {
                 throw new RuntimeException(e);
             }
-        };
+        });
     }
 
     /**
@@ -621,8 +645,8 @@ private Supplier<Object> retrieveAndParseStream(String path, ForkJoinPool execut
      * @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 void retrieveAndParseStream(String path, Consumer<Object> results, Consumer<IOException> err) throws IOException {
+        getObjectStream(retrieveStream(path), d -> results.accept(JSONParser.parse(new String(d))), err);
     }
 
     private byte[] retrieve(String path) throws IOException {
@@ -652,11 +676,10 @@ private static byte[] get(URL target) throws IOException {
         }
     }
 
-    private void getObjectStream(String path, Consumer<byte[]> processor) {
+    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];
@@ -669,7 +692,7 @@ private void getObjectStream(String path, Consumer<byte[]> processor) {
                 }
             }
         } catch (IOException e) {
-            e.printStackTrace();
+            error.accept(e);
         }
     }
 
@@ -682,7 +705,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();
     }
 
diff --git a/src/main/java/io/ipfs/api/Multipart.java b/src/main/java/io/ipfs/api/Multipart.java
@@ -82,7 +82,7 @@ private static String encode(String in) {
     }
 
     public void addFilePart(String fieldName, Path parent, NamedStreamable uploadFile) {
-        Optional<String> fileName = uploadFile.getName().map(n -> encode(parent.resolve(n).toString()));
+        Optional<String> fileName = uploadFile.getName().map(n -> encode(parent.resolve(n).toString().replace('\\','/')));
         writer.append("--").append(boundary).append(LINE_FEED);
         if (!fileName.isPresent())
             writer.append("Content-Disposition: file; name=\"").append(fieldName).append("\";").append(LINE_FEED);
diff --git a/src/main/java/io/ipfs/api/Peer.java b/src/main/java/io/ipfs/api/Peer.java
@@ -30,4 +30,9 @@ public static Peer fromJSON(Object json) {
         long latency = val.apply("Latency").length() > 0 ? Long.parseLong(val.apply("Latency")) : -1;
         return new Peer(new MultiAddress(val.apply("Addr")), Cid.decode(val.apply("Peer")), latency, val.apply("Muxer"), val.apply("Streams"));
     }
+
+    @Override
+    public String toString() {
+        return id + "@" + address;
+    }
 }
diff --git a/src/test/java/io/ipfs/api/APITest.java b/src/test/java/io/ipfs/api/APITest.java
@@ -9,6 +9,7 @@
 import java.io.*;
 import java.nio.file.*;
 import java.util.*;
+import java.util.concurrent.*;
 import java.util.function.*;
 import java.util.stream.*;
 
@@ -381,12 +382,12 @@ public void bulkBlockTest() throws IOException {
     }
 
     @Test
-    public void pubsubSynchronous() throws IOException {
+    public void pubsubSynchronous() throws Exception {
         String topic = "topic" + System.nanoTime();
         List<Map<String, Object>> res = Collections.synchronizedList(new ArrayList<>());
         new Thread(() -> {
             try {
-                ipfs.pubsub.sub(topic, res::add);
+                ipfs.pubsub.sub(topic, res::add, t -> t.printStackTrace());
             } catch (IOException e) {
                 throw new RuntimeException(e);}
         }).start();
@@ -406,17 +407,14 @@ public void pubsubSynchronous() throws IOException {
     }
 
     @Test
-    public void pubsub() throws IOException {
-        Object ls = ipfs.pubsub.ls();
-        Object peers = ipfs.pubsub.peers();
+    public void pubsub() throws Exception {
         String topic = "topic" + System.nanoTime();
-        Supplier<Map<String, Object>> sub = ipfs.pubsub.sub(topic);
-        Map<String, Object> first = sub.get();
-        Assert.assertTrue(first.equals(Collections.emptyMap()));
+        Stream<Map<String, Object>> sub = ipfs.pubsub.sub(topic);
         String data = "Hello!";
         Object pub = ipfs.pubsub.pub(topic, data);
-        Map second = sub.get();
-        Assert.assertTrue( ! second.equals(Collections.emptyMap()));
+        Object pub2 = ipfs.pubsub.pub(topic, "G'day");
+        List<Map> results = sub.limit(2).collect(Collectors.toList());
+        Assert.assertTrue( ! results.get(0).equals(Collections.emptyMap()));
     }
 
     private static String toEscapedHex(byte[] in) throws IOException {
@@ -613,7 +611,7 @@ public void dhtTest() throws IOException {
         Map put = ipfs.dht.put("somekey", "somevalue");
         Map findprovs = ipfs.dht.findprovs(pointer);
         List<Peer> peers = ipfs.swarm.peers();
-        Map query = ipfs.dht.query(peers.get(0).address);
+        Map query = ipfs.dht.query(peers.get(0).id);
         Map find = ipfs.dht.findpeer(peers.get(0).id);
     }
 

Original file line number	Diff line number	Diff line change
`@@ -82,7 +82,7 @@ private static String encode(String in) {`
`82`	`82`	`}`
`83`	`83`
`84`	`84`	`public void addFilePart(String fieldName, Path parent, NamedStreamable uploadFile) {`
`85`		`- Optional<String> fileName = uploadFile.getName().map(n -> encode(parent.resolve(n).toString()));`
	`85`	`+ Optional<String> fileName = uploadFile.getName().map(n -> encode(parent.resolve(n).toString().replace('\\','/')));`
`86`	`86`	`writer.append("--").append(boundary).append(LINE_FEED);`
`87`	`87`	`if (!fileName.isPresent())`
`88`	`88`	`writer.append("Content-Disposition: file; name=\"").append(fieldName).append("\";").append(LINE_FEED);`
Original file line number	Diff line number	Diff line change
`@@ -30,4 +30,9 @@ public static Peer fromJSON(Object json) {`
`30`	`30`	`long latency = val.apply("Latency").length() > 0 ? Long.parseLong(val.apply("Latency")) : -1;`
`31`	`31`	`return new Peer(new MultiAddress(val.apply("Addr")), Cid.decode(val.apply("Peer")), latency, val.apply("Muxer"), val.apply("Streams"));`
`32`	`32`	`}`
	`33`	`+`
	`34`	`+ @Override`
	`35`	`+ public String toString() {`
	`36`	`+ return id + "@" + address;`
	`37`	`+ }`
`33`	`38`	`}`