Add some tests and refactor examples.

README.md

@@ -4,8 +4,6 @@ Currently implements an altruistic bitswap strategy over mdns.
 ## Getting started
 ```rust
 #![feature(async_await, await_macro, futures_api)]
-use futures::future::FutureObj;
-use futures::prelude::*;
 use ipfs::{Block, Ipfs, IpfsOptions, Types};
 
 fn main() {
@@ -15,8 +13,8 @@ fn main() {
     let block = Block::from("hello block2\n");
     let cid = Block::from("hello block\n").cid();
 
-    tokio::run(FutureObj::new(Box::new(async move {
-        tokio::spawn(ipfs.start_daemon().compat());
+    tokio::run_async(async move {
+        tokio::spawn_async(ipfs.start_daemon());
         await!(ipfs.init_repo()).unwrap();
         await!(ipfs.open_repo()).unwrap();
 
@@ -25,8 +23,8 @@ fn main() {
         println!("Received block with contents: {:?}",
                  String::from_utf8_lossy(&block.data()));
 
-        Ok(())
-    })).compat());
+        ipfs.exit_daemon();
+    });
 }
 ```
 

examples/client1.rs

@@ -1,6 +1,4 @@
 #![feature(async_await, await_macro, futures_api)]
-use futures::future::FutureObj;
-use futures::prelude::*;
 use ipfs::{Block, Ipfs, IpfsOptions, Types};
 
 fn main() {
@@ -10,8 +8,8 @@ fn main() {
     let block = Block::from("hello block2\n");
     let cid = Block::from("hello block\n").cid();
 
-    tokio::run(FutureObj::new(Box::new(async move {
-        tokio::spawn(ipfs.start_daemon().compat());
+    tokio::run_async(async move {
+        tokio::spawn_async(ipfs.start_daemon());
         await!(ipfs.init_repo()).unwrap();
         await!(ipfs.open_repo()).unwrap();
 
@@ -19,7 +17,5 @@ fn main() {
         let block = await!(ipfs.get_block(cid.clone())).unwrap();
         println!("Received block with contents: {:?}",
                  String::from_utf8_lossy(&block.data()));
-
-        Ok(())
-    })).compat());
+    });
 }

examples/client2.rs

@@ -1,6 +1,4 @@
 #![feature(async_await, await_macro, futures_api)]
-use futures::future::FutureObj;
-use futures::prelude::*;
 use ipfs::{Block, Ipfs, IpfsOptions, RepoTypes, SwarmTypes, IpfsTypes};
 
 #[derive(Clone)]
@@ -24,13 +22,12 @@ fn main() {
     let block = Block::from("hello block\n");
     let cid = Block::from("hello block2\n").cid();
 
-    tokio::run(FutureObj::new(Box::new(async move {
-        tokio::spawn(ipfs.start_daemon().compat());
+    tokio::run_async(async move {
+        tokio::spawn_async(ipfs.start_daemon());
 
         await!(ipfs.put_block(block)).unwrap();
         let block = await!(ipfs.get_block(cid)).unwrap();
         println!("Received block with contents: {:?}",
                  String::from_utf8_lossy(&block.data()));
-        Ok(())
-    })).compat());
+    });
 }

src/ipld/path.rs

@@ -69,6 +69,12 @@ impl IpldPath {
     }
 }
 
+impl From<Cid> for IpldPath {
+    fn from(cid: Cid) -> Self {
+        IpldPath::new(cid)
+    }
+}
+
 #[derive(Clone, Debug, PartialEq)]
 pub enum SubPath {
     Key(String),

src/lib.rs

@@ -113,6 +113,7 @@ enum IpfsEvent {
     WantBlock(Cid),
     ProvideBlock(Cid),
     UnprovideBlock(Cid),
+    Exit,
 }
 
 impl<Types: IpfsTypes> Ipfs<Types> {
@@ -190,6 +191,11 @@ impl<Types: IpfsTypes> Ipfs<Types> {
             swarm: Box::new(swarm),
         }
     }
+
+    /// Exit daemon.
+    pub fn exit_daemon(&mut self) {
+        self.events.lock().unwrap().push_back(IpfsEvent::Exit)
+    }
 }
 
 pub struct IpfsFuture<Types: SwarmTypes> {
@@ -198,7 +204,7 @@ pub struct IpfsFuture<Types: SwarmTypes> {
 }
 
 impl<Types: SwarmTypes> Future for IpfsFuture<Types> {
-    type Output = Result<(), ()>;
+    type Output = ();
 
     fn poll(self: Pin<&mut Self>, _waker: &Waker) -> Poll<Self::Output> {
         let _self = self.get_mut();
@@ -214,13 +220,16 @@ impl<Types: SwarmTypes> Future for IpfsFuture<Types> {
                     IpfsEvent::UnprovideBlock(cid) => {
                         _self.swarm.stop_providing_block(&cid);
                     }
+                    IpfsEvent::Exit => {
+                        return Poll::Ready(());
+                    }
                 }
             }
             let poll = _self.swarm.poll().expect("Error while polling swarm");
             match poll {
                 Async::Ready(Some(_)) => {},
                 Async::Ready(None) => {
-                    return Poll::Ready(Ok(()));
+                    return Poll::Ready(());
                 },
                 Async::NotReady => {
                     return Poll::Pending;
@@ -232,18 +241,53 @@ impl<Types: SwarmTypes> Future for IpfsFuture<Types> {
 
 #[cfg(test)]
 mod tests {
-    /*
     use super::*;
 
+    #[derive(Clone)]
+    struct Types;
+
+    impl RepoTypes for Types {
+        type TBlockStore = crate::repo::mem::MemBlockStore;
+        type TDataStore = crate::repo::mem::MemDataStore;
+    }
+
+    impl SwarmTypes for Types {
+        type TStrategy = crate::bitswap::strategy::AltruisticStrategy<Self>;
+    }
+
+    impl IpfsTypes for Types {}
+
     #[test]
     fn test_put_and_get_block() {
         let options = IpfsOptions::test();
-        let mut ipfs = Ipfs::new(options);
+        let mut ipfs = Ipfs::<Types>::new(options);
         let block = Block::from("hello block\n");
-        let cid = ipfs.put_block(block.clone());
-        let future = ipfs.get_block(cid).map(move |new_block| {
+
+        tokio::run_async(async move {
+            tokio::spawn_async(ipfs.start_daemon());
+
+            let cid = await!(ipfs.put_block(block.clone())).unwrap();
+            let new_block = await!(ipfs.get_block(cid)).unwrap();
             assert_eq!(block, new_block);
+
+            ipfs.exit_daemon();
         });
-        run_ipfs(ipfs, future);
-    }*/
+    }
+
+    #[test]
+    fn test_put_and_get_dag() {
+        let options = IpfsOptions::test();
+        let mut ipfs = Ipfs::<Types>::new(options);
+
+        tokio::run_async(async move {
+            tokio::spawn_async(ipfs.start_daemon());
+
+            let data: Ipld = vec![-1, -2, -3].into();
+            let cid = await!(ipfs.put_dag(data.clone())).unwrap();
+            let new_data = await!(ipfs.get_dag(cid.into())).unwrap();
+            assert_eq!(Some(data), new_data);
+
+            ipfs.exit_daemon();
+        });
+    }
 }