Refs local endpoint

http/src/v0.rs

@@ -6,6 +6,7 @@ pub mod block;
 pub mod dag;
 pub mod id;
 pub mod pubsub;
+pub mod refs;
 pub mod swarm;
 pub mod version;
 
@@ -53,7 +54,8 @@ where
             .or(warp::path!("pin" / ..).and_then(not_implemented))
             .or(warp::path!("ping" / ..).and_then(not_implemented))
             .or(pubsub::routes(ipfs))
-            .or(warp::path!("refs" / ..).and_then(not_implemented))
+            .or(refs::local(ipfs))
+            // .or(warp::path!("refs").and_then(refs::of_path))
             .or(warp::path!("repo" / ..).and_then(not_implemented))
             .or(warp::path!("stats" / ..).and_then(not_implemented))
             .or(swarm::connect(ipfs))

http/src/v0/refs.rs

@@ -0,0 +1,92 @@
+use futures::stream;
+use ipfs::error::Error;
+use ipfs::{Ipfs, IpfsTypes};
+use serde::Serialize;
+use warp::hyper::Body;
+use warp::{path, Filter, Rejection, Reply};
+
+use super::support::{with_ipfs, StringError};
+
+#[derive(Serialize, Debug)]
+struct RefsResponseItem {
+    #[serde(rename = "Err")]
+    err: String,
+
+    #[serde(rename = "Ref")]
+    refs: String,
+}
+
+/// Handling of https://docs-beta.ipfs.io/reference/http/api/#api-v0-refs-local
+pub fn local<T: IpfsTypes>(
+    ipfs: &Ipfs<T>,
+) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
+    path!("refs" / "local")
+        .and(with_ipfs(ipfs))
+        .and_then(inner_local)
+}
+
+async fn inner_local<T: IpfsTypes>(ipfs: Ipfs<T>) -> Result<impl Reply, Rejection> {
+    let refs: Vec<Result<String, Error>> = ipfs
+        .refs_local()
+        .await
+        .map_err(StringError::from)?
+        .into_iter()
+        .map(|cid| cid.to_string())
+        .map(|refs| RefsResponseItem {
+            refs,
+            err: "".to_string(),
+        })
+        .map(|response| {
+            serde_json::to_string(&response).map_err(|e| {
+                eprintln!("error from serde_json: {}", e);
+                HandledErr
+            }).unwrap()
+        })
+        .map(|ref_json| Ok(format!("{}{}", ref_json, "\n")))
+        .collect();
+
+    let stream = stream::iter(refs);
+    Ok(warp::reply::Response::new(Body::wrap_stream(stream)))
+}
+
+#[derive(Debug)]
+struct HandledErr;
+
+impl std::error::Error for HandledErr {}
+
+use std::fmt;
+
+impl fmt::Display for HandledErr {
+    fn fmt(&self, _fmt: &mut fmt::Formatter) -> fmt::Result {
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::inner_local;
+    use ipfs::Block;
+    use libipld::cid::Cid;
+    use libipld::cid::Codec;
+    use multihash::Sha2_256;
+
+    #[tokio::test]
+    async fn test_inner_local() {
+        use ipfs::{IpfsOptions, UninitializedIpfs};
+
+        let options = IpfsOptions::inmemory_with_generated_keys(false);
+
+        let (mut ipfs, fut) = UninitializedIpfs::new(options).await.start().await.unwrap();
+        drop(fut);
+
+        for data in &[b"1", b"2", b"3"] {
+            let data_slice = data.to_vec().into_boxed_slice();
+            let cid = Cid::new_v1(Codec::Raw, Sha2_256::digest(&data_slice));
+            let block = Block::new(data_slice, cid);
+            ipfs.put_block(block.clone()).await.unwrap();
+        }
+
+        let _result = inner_local(ipfs).await;
+        // println!("{:?}", result.unwrap());
+    }
+}

src/lib.rs

@@ -522,6 +522,10 @@ impl<Types: IpfsTypes> Ipfs<Types> {
         Ok(rx.await?)
     }
 
+    pub async fn refs_local(&self) -> Result<Vec<Cid>, Error> {
+        Ok(self.repo.list_blocks().await?)
+    }
+
     pub async fn bitswap_stats(&self) -> Result<BitswapStats, Error> {
         let (tx, rx) = oneshot_channel();
 

src/repo/fs.rs

@@ -115,6 +115,12 @@ impl BlockStore for FsBlockStore {
         }
         Ok(())
     }
+
+    async fn list(&self) -> Result<Vec<Cid>, Error> {
+        // unwrapping as we want to panic on poisoned lock
+        let guard = self.cids.lock().unwrap();
+        Ok(guard.iter().cloned().collect())
+    }
 }
 
 #[derive(Clone, Debug)]
@@ -290,6 +296,30 @@ mod tests {
         std::fs::remove_dir_all(&tmp).ok();
     }
 
+    #[async_std::test]
+    async fn test_fs_blockstore_list() {
+        let mut tmp = temp_dir();
+        tmp.push("blockstore_list");
+        std::fs::remove_dir_all(&tmp).ok();
+
+        let block_store = FsBlockStore::new(tmp.clone().into());
+        block_store.init().await.unwrap();
+        block_store.open().await.unwrap();
+
+        for data in &[b"1", b"2", b"3"] {
+            let data_slice = data.to_vec().into_boxed_slice();
+            let cid = Cid::new_v1(Codec::Raw, Sha2_256::digest(&data_slice));
+            let block = Block::new(data_slice, cid);
+            block_store.put(block.clone()).await.unwrap();
+        }
+
+        let cids = block_store.list().await.unwrap();
+        assert_eq!(cids.len(), 3);
+        for cid in cids.iter() {
+            assert!(block_store.contains(cid).await.unwrap());
+        }
+    }
+
     #[async_std::test]
     #[cfg(feature = "rocksdb")]
     fn test_rocks_datastore() {

src/repo/mem.rs

@@ -6,6 +6,8 @@ use async_std::sync::{Arc, Mutex};
 use async_trait::async_trait;
 use bitswap::Block;
 use libipld::cid::Cid;
+
+// FIXME: Transition to Persistent Map to make iterating more consistent
 use std::collections::HashMap;
 
 #[derive(Clone, Debug)]
@@ -61,6 +63,11 @@ impl BlockStore for MemBlockStore {
         self.blocks.lock().await.remove(cid);
         Ok(())
     }
+
+    async fn list(&self) -> Result<Vec<Cid>, Error> {
+        let guard = self.blocks.lock().await;
+        Ok(guard.iter().map(|(cid, _block)| cid).cloned().collect())
+    }
 }
 
 #[derive(Clone, Debug)]
@@ -164,6 +171,29 @@ mod tests {
         assert_eq!(get.await.unwrap(), None);
     }
 
+    #[async_std::test]
+    async fn test_mem_blockstore_list() {
+        let tmp = temp_dir();
+        let mem_store = MemBlockStore::new(tmp.into());
+
+        mem_store.init().await.unwrap();
+        mem_store.open().await.unwrap();
+
+        for data in &[b"1", b"2", b"3"] {
+            let data_slice = data.to_vec().into_boxed_slice();
+            let cid = Cid::new_v1(Codec::Raw, Sha2_256::digest(&data_slice));
+            let block = Block::new(data_slice, cid);
+            mem_store.put(block.clone()).await.unwrap();
+            assert!(mem_store.contains(block.cid()).await.unwrap());
+        }
+
+        let cids = mem_store.list().await.unwrap();
+        assert_eq!(cids.len(), 3);
+        for cid in cids.iter() {
+            assert!(mem_store.contains(cid).await.unwrap());
+        }
+    }
+
     #[async_std::test]
     async fn test_mem_datastore() {
         let tmp = temp_dir();

src/repo/mod.rs

@@ -63,6 +63,7 @@ pub trait BlockStore: Debug + Clone + Send + Sync + Unpin + 'static {
     async fn get(&self, cid: &Cid) -> Result<Option<Block>, Error>;
     async fn put(&self, block: Block) -> Result<(Cid, BlockPut), Error>;
     async fn remove(&self, cid: &Cid) -> Result<(), Error>;
+    async fn list(&self) -> Result<Vec<Cid>, Error>;
 }
 
 #[async_trait]
@@ -183,6 +184,10 @@ impl<TRepoTypes: RepoTypes> Repo<TRepoTypes> {
         }
     }
 
+    pub async fn list_blocks(&self) -> Result<Vec<Cid>, Error> {
+        Ok(self.block_store.list().await?)
+    }
+
     /// Remove block from the block store.
     pub async fn remove_block(&self, cid: &Cid) -> Result<(), Error> {
         if self.is_pinned(cid).await? {