Upgrading Tokio for 0.2.0, adding examples with actix and actix-web

Cargo.toml

@@ -10,14 +10,18 @@ keywords = ["asynchronous", "file", "pipe", "stdio", "tokio"]
 categories = ["asynchronous"]
 license = "MIT/Apache-2.0"
 exclude = [".gitignore", "tests/seek.txt"]
+edition = "2018"
 
 [dependencies]
 bytes = "0.4.2"
 libc = "0.2.21"
 mio = "0.6.6"
-tokio-io = "0.1.1"
-tokio-reactor = "0.1.1"
+tokio = { version = "0.2.6", features = ["io-util", "io-driver", "stream"] }
+tokio-util = { version = "0.2.0", features = ["codec"] }
 
 [dev-dependencies]
-futures = "0.1.10"
-tokio = "0.1.5"
+futures = "0.3.8"
+tokio = { version = "0.2", features = ["full", "io-util"] }
+actix = "0.9.0"
+actix-rt = "1.0.0"
+actix-web = "2.0.0-rc"

examples/seek.rs

@@ -1,21 +1,23 @@
-extern crate futures;
 extern crate tokio;
 extern crate tokio_file_unix;
 
+use tokio::prelude::*;
+use tokio::io::AsyncWriteExt;
 use std::fs::File as StdFile;
 use std::io::{Seek, SeekFrom};
 
-use futures::Future;
-
-fn main() {
+#[tokio::main]
+async fn main() {
     let file = StdFile::create("tests/seek.txt").unwrap();
     file.set_len(0x11).unwrap();
     let file = tokio_file_unix::File::new_nb(file).unwrap();
-    let file = file.into_io(&tokio::reactor::Handle::current()).unwrap();
+    let mut file = file.into_io().unwrap();
 
-    let (mut file, _) = tokio::io::write_all(file, b"aaaaAAAAaaaaAAAA\n").wait().unwrap();
+    file.write_all(b"aaaaAAAAaaaaAAAA\n").await.unwrap();
     file.get_mut().seek(SeekFrom::Start(8)).unwrap();
-    let (mut file, _) = tokio::io::write_all(file, [b'b'; 8]).wait().unwrap();
+    file.write_all(&[b'b'; 8]).await.unwrap();
     file.get_mut().seek(SeekFrom::Start(2)).unwrap();
-    tokio::io::write_all(file, [b'c'; 4]).wait().unwrap();
+    file.write_all(&[b'c'; 4]).await.unwrap();
+
+    ()
 }

examples/stdin.rs

@@ -1,41 +1,22 @@
-extern crate futures;
 extern crate tokio;
 extern crate tokio_file_unix;
 
-use futures::{Future, future};
+use tokio_util::codec::FramedRead;
+use tokio_util::codec::LinesCodec;
+use crate::tokio::stream::StreamExt;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     // convert stdin into a nonblocking file;
     // this is the only part that makes use of tokio_file_unix
     let file = tokio_file_unix::raw_stdin().unwrap();
     let file = tokio_file_unix::File::new_nb(file).unwrap();
-    let file = file.into_reader(&tokio::reactor::Handle::current()).unwrap();
+    let file = file.into_io().unwrap();
 
-    println!("Type something and hit enter!");
-    tokio::run(future::loop_fn((file, Vec::new()), |(file, line)| {
-        // read each line
-        tokio::io::read_until(file, b'\n', line).map(|(file, mut line)| {
-
-            // demonstrate that the event loop isn't blocked by I/O!
-            let one_sec_from_now =
-                std::time::Instant::now()
-                + std::time::Duration::new(1, 0);
-            tokio::spawn(
-                tokio::timer::Delay::new(one_sec_from_now)
-                .map_err(|_| ())
-                .map(|()| eprintln!(" ... timeout works!"))
-            );
+    let mut framed = FramedRead::new(file, LinesCodec::new());
 
-            if line.ends_with(b"\n") {
-                println!("Got: {:?}", std::str::from_utf8(&line));
-                line.clear();
-                future::Loop::Continue((file, line))
-            } else {                    // EOF
-                if !line.is_empty() {
-                    println!("Got: {:?}", std::str::from_utf8(&line));
-                }
-                future::Loop::Break(())
-            }
-        })
-    }).map_err(|e| panic!("{:?}", e)));
+    println!("Type something and hit enter!");
+    while let Some(got) = framed.next().await {
+        println!("Got: {:?}", got);
+    }
 }

examples/stdin_actix.rs

@@ -0,0 +1,20 @@
+extern crate tokio;
+extern crate tokio_file_unix;
+
+use tokio_util::codec::FramedRead;
+use tokio_util::codec::LinesCodec;
+use crate::tokio::stream::StreamExt;
+
+#[actix_rt::main]
+async fn main() {
+    let file = tokio_file_unix::raw_stdin().unwrap();
+    let file = tokio_file_unix::File::new_nb(file).unwrap();
+    let file = file.into_io().unwrap();
+
+    let mut framed = FramedRead::new(file, LinesCodec::new());
+
+    println!("Type something and hit enter!");
+    while let Some(got) = framed.next().await {
+        println!("Got: {:?}", got);
+    }
+}

examples/stdin_actix_web.rs

@@ -0,0 +1,55 @@
+extern crate futures;
+extern crate tokio;
+extern crate tokio_file_unix;
+
+use futures::{pin_mut, select};
+use futures::future::FutureExt;
+use tokio_util::codec::FramedRead;
+use tokio_util::codec::LinesCodec;
+use crate::tokio::stream::StreamExt;
+use actix_web::{get, web, App, HttpServer, Responder};
+use actix_web::client::{Client};
+
+#[get("/{something}")]
+async fn index(info: web::Path<String>) -> impl Responder {
+    format!("Hello Got this: {}", info)
+}
+
+#[actix_rt::main]
+async fn main() -> std::io::Result<()> {
+    println!("Type something and hit enter!");
+    let stdin_fut = async {
+        let file = tokio_file_unix::raw_stdin().unwrap();
+        let file = tokio_file_unix::File::new_nb(file).unwrap();
+        let file = file.into_io().unwrap();
+
+        let client = Client::default();
+
+        let mut framed = FramedRead::new(file, LinesCodec::new());
+
+        while let Some(got) = framed.next().await {
+            println!("Sending this: {:?}", got);
+
+            let mut response = match client.get(format!("http://127.0.0.1:8080/{}", got.unwrap())).send().await {
+                Err(e) => panic!("{:?}", e),
+                Ok(t) => t
+            };
+
+            let body = response.body().await.unwrap();
+
+            println!("Got bytes: {:?}", String::from_utf8(body.to_vec()).unwrap());
+        }
+        Ok(())
+    }.fuse();
+
+    let server_fut = HttpServer::new(|| App::new().service(index))
+        .bind("127.0.0.1:8080")?
+        .run()
+        .fuse();
+
+    pin_mut!(stdin_fut, server_fut);
+    select! {
+        result = stdin_fut => result,
+        result = server_fut => result,
+    }
+}