dag get return stream & fixes

2 files changed, 26 insertions, 7 deletions

diff --git a/ipfs-api/examples/dag.rs b/ipfs-api/examples/dag.rs
index 1846376..9dff2bb 100644
--- a/ipfs-api/examples/dag.rs
+++ b/ipfs-api/examples/dag.rs
@@ -6,6 +6,7 @@
 // copied, modified, or distributed except according to those terms.
 //
 
+use futures::TryStreamExt;
 use ipfs_api::IpfsClient;
 use std::io::Cursor;
 
@@ -19,7 +20,7 @@ async fn main() {
 
     let client = IpfsClient::default();
 
-    let dag_node = Cursor::new(r#"{ "hello" : "world" }"#);
+    let dag_node = Cursor::new(r#"{"hello":"world"}"#);
 
     let response = client
         .dag_put(dag_node)
@@ -28,7 +29,17 @@ async fn main() {
 
     let cid = response.cid.cid_string;
 
-    let response = client.dag_get(&cid).await.expect("error getting dag node");
-
-    println!("dag node => {}", response);
+    match client
+        .dag_get(&cid)
+        .map_ok(|chunk| chunk.to_vec())
+        .try_concat()
+        .await
+    {
+        Ok(bytes) => {
+            println!("{}", String::from_utf8_lossy(&bytes[..]));
+        }
+        Err(e) => {
+            eprintln!("error reading dag node: {}", e);
+        }
+    }
 }
diff --git a/ipfs-api/src/client/internal.rs b/ipfs-api/src/client/internal.rs
index 6785180..17059fd 100644
--- a/ipfs-api/src/client/internal.rs
+++ b/ipfs-api/src/client/internal.rs
@@ -845,21 +845,29 @@ impl IpfsClient {
     /// Returns information about a dag node in Ipfs.
     ///
     /// ```no_run
+    /// use futures::TryStreamExt;
     /// use ipfs_api::IpfsClient;
     ///
     /// let client = IpfsClient::default();
-    /// let res = client.dag_get("QmXdNSQx7nbdRvkjGCEQgVjVtVwsHvV8NmV2a8xzQVwuFA");
+    /// let hash = "QmXdNSQx7nbdRvkjGCEQgVjVtVwsHvV8NmV2a8xzQVwuFA";
+    /// let res = client
+    ///     .dag_get(hash)
+    ///     .map_ok(|chunk| chunk.to_vec())
+    ///     .try_concat();
     /// ```
     ///
     #[inline]
-    pub async fn dag_get(&self, path: &str) -> Result<String, Error> {
-        self.request_string(request::DagGet { path }, None).await
+    pub fn dag_get(&self, path: &str) -> impl Stream<Item = Result<Bytes, Error>> {
+        impl_stream_api_response! {
+            (self, request::DagGet { path }, None) => request_stream_bytes
+        }
     }
 
     /// Add a DAG node to Ipfs.
     ///
     /// ```no_run
     /// use ipfs_api::IpfsClient;
+    /// use std::io::Cursor;
     ///
     /// let client = IpfsClient::default();
     /// let data = Cursor::new(r#"{ "hello" : "world" }"#);