Async/Await Support (continuation of #191)

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.7.0
+
+* async-await support [#229](https://github.com/softprops/shiplift/pull/229)
+
 # 0.6.0
 
 * add chrono as an optional feature, enabled by default [#190](https://github.com/softprops/shiplift/pull/190)

Cargo.toml

@@ -18,28 +18,31 @@ coveralls = { repository = "softprops/shipflit" }
 maintenance = { status = "actively-developed" }
 
 [dependencies]
-log = "0.4"
-mime = "0.3"
 base64 = "0.11"
 byteorder = "1.3"
 bytes = "0.4"
 chrono = { version = "0.4", optional = true, features = ["serde"] }
 flate2 = "1.0"
-futures = "0.1"
-hyper = "0.12"
-hyper-openssl = { version = "0.7", optional = true }
-hyperlocal = { version = "0.6", optional = true }
+futures-util = "0.3"
+futures_codec = "0.3"
+hyper = "0.13"
+hyper-openssl = { version = "0.8", optional = true }
+hyperlocal = { version = "0.7", optional = true }
+log = "0.4"
+mime = "0.3"
 openssl = { version = "0.10", optional = true }
+pin-project = "0.4"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 tar = "0.4"
-tokio = "0.1"
-tokio-codec = "0.1"
-tokio-io = "0.1"
+tokio = "0.2"
 url = "2.1"
 
 [dev-dependencies]
 env_logger = "0.7"
+# Required for examples to run
+futures = "0.3.1"
+tokio = { version = "0.2.6", features = ["macros"] }
 
 [features]
 default = ["chrono", "unix-socket", "tls"]

examples/attach.rs

@@ -0,0 +1,32 @@
+use futures::StreamExt;
+use shiplift::{tty::TtyChunk, Docker};
+use std::env;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let docker = Docker::new();
+    let id = env::args()
+        .nth(1)
+        .expect("You need to specify a container id");
+
+    let tty_multiplexer = docker.containers().get(&id).attach().await?;
+
+    let (mut reader, _writer) = tty_multiplexer.split();
+
+    while let Some(tty_result) = reader.next().await {
+        match tty_result {
+            Ok(chunk) => print_chunk(chunk),
+            Err(e) => eprintln!("Error: {}", e),
+        }
+    }
+
+    Ok(())
+}
+
+fn print_chunk(chunk: TtyChunk) {
+    match chunk {
+        TtyChunk::StdOut(bytes) => println!("Stdout: {}", std::str::from_utf8(&bytes).unwrap()),
+        TtyChunk::StdErr(bytes) => eprintln!("Stdout: {}", std::str::from_utf8(&bytes).unwrap()),
+        TtyChunk::StdIn(_) => unreachable!(),
+    }
+}

examples/containercopyfrom.rs

@@ -1,26 +1,27 @@
+use futures::TryStreamExt;
 use shiplift::Docker;
 use std::{env, path};
-use tokio::prelude::{Future, Stream};
+use tar::Archive;
 
-fn main() {
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let docker = Docker::new();
     let id = env::args()
         .nth(1)
         .expect("Usage: cargo run --example containercopyfrom -- <container> <path in container>");
     let path = env::args()
         .nth(2)
         .expect("Usage: cargo run --example containercopyfrom -- <container> <path in container>");
-    let fut = docker
+
+    let bytes = docker
         .containers()
         .get(&id)
         .copy_from(path::Path::new(&path))
-        .collect()
-        .and_then(|stream| {
-            let tar = stream.concat();
-            let mut archive = tar::Archive::new(tar.as_slice());
-            archive.unpack(env::current_dir()?)?;
-            Ok(())
-        })
-        .map_err(|e| eprintln!("Error: {}", e));
-    tokio::run(fut);
+        .try_concat()
+        .await?;
+
+    let mut archive = Archive::new(&bytes[..]);
+    archive.unpack(env::current_dir()?)?;
+
+    Ok(())
 }

examples/containercopyinto.rs

@@ -1,8 +1,8 @@
 use shiplift::Docker;
-use std::env;
-use tokio::prelude::Future;
+use std::{env, fs::File, io::Read};
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let docker = Docker::new();
     let path = env::args()
         .nth(1)
@@ -11,17 +11,17 @@ fn main() {
         .nth(2)
         .expect("Usage: cargo run --example containercopyinto -- <local path> <container>");
 
-    use std::{fs::File, io::prelude::*};
-
     let mut file = File::open(&path).unwrap();
     let mut bytes = Vec::new();
     file.read_to_end(&mut bytes)
         .expect("Cannot read file on the localhost.");
 
-    let fut = docker
+    if let Err(e) = docker
         .containers()
         .get(&id)
-        .copy_file_into(path, &bytes[..])
-        .map_err(|e| eprintln!("Error: {}", e));
-    tokio::run(fut);
+        .copy_file_into(path, &bytes)
+        .await
+    {
+        eprintln!("Error: {}", e)
+    }
 }

examples/containercreate.rs

@@ -1,16 +1,19 @@
 use shiplift::{ContainerOptions, Docker};
 use std::env;
-use tokio::prelude::Future;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let docker = Docker::new();
     let image = env::args()
         .nth(1)
         .expect("You need to specify an image name");
-    let fut = docker
+
+    match docker
         .containers()
         .create(&ContainerOptions::builder(image.as_ref()).build())
-        .map(|info| println!("{:?}", info))
-        .map_err(|e| eprintln!("Error: {}", e));
-    tokio::run(fut);
+        .await
+    {
+        Ok(info) => println!("{:?}", info),
+        Err(e) => eprintln!("Error: {}", e),
+    }
 }

examples/containerdelete.rs

@@ -1,16 +1,14 @@
 use shiplift::Docker;
 use std::env;
-use tokio::prelude::Future;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let docker = Docker::new();
     let id = env::args()
         .nth(1)
         .expect("You need to specify an container id");
-    let fut = docker
-        .containers()
-        .get(&id)
-        .delete()
-        .map_err(|e| eprintln!("Error: {}", e));
-    tokio::run(fut);
+
+    if let Err(e) = docker.containers().get(&id).delete().await {
+        eprintln!("Error: {}", e)
+    }
 }

examples/containerexec.rs

@@ -1,8 +1,9 @@
-use shiplift::{tty::StreamType, Docker, ExecContainerOptions};
-use std::env;
-use tokio::prelude::{Future, Stream};
+use futures::StreamExt;
+use shiplift::{tty::TtyChunk, Docker, ExecContainerOptions};
+use std::{env, str::from_utf8};
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let docker = Docker::new();
     let id = env::args()
         .nth(1)
@@ -18,19 +19,19 @@ fn main() {
         .attach_stdout(true)
         .attach_stderr(true)
         .build();
-    let fut = docker
-        .containers()
-        .get(&id)
-        .exec(&options)
-        .for_each(|chunk| {
-            match chunk.stream_type {
-                StreamType::StdOut => println!("Stdout: {}", chunk.as_string_lossy()),
-                StreamType::StdErr => eprintln!("Stderr: {}", chunk.as_string_lossy()),
-                StreamType::StdIn => unreachable!(),
-            }
-            Ok(())
-        })
-        .map_err(|e| eprintln!("Error: {}", e));
 
-    tokio::run(fut);
+    while let Some(exec_result) = docker.containers().get(&id).exec(&options).next().await {
+        match exec_result {
+            Ok(chunk) => print_chunk(chunk),
+            Err(e) => eprintln!("Error: {}", e),
+        }
+    }
+}
+
+fn print_chunk(chunk: TtyChunk) {
+    match chunk {
+        TtyChunk::StdOut(bytes) => println!("Stdout: {}", from_utf8(&bytes).unwrap()),
+        TtyChunk::StdErr(bytes) => eprintln!("Stdout: {}", from_utf8(&bytes).unwrap()),
+        TtyChunk::StdIn(_) => unreachable!(),
+    }
 }

examples/containerinspect.rs

@@ -1,17 +1,15 @@
 use shiplift::Docker;
 use std::env;
-use tokio::prelude::Future;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let docker = Docker::new();
     let id = env::args()
         .nth(1)
         .expect("Usage: cargo run --example containerinspect -- <container>");
-    let fut = docker
-        .containers()
-        .get(&id)
-        .inspect()
-        .map(|container| println!("{:#?}", container))
-        .map_err(|e| eprintln!("Error: {}", e));
-    tokio::run(fut);
+
+    match docker.containers().get(&id).inspect().await {
+        Ok(container) => println!("{:#?}", container),
+        Err(e) => eprintln!("Error: {}", e),
+    }
 }

examples/containers.rs

@@ -1,18 +1,15 @@
 use shiplift::Docker;
-use tokio::prelude::Future;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     env_logger::init();
     let docker = Docker::new();
-    let fut = docker
-        .containers()
-        .list(&Default::default())
-        .map(|containers| {
+    match docker.containers().list(&Default::default()).await {
+        Ok(containers) => {
             for c in containers {
                 println!("container -> {:#?}", c)
             }
-        })
-        .map_err(|e| eprintln!("Error: {}", e));
-
-    tokio::run(fut);
+        }
+        Err(e) => eprintln!("Error: {}", e),
+    }
 }

examples/custom_host.rs

@@ -1,13 +1,10 @@
-use futures::Future;
 use shiplift::Docker;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let docker = Docker::host("http://yourhost".parse().unwrap());
-
-    let fut = docker
-        .ping()
-        .map(|pong| println!("Ping: {}", pong))
-        .map_err(|e| eprintln!("Error: {}", e));
-
-    tokio::run(fut);
+    match docker.ping().await {
+        Ok(pong) => println!("Ping: {}", pong),
+        Err(e) => eprintln!("Error: {}", e),
+    }
 }

examples/events.rs

@@ -1,16 +1,15 @@
+use futures::StreamExt;
 use shiplift::Docker;
-use tokio::prelude::{Future, Stream};
 
-fn main() {
+#[tokio::main]
+async fn main() {
     let docker = Docker::new();
     println!("listening for events");
 
-    let fut = docker
-        .events(&Default::default())
-        .for_each(|e| {
-            println!("event -> {:?}", e);
-            Ok(())
-        })
-        .map_err(|e| eprintln!("Error: {}", e));
-    tokio::run(fut);
+    while let Some(event_result) = docker.events(&Default::default()).next().await {
+        match event_result {
+            Ok(event) => println!("event -> {:?}", event),
+            Err(e) => eprintln!("Error: {}", e),
+        }
+    }
 }

examples/export.rs

@@ -1,8 +1,10 @@
-use shiplift::{errors::Error, Docker};
+use futures::StreamExt;
 use std::{env, fs::OpenOptions, io::Write};
-use tokio::prelude::{Future, Stream};
 
-fn main() {
+use shiplift::{errors::Error, Docker};
+
+#[tokio::main]
+async fn main() {
     let docker = Docker::new();
     let id = env::args().nth(1).expect("You need to specify an image id");
 
@@ -11,17 +13,13 @@ fn main() {
         .create(true)
         .open(format!("{}.tar", &id))
         .unwrap();
+
     let images = docker.images();
-    let fut = images
-        .get(&id)
-        .export()
-        .for_each(move |bytes| {
-            export_file
-                .write(&bytes[..])
-                .map(|n| println!("copied {} bytes", n))
-                .map_err(Error::IO)
-        })
-        .map_err(|e| eprintln!("Error: {}", e));
 
-    tokio::run(fut)
+    while let Some(export_result) = images.get(&id).export().next().await {
+        match export_result.and_then(|bytes| export_file.write(&bytes).map_err(Error::from)) {
+            Ok(n) => println!("copied {} bytes", n),
+            Err(e) => eprintln!("Error: {}", e),
+        }
+    }
 }