wip salvage

This is originally from the non-merging branch by David in
#126.

Co-authored-by: David Craven <[email protected]>

Author: 2 people

bitswap/src/block.rs

@@ -18,4 +18,8 @@ impl Block {
     pub fn data(&self) -> &[u8] {
         &self.data
     }
+
+    pub fn to_vec(self) -> Vec<u8> {
+        self.data.into()
+    }
 }

http/Cargo.toml

@@ -27,3 +27,5 @@ percent-encoding = "*"
 openssl = "0.10"
 
 prost = "0.6.1"
+cid = "0.4"
+multihash = "0.10"

http/src/v0.rs

@@ -5,6 +5,8 @@ pub mod id;
 pub mod pubsub;
 pub mod swarm;
 pub mod version;
+pub mod block;
+pub mod bitswap;
 
 pub mod support;
 pub use support::recover_as_message_response;
@@ -32,7 +34,10 @@ where
             // Placeholder paths
             // https://docs.rs/warp/0.2.2/warp/macro.path.html#path-prefixes
             .or(warp::path!("add").and_then(not_implemented))
-            .or(warp::path!("bitswap" / ..).and_then(not_implemented))
+            .or(bitswap::wantlist(ipfs))
+            .or(bitswap::stat(ipfs))
+            .or(block::get(ipfs))
+            .or(block::put(ipfs))
             .or(warp::path!("block" / ..).and_then(not_implemented))
             .or(warp::path!("bootstrap" / ..).and_then(not_implemented))
             .or(warp::path!("config" / ..).and_then(not_implemented))

http/src/v0/bitswap.rs

@@ -0,0 +1,99 @@
+use crate::v0::support::{with_ipfs, InvalidPeerId, StringError};
+use ipfs::{BitswapStats, Ipfs, IpfsTypes};
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+use warp::{path, query, reply, Filter, Rejection, Reply};
+
+#[derive(Debug, Deserialize)]
+pub struct WantlistQuery {
+    peer: Option<String>,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct WantlistResponse {
+    keys: Vec<Value>,
+}
+
+async fn wantlist_query<T: IpfsTypes>(ipfs: Ipfs<T>, query: WantlistQuery) -> Result<impl Reply, Rejection> {
+    let peer_id = if let Some(peer_id) = query.peer {
+        let peer_id = peer_id.parse().map_err(|_| InvalidPeerId)?;
+        Some(peer_id)
+    } else {
+        None
+    };
+    let cids = ipfs
+        .bitswap_wantlist(peer_id)
+        .await
+        .map_err(StringError::from)?;
+    let keys = cids
+        .into_iter()
+        .map(|(cid, _)| json!({"/": cid.to_string()}))
+        .collect();
+    let response = WantlistResponse { keys };
+    Ok(reply::json(&response))
+}
+
+pub fn wantlist<T: IpfsTypes>(ipfs: &Ipfs<T>) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
+    path!("bitswap" / "wantlist")
+        .and(with_ipfs(ipfs))
+        .and(query::<WantlistQuery>())
+        .and_then(wantlist_query)
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct StatResponse {
+    blocks_received: u64,
+    blocks_sent: u64,
+    data_received: u64,
+    data_sent: u64,
+    dup_blks_received: u64,
+    dup_data_received: u64,
+    messages_received: u64,
+    peers: Vec<String>,
+    provide_buf_len: i32,
+    wantlist: Vec<Value>,
+}
+
+impl From<BitswapStats> for StatResponse {
+    fn from(stats: BitswapStats) -> Self {
+        let wantlist = stats
+            .wantlist
+            .into_iter()
+            .map(|(cid, _)| json!({"/": cid.to_string()}))
+            .collect();
+        let peers = stats
+            .peers
+            .into_iter()
+            .map(|peer_id| peer_id.to_string())
+            .collect();
+        Self {
+            blocks_received: stats.blocks_received,
+            blocks_sent: stats.blocks_sent,
+            data_received: stats.data_received,
+            data_sent: stats.data_sent,
+            dup_blks_received: stats.dup_blks_received,
+            dup_data_received: stats.dup_data_received,
+            peers,
+            wantlist,
+            messages_received: 0,
+            provide_buf_len: 0,
+        }
+    }
+}
+
+async fn stat_query<T: IpfsTypes>(ipfs: Ipfs<T>) -> Result<impl Reply, Rejection> {
+    let stats: StatResponse = ipfs
+        .bitswap_stats()
+        .await
+        .map_err(StringError::from)?
+        .into();
+    Ok(reply::json(&stats))
+}
+
+pub fn stat<T: IpfsTypes>(ipfs: &Ipfs<T>) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
+    path!("bitswap" / "stat")
+        .and(with_ipfs(ipfs))
+        .and_then(stat_query)
+}

http/src/v0/block.rs

@@ -0,0 +1,133 @@
+use crate::v0::support::{with_ipfs, InvalidMultipartFormData, StringError};
+use cid::{Codec, Version};
+use futures::stream::StreamExt;
+use ipfs::{Cid, Ipfs, IpfsTypes};
+use serde::{Deserialize, Serialize};
+use warp::{http::Response, multipart, path, query, reply, Filter, Rejection, Reply, Buf};
+
+#[derive(Debug, Deserialize)]
+pub struct GetQuery {
+    arg: String,
+}
+
+async fn get_query<T: IpfsTypes>(mut ipfs: Ipfs<T>, query: GetQuery) -> Result<impl Reply, Rejection> {
+    let cid: Cid = query.arg.parse().map_err(StringError::from)?;
+    let data = ipfs
+        .get_block(&cid)
+        .await
+        .map_err(StringError::from)?
+        .to_vec();
+
+    let response = Response::builder().body(data);
+    Ok(response)
+}
+
+pub fn get<T: IpfsTypes>(ipfs: &Ipfs<T>) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
+    path!("block" / "get")
+        .and(with_ipfs(ipfs))
+        .and(query::<GetQuery>())
+        .and_then(get_query)
+}
+
+#[derive(Debug, Deserialize)]
+pub struct PutQuery {
+    format: Option<String>,
+    mhtype: Option<String>,
+    version: Option<u8>,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct PutResponse {
+    key: String,
+    size: usize,
+}
+
+async fn put_query<T: IpfsTypes>(
+    mut ipfs: Ipfs<T>,
+    query: PutQuery,
+    mut form: multipart::FormData,
+) -> Result<impl Reply, Rejection> {
+    let format = match query
+        .format
+        .as_ref()
+        .map(|s| s.as_str())
+        .unwrap_or("dag-pb")
+    {
+        "dag-cbor" => Codec::DagCBOR,
+        "dag-pb" => Codec::DagProtobuf,
+        "dag-json" => Codec::DagJSON,
+        "raw" => Codec::Raw,
+        _ => return Err(StringError::from("unknown codec").into()),
+    };
+    let hasher = match query
+        .mhtype
+        .as_ref()
+        .map(|s| s.as_str())
+        .unwrap_or("sha2-256")
+    {
+        "sha2-256" => multihash::Sha2_256::digest,
+        "sha2-512" => multihash::Sha2_512::digest,
+        _ => return Err(StringError::from("unknown hash").into()),
+    };
+    let version = match query.version.unwrap_or(0) {
+        0 => Version::V0,
+        1 => Version::V1,
+        _ => return Err(StringError::from("invalid cid version").into()),
+    };
+    let mut buf = form
+        .next()
+        .await
+        .ok_or(InvalidMultipartFormData)?
+        .map_err(|_| InvalidMultipartFormData)?
+        .data()
+        .await
+        .ok_or(InvalidMultipartFormData)?
+        .map_err(|_| InvalidMultipartFormData)?;
+    let data = buf.to_bytes().as_ref().to_vec().into_boxed_slice();
+    let digest = hasher(&data);
+    let cid = Cid::new(version, format, digest).map_err(StringError::from)?;
+    let response = PutResponse {
+        key: cid.to_string(),
+        size: data.len(),
+    };
+    let block = ipfs::Block { cid, data };
+    ipfs.put_block(block).await.map_err(StringError::from)?;
+    Ok(reply::json(&response))
+}
+
+pub fn put<T: IpfsTypes>(ipfs: &Ipfs<T>) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
+    path!("block" / "put")
+        .and(with_ipfs(ipfs))
+        .and(query::<PutQuery>())
+        .and(multipart::form())
+        .and_then(put_query)
+}
+
+#[derive(Debug, Deserialize)]
+pub struct RmQuery {}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct RmResponse {}
+
+async fn rm_query<T: IpfsTypes>(_ipfs: Ipfs<T>, _query: RmQuery) -> Result<impl Reply, Rejection> {
+    //let _data = ipfs.put_block(query.cid).await
+    //    .map_err(|e| reject::custom(StringError::from(e)))?;
+    let response = RmResponse {};
+    Ok(reply::json(&response))
+}
+
+pub fn rm<T: IpfsTypes>(ipfs: &Ipfs<T>) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
+    path!("block" / "rm")
+        .and(with_ipfs(ipfs))
+        .and(query::<RmQuery>())
+        .and_then(rm_query)
+}
+
+pub fn stat<T: IpfsTypes>(ipfs: &Ipfs<T>) -> impl Filter<Extract = impl Reply, Error = Rejection> + Clone {
+    path!("block" / "stat")
+        .and(with_ipfs(ipfs))
+        .and(query::<RmQuery>())
+        .and_then(rm_query)
+}

http/src/v0/support.rs

@@ -69,32 +69,50 @@ impl warp::reject::Reject for NonUtf8Topic {}
 pub(crate) struct RequiredArgumentMissing(pub(crate) &'static [u8]);
 impl warp::reject::Reject for RequiredArgumentMissing {}
 
+#[derive(Debug)]
+pub(crate) struct InvalidMultipartFormData;
+impl warp::reject::Reject for InvalidMultipartFormData {}
+impl From<InvalidMultipartFormData> for warp::Rejection {
+    fn from(err: InvalidMultipartFormData) -> warp::Rejection {
+        warp::reject::custom(err)
+    }
+}
+
 /// Marker for `warp` specific rejections when something is unimplemented
 #[derive(Debug)]
 pub(crate) struct NotImplemented;
 impl warp::reject::Reject for NotImplemented {}
-impl Into<warp::reject::Rejection> for NotImplemented {
-    fn into(self) -> warp::reject::Rejection {
-        warp::reject::custom(self)
+impl From<NotImplemented> for warp::Rejection {
+    fn from(err: NotImplemented) -> warp::Rejection {
+        warp::reject::custom(err)
     }
 }
 
 /// PeerId parsing error, details from `libp2p::identity::ParseError` are lost.
 #[derive(Debug)]
 pub(crate) struct InvalidPeerId;
 impl warp::reject::Reject for InvalidPeerId {}
+impl From<InvalidPeerId> for warp::Rejection {
+    fn from(err: InvalidPeerId) -> warp::Rejection {
+        warp::reject::custom(err)
+    }
+}
 
 /// Default placeholder for ipfs::Error but once we get more typed errors we could start making
 /// them more readable, if needed.
+// TODO: needs to be considered if this is even needed..
 #[derive(Debug)]
 pub(crate) struct StringError(Cow<'static, str>);
 impl warp::reject::Reject for StringError {}
+impl From<StringError> for warp::Rejection {
+    fn from(err: StringError) -> warp::Rejection {
+        warp::reject::custom(err)
+    }
+}
 
-// FIXME: it's a bit questionable to keep this but in the beginning it might help us glide in the
-// right direction.
-impl From<ipfs::Error> for StringError {
-    fn from(e: ipfs::Error) -> Self {
-        Self(format!("{}", e).into())
+impl<D: std::fmt::Display> From<D> for StringError {
+    fn from(d: D) -> Self {
+        Self(format!("{}", d).into())
     }
 }
 

src/lib.rs

@@ -632,6 +632,18 @@ impl<Types: SwarmTypes> Future for IpfsFuture<Types> {
     }
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct BitswapStats {
+    pub blocks_sent: u64,
+    pub data_sent: u64,
+    pub blocks_received: u64,
+    pub data_received: u64,
+    pub dup_blks_received: u64,
+    pub dup_data_received: u64,
+    pub peers: Vec<PeerId>,
+    pub wantlist: Vec<(Cid, bitswap::Priority)>,
+}
+
 #[doc(hidden)]
 pub use node::Node;