banyan_utils/
ipfs.rs

1//! helper methods to work with ipfs/ipld
2use anyhow::{anyhow, Result};
3use banyan::store::{BlockWriter, ReadOnlyStore};
4use futures::prelude::*;
5use libipld::Cid;
6use serde::{de::IgnoredAny, de::Visitor, Deserialize, Deserializer, Serialize, Serializer};
7use std::{convert::TryInto, fmt, str::FromStr};
8
9use crate::tags::Sha256Digest;
10
11pub fn block_get(key: &Cid) -> Result<Box<[u8]>> {
12    let url = reqwest::Url::parse_with_params(
13        "http://localhost:5001/api/v0/block/get",
14        &[("arg", format!("{}", key))],
15    )?;
16    let client = reqwest::blocking::Client::new();
17    let data: Vec<u8> = client.post(url).send()?.bytes()?.to_vec();
18    Ok(data.into())
19}
20
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct Base64Blob(pub Vec<u8>);
23
24impl Serialize for Base64Blob {
25    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
26    where
27        S: Serializer,
28    {
29        serializer.serialize_str(base64::encode(&self.0).as_ref())
30    }
31}
32
33impl<'de> Deserialize<'de> for Base64Blob {
34    fn deserialize<D>(deserializer: D) -> Result<Base64Blob, D::Error>
35    where
36        D: Deserializer<'de>,
37    {
38        struct MyVisitor();
39
40        impl<'de> Visitor<'de> for MyVisitor {
41            type Value = Base64Blob;
42
43            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
44                formatter.write_str("string")
45            }
46
47            fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
48            where
49                E: serde::de::Error,
50            {
51                base64::decode(v).map(Base64Blob).map_err(|err| {
52                    serde::de::Error::custom(format!("Error decoding base64 string: {}", err))
53                })
54            }
55        }
56
57        deserializer.deserialize_any(MyVisitor())
58    }
59}
60
61#[derive(Debug, Clone, Deserialize)]
62struct IpfsPubsubEventIo {
63    // We only need this field, skip deserializing everything else. We still keep the rest of the
64    // fields to ensure that we are parsing the right thing.
65    data: Base64Blob,
66    #[serde(rename = "from")]
67    _from: IgnoredAny,
68    #[serde(rename = "seqno")]
69    _seqno: IgnoredAny,
70    #[serde(rename = "topicIDs")]
71    _topic_ids: IgnoredAny,
72}
73
74pub fn pubsub_sub(topic: &str) -> Result<impl Stream<Item = reqwest::Result<Vec<u8>>>> {
75    let url = reqwest::Url::parse_with_params(
76        "http://localhost:5001/api/v0/pubsub/sub",
77        &[("arg", topic)],
78    )?;
79    let client = reqwest::Client::new();
80    let data = client
81        .post(url)
82        .send()
83        .map(|response| match response {
84            Ok(response) => response
85                .bytes_stream()
86                .map_ok(|bytes| {
87                    let data: IpfsPubsubEventIo = serde_json::from_slice(&bytes).unwrap();
88                    data.data.0
89                })
90                .left_stream(),
91            Err(cause) => futures::stream::once(future::err(cause)).right_stream(),
92        })
93        .flatten_stream();
94    Ok(data)
95}
96
97pub async fn pubsub_pub(topic: &str, data: &[u8]) -> Result<()> {
98    use percent_encoding::{percent_encode, NON_ALPHANUMERIC};
99    let topic = percent_encode(topic.as_bytes(), NON_ALPHANUMERIC).to_string();
100    let data = percent_encode(data, NON_ALPHANUMERIC).to_string();
101    let url = reqwest::Url::parse(&format!(
102        "http://localhost:5001/api/v0/pubsub/pub?arg={}&arg={}",
103        topic, data
104    ))?;
105    let client = reqwest::Client::new();
106    let _ = client.post(url).send().await?.bytes().await?.to_vec();
107    Ok(())
108}
109
110fn format_codec(codec: u64) -> Result<&'static str> {
111    match codec {
112        0x71 => Ok("cbor"),
113        0x70 => Ok("protobuf"),
114        0x55 => Ok("raw"),
115        _ => Err(anyhow!("unsupported codec {}", codec)),
116    }
117}
118
119pub fn block_put(data: &[u8], codec: u64, pin: bool) -> Result<Cid> {
120    let url = reqwest::Url::parse_with_params(
121        "http://localhost:5001/api/v0/block/put",
122        &[("format", format_codec(codec)?), ("pin", &pin.to_string())],
123    )?;
124    let client = reqwest::blocking::Client::new();
125    let form = reqwest::blocking::multipart::Form::new().part(
126        "file",
127        reqwest::blocking::multipart::Part::bytes(data.to_vec()),
128    );
129    let res: IpfsBlockPutResponseIo = client.post(url).multipart(form).send()?.json()?;
130    let cid = Cid::from_str(&res.key)?;
131    Ok(cid)
132}
133
134#[derive(Clone)]
135pub struct IpfsStore;
136
137impl IpfsStore {
138    pub fn new() -> anyhow::Result<Self> {
139        Ok(Self)
140    }
141}
142
143impl ReadOnlyStore<Sha256Digest> for IpfsStore {
144    fn get(&self, link: &Sha256Digest) -> Result<Box<[u8]>> {
145        let cid: Cid = (*link).into();
146        std::thread::spawn(move || crate::ipfs::block_get(&cid))
147            .join()
148            .map_err(|_| anyhow!("join error!"))?
149    }
150}
151
152impl BlockWriter<Sha256Digest> for IpfsStore {
153    fn put(&mut self, data: Vec<u8>) -> Result<Sha256Digest> {
154        let cid = std::thread::spawn(move || crate::ipfs::block_put(&data, 0x71, false))
155            .join()
156            .map_err(|_| anyhow!("join error!"))??;
157        assert!(cid.hash().code() == 0x12);
158        assert!(cid.hash().digest().len() == 32);
159        cid.try_into()
160    }
161}
162
163#[derive(Deserialize)]
164struct IpfsBlockPutResponseIo {
165    #[serde(rename = "Key")]
166    key: String,
167}
banyan_utils/ipfs.rs

banyan_utils/
ipfs.rs
