Actix web 2.0 (#5343)

* upgrade to actix-2.0

* fix prep queries names

frameworks/Rust/actix/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix"
-version = "1.0.0"
+version = "2.0.0"
 edition = "2018"
 
 [[bin]]
@@ -24,31 +24,30 @@ name = "actix-raw"
 path = "src/main_raw.rs"
 
 [dependencies]
-actix = { version="0.8.2", features=["http"] }
-actix-web = { version="1.0.0-rc", default-features = false }
-actix-http = { version="0.2.1", default-features = false }
-actix-rt = "0.2.2"
-actix-codec = "0.1.2"
-actix-server = "0.5.1"
-actix-service = "0.4.0"
-jemallocator = "0.3.0"
+actix = { version="0.9.0", features=["http"] }
+actix-web = { version="2.0.0", default-features = false }
+actix-http = { version="1.0.1", default-features = false }
+actix-rt = "1.0.0"
+actix-codec = "0.2.0"
+actix-server = "1.0.0"
+actix-service = "1.0.1"
+jemallocator = "0.3.2"
 askama = "0.8"
 markup = "0.3.1"
 yarte = "0.3"
 serde = "1.0"
 serde_json = "1.0"
 serde_derive = "1.0"
-env_logger = "0.6"
+env_logger = "0.7"
 rand = "0.5"
-bytes = "0.4"
+bytes = "0.5.3"
 num_cpus = "1.0"
-futures = "0.1.25"
-http = "0.1"
-diesel = { version = "1.2", features = ["postgres"] }
+futures = "0.3.1"
+http = "0.2"
+diesel = { version = "1.4.3", features = ["postgres"] }
 url = { version="1.7", features=["query_encoding"] }
-log = { version = "0.4", features = ["release_max_level_off"] }
-v_htmlescape = "0.4"
-tokio-tcp = "0.1"
+log = { version = "0.4", features = ["release_max_level_debug"] }
+v_htmlescape = "0.4.5"
 tokio-postgres = { git="https://github.com/fafhrd91/rust-postgres.git" }
 
 [build-dependencies]
@@ -59,6 +58,3 @@ bindgen = "0.42"
 lto = true
 opt-level = 3
 codegen-units = 1
-
-[patch.crates-io]
-tokio-reactor = { git="https://github.com/fafhrd91/tokio.git" }

frameworks/Rust/actix/actix-core.dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.36
+FROM rust:1.40
 
 ADD ./ /actix
 WORKDIR /actix

frameworks/Rust/actix/actix-diesel.dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.36
+FROM rust:1.40
 
 ADD ./ /actix
 WORKDIR /actix

frameworks/Rust/actix/actix-pg.dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.36
+FROM rust:1.40
 
 ADD ./ /actix
 WORKDIR /actix

frameworks/Rust/actix/actix-raw.dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.36
+FROM rust:1.40
 
 ADD ./ /actix
 WORKDIR /actix

frameworks/Rust/actix/actix.dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.36
+FROM rust:1.40
 
 ADD ./ /actix
 WORKDIR /actix

frameworks/Rust/actix/src/db_pg.rs

@@ -1,10 +1,10 @@
 use std::fmt::Write;
 use std::io;
 
-use actix::fut;
 use actix::prelude::*;
 use bytes::{Bytes, BytesMut};
-use futures::{stream, Future, Stream};
+use futures::stream::futures_unordered::FuturesUnordered;
+use futures::{FutureExt, StreamExt, TryStreamExt};
 use rand::{thread_rng, Rng, ThreadRng};
 use tokio_postgres::{connect, Client, NoTls, Statement};
 
@@ -13,9 +13,9 @@ use crate::utils::{Fortune, Writer};
 
 /// Postgres interface
 pub struct PgConnection {
-    cl: Option<Client>,
-    fortune: Option<Statement>,
-    world: Option<Statement>,
+    cl: Client,
+    fortune: Statement,
+    world: Statement,
     rng: ThreadRng,
 }
 
@@ -24,47 +24,24 @@ impl Actor for PgConnection {
 }
 
 impl PgConnection {
-    pub fn connect(db_url: &str) -> Addr<PgConnection> {
-        let hs = connect(db_url, NoTls);
-
-        PgConnection::create(move |ctx| {
-            let act = PgConnection {
-                cl: None,
-                fortune: None,
-                world: None,
-                rng: thread_rng(),
-            };
-
-            hs.map_err(|_| panic!("can not connect to postgresql"))
-                .into_actor(&act)
-                .and_then(|(mut cl, conn), act, ctx| {
-                    ctx.wait(
-                        cl.prepare("SELECT id, message FROM fortune")
-                            .map_err(|_| ())
-                            .into_actor(act)
-                            .and_then(|st, act, _| {
-                                act.fortune = Some(st);
-                                fut::ok(())
-                            }),
-                    );
-                    ctx.wait(
-                        cl.prepare("SELECT id, randomnumber FROM world WHERE id=$1")
-                            .map_err(|_| ())
-                            .into_actor(act)
-                            .and_then(|st, act, _| {
-                                act.world = Some(st);
-                                fut::ok(())
-                            }),
-                    );
-
-                    act.cl = Some(cl);
-                    Arbiter::spawn(conn.map_err(|e| panic!("{}", e)));
-                    fut::ok(())
-                })
-                .wait(ctx);
-
-            act
-        })
+    pub async fn connect(db_url: &str) -> Result<Addr<PgConnection>, io::Error> {
+        let (cl, conn) = connect(db_url, NoTls)
+            .await
+            .expect("can not connect to postgresql");
+        actix_rt::spawn(conn.map(|res| panic!("{:?}", res)));
+
+        let fortune = cl.prepare("SELECT id, message FROM fortune").await.unwrap();
+        let world = cl
+            .prepare("SELECT id, randomnumber FROM world WHERE id=$1")
+            .await
+            .unwrap();
+
+        Ok(PgConnection::create(move |_| PgConnection {
+            cl,
+            fortune,
+            world,
+            rng: thread_rng(),
+        }))
     }
 }
 
@@ -75,32 +52,29 @@ impl Message for RandomWorld {
 }
 
 impl Handler<RandomWorld> for PgConnection {
-    type Result = ResponseFuture<Bytes, io::Error>;
+    type Result = ResponseFuture<Result<Bytes, io::Error>>;
 
     fn handle(&mut self, _: RandomWorld, _: &mut Self::Context) -> Self::Result {
         let random_id = self.rng.gen_range::<i32>(1, 10_001);
-
-        Box::new(
-            self.cl
-                .as_mut()
-                .unwrap()
-                .query(self.world.as_ref().unwrap(), &[&random_id])
-                .into_future()
-                .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e.0)))
-                .map(|(row, _)| {
-                    let row = row.unwrap();
-                    let mut body = BytesMut::with_capacity(33);
-                    serde_json::to_writer(
-                        Writer(&mut body),
-                        &World {
-                            id: row.get(0),
-                            randomnumber: row.get(1),
-                        },
-                    )
-                    .unwrap();
-                    body.freeze()
-                }),
-        )
+        let fut = self.cl.query_one(&self.world, &[&random_id]);
+
+        Box::pin(async move {
+            let row = fut
+                .await
+                .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))?;
+
+            let mut body = BytesMut::with_capacity(33);
+            serde_json::to_writer(
+                Writer(&mut body),
+                &World {
+                    id: row.get(0),
+                    randomnumber: row.get(1),
+                },
+            )
+            .unwrap();
+
+            Ok(body.freeze())
+        })
     }
 }
 
@@ -111,32 +85,28 @@ impl Message for RandomWorlds {
 }
 
 impl Handler<RandomWorlds> for PgConnection {
-    type Result = ResponseFuture<Vec<World>, io::Error>;
+    type Result = ResponseFuture<Result<Vec<World>, io::Error>>;
 
     fn handle(&mut self, msg: RandomWorlds, _: &mut Self::Context) -> Self::Result {
-        let mut worlds = Vec::with_capacity(msg.0 as usize);
+        let worlds = FuturesUnordered::new();
         for _ in 0..msg.0 {
             let w_id: i32 = self.rng.gen_range(1, 10_001);
             worlds.push(
                 self.cl
-                    .as_mut()
-                    .unwrap()
-                    .query(self.world.as_ref().unwrap(), &[&w_id])
-                    .into_future()
-                    .map_err(|e| {
-                        io::Error::new(io::ErrorKind::Other, format!("{:?}", e.0))
-                    })
-                    .map(|(row, _)| {
-                        let row = row.unwrap();
-                        World {
+                    .query_one(&self.world, &[&w_id])
+                    .map(|res| match res {
+                        Err(e) => {
+                            Err(io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
+                        }
+                        Ok(row) => Ok(World {
                             id: row.get(0),
                             randomnumber: row.get(1),
-                        }
+                        }),
                     }),
             );
         }
 
-        Box::new(stream::futures_unordered(worlds).collect())
+        Box::pin(worlds.try_collect())
     }
 }
 
@@ -147,62 +117,53 @@ impl Message for UpdateWorld {
 }
 
 impl Handler<UpdateWorld> for PgConnection {
-    type Result = ResponseActFuture<Self, Vec<World>, io::Error>;
+    type Result = ResponseFuture<Result<Vec<World>, io::Error>>;
 
     fn handle(&mut self, msg: UpdateWorld, _: &mut Self::Context) -> Self::Result {
-        let mut worlds = Vec::with_capacity(msg.0 as usize);
+        let worlds = FuturesUnordered::new();
         for _ in 0..msg.0 {
-            let id = self.rng.gen_range(1, 10_001);
+            let id: i32 = self.rng.gen_range(1, 10_001);
             let w_id: i32 = self.rng.gen_range(1, 10_001);
-            worlds.push(
-                self.cl
-                    .as_mut()
-                    .unwrap()
-                    .query(self.world.as_ref().unwrap(), &[&w_id])
-                    .into_future()
-                    .map_err(|e| {
-                        io::Error::new(io::ErrorKind::Other, format!("{:?}", e.0))
-                    })
-                    .map(move |(row, _)| {
-                        let row = row.unwrap();
+            worlds.push(self.cl.query_one(&self.world, &[&w_id]).map(
+                move |res| match res {
+                    Err(e) => {
+                        Err(io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
+                    }
+                    Ok(row) => {
                         let mut world = World {
                             id: row.get(0),
                             randomnumber: row.get(1),
                         };
                         world.randomnumber = id;
-                        world
-                    }),
-            );
+                        Ok(world)
+                    }
+                },
+            ));
         }
 
-        Box::new(
-            stream::futures_unordered(worlds)
-                .collect()
-                .into_actor(self)
-                .and_then(move |mut worlds, act, _| {
-                    let mut update = String::with_capacity(120 + 6 * msg.0 as usize);
-                    update
-                        .push_str("UPDATE world SET randomnumber = temp.randomnumber FROM (VALUES ");
-
-                    for w in &worlds {
-                        let _ = write!(&mut update, "({}, {}),", w.id, w.randomnumber);
-                    }
-                    worlds.sort_by_key(|w| w.id);
-
-                    update.pop();
-                    update
-                        .push_str(" ORDER BY 1) AS temp(id, randomnumber) WHERE temp.id = world.id");
-
-                    act.cl
-                        .as_mut()
-                        .unwrap()
-                        .simple_query(&update)
-                        .collect()
-                        .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
-                        .into_actor(act)
-                        .map(|_, _, _| worlds)
-                }),
-        )
+        let cl = self.cl.clone();
+        Box::pin(async move {
+            let worlds: Vec<World> = worlds.try_collect().await?;
+
+            let mut update = String::with_capacity(120 + 6 * msg.0 as usize);
+            update.push_str(
+                "UPDATE world SET randomnumber = temp.randomnumber FROM (VALUES ",
+            );
+
+            for w in &worlds {
+                let _ = write!(&mut update, "({}, {}),", w.id, w.randomnumber);
+            }
+            update.pop();
+            update.push_str(
+                " ORDER BY 1) AS temp(id, randomnumber) WHERE temp.id = world.id",
+            );
+
+            cl.simple_query(&update)
+                .await
+                .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))?;
+
+            Ok(worlds)
+        })
     }
 }
 
@@ -213,32 +174,32 @@ impl Message for TellFortune {
 }
 
 impl Handler<TellFortune> for PgConnection {
-    type Result = ResponseFuture<Vec<Fortune>, io::Error>;
+    type Result = ResponseFuture<Result<Vec<Fortune>, io::Error>>;
 
     fn handle(&mut self, _: TellFortune, _: &mut Self::Context) -> Self::Result {
-        let items = vec![Fortune {
+        let mut items = vec![Fortune {
             id: 0,
             message: "Additional fortune added at request time.".to_string(),
         }];
-
-        Box::new(
-            self.cl
-                .as_mut()
-                .unwrap()
-                .query(self.fortune.as_ref().unwrap(), &[])
-                .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
-                .fold(items, move |mut items, row| {
-                    items.push(Fortune {
-                        id: row.get(0),
-                        message: row.get(1),
-                    });
-                    Ok::<_, io::Error>(items)
-                })
-                .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
-                .map(|mut items| {
-                    items.sort_by(|it, next| it.message.cmp(&next.message));
-                    items
-                }),
-        )
+        let fut = self.cl.query_raw(&self.fortune, &[]);
+
+        Box::pin(async move {
+            let mut stream = fut
+                .await
+                .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))?;
+
+            while let Some(row) = stream.next().await {
+                let row = row.map_err(|e| {
+                    io::Error::new(io::ErrorKind::Other, format!("{:?}", e))
+                })?;
+                items.push(Fortune {
+                    id: row.get(0),
+                    message: row.get(1),
+                });
+            }
+
+            items.sort_by(|it, next| it.message.cmp(&next.message));
+            Ok(items)
+        })
     }
 }

frameworks/Rust/actix/src/db_pg_direct.rs

@@ -3,8 +3,8 @@ use std::io;
 
 use actix_http::Error;
 use bytes::{Bytes, BytesMut};
-use futures::future::join_all;
-use futures::{stream, Future, Stream};
+use futures::stream::futures_unordered::FuturesUnordered;
+use futures::{Future, FutureExt, StreamExt, TryStreamExt};
 use rand::{thread_rng, Rng, ThreadRng};
 use tokio_postgres::{connect, Client, NoTls, Statement};
 
@@ -20,159 +20,153 @@ pub struct PgConnection {
 }
 
 impl PgConnection {
-    pub fn connect(db_url: &str) -> impl Future<Item = PgConnection, Error = ()> {
-        let hs = connect(db_url, NoTls);
-
-        hs.map_err(|_| panic!("can not connect to postgresql"))
-            .and_then(|(mut cl, conn)| {
-                actix_rt::spawn(conn.map_err(|e| panic!("{}", e)));
-
-                join_all(vec![
-                    cl.prepare("SELECT id, message FROM fortune"),
-                    cl.prepare("SELECT id, randomnumber FROM world WHERE id=$1"),
-                ])
-                .map_err(|_| ())
-                .map(move |mut st| {
-                    let world = st.pop().unwrap();
-                    let fortune = st.pop().unwrap();
-                    PgConnection {
-                        cl,
-                        fortune,
-                        world,
-                        rng: thread_rng(),
-                    }
-                })
-            })
+    pub async fn connect(db_url: &str) -> PgConnection {
+        let (cl, conn) = connect(db_url, NoTls)
+            .await
+            .expect("can not connect to postgresql");
+        actix_rt::spawn(conn.map(|res| panic!("{:?}", res)));
+
+        let fortune = cl.prepare("SELECT id, message FROM fortune").await.unwrap();
+        let world = cl
+            .prepare("SELECT id, randomnumber FROM world WHERE id=$1")
+            .await
+            .unwrap();
+
+        PgConnection {
+            cl,
+            fortune,
+            world,
+            rng: thread_rng(),
+        }
     }
 }
 
 impl PgConnection {
-    pub fn get_world(&mut self) -> impl Future<Item = Bytes, Error = Error> {
+    pub fn get_world(&mut self) -> impl Future<Output = Result<Bytes, Error>> {
         let random_id = self.rng.gen_range::<i32>(1, 10_001);
+        let fut = self.cl.query_one(&self.world, &[&random_id]);
 
-        self.cl
-            .query(&self.world, &[&random_id])
-            .into_future()
-            .map_err(|e| {
-                Error::from(io::Error::new(io::ErrorKind::Other, format!("{:?}", e.0)))
-            })
-            .map(|(row, _)| {
-                let row = row.unwrap();
-                let mut body = BytesMut::with_capacity(33);
-                serde_json::to_writer(
-                    Writer(&mut body),
-                    &World {
-                        id: row.get(0),
-                        randomnumber: row.get(1),
-                    },
-                )
-                .unwrap();
-                body.freeze()
-            })
+        async move {
+            let row = fut.await.map_err(|e| {
+                Error::from(io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
+            })?;
+
+            let mut body = BytesMut::with_capacity(33);
+            serde_json::to_writer(
+                Writer(&mut body),
+                &World {
+                    id: row.get(0),
+                    randomnumber: row.get(1),
+                },
+            )
+            .unwrap();
+
+            Ok(body.freeze())
+        }
     }
 
     pub fn get_worlds(
         &mut self,
         num: usize,
-    ) -> impl Future<Item = Vec<World>, Error = io::Error> {
-        let mut worlds = Vec::with_capacity(num);
+    ) -> impl Future<Output = Result<Vec<World>, io::Error>> {
+        let worlds = FuturesUnordered::new();
         for _ in 0..num {
             let w_id: i32 = self.rng.gen_range(1, 10_001);
             worlds.push(
                 self.cl
-                    .query(&self.world, &[&w_id])
-                    .into_future()
-                    .map_err(|e| {
-                        io::Error::new(io::ErrorKind::Other, format!("{:?}", e.0))
-                    })
-                    .map(|(row, _)| {
-                        let row = row.unwrap();
-                        World {
+                    .query_one(&self.world, &[&w_id])
+                    .map(|res| match res {
+                        Err(e) => {
+                            Err(io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
+                        }
+                        Ok(row) => Ok(World {
                             id: row.get(0),
                             randomnumber: row.get(1),
-                        }
+                        }),
                     }),
             );
         }
 
-        stream::futures_unordered(worlds).collect()
+        worlds.try_collect()
     }
 
     pub fn update(
         &mut self,
         num: usize,
-    ) -> impl Future<Item = Vec<World>, Error = io::Error> {
-        let mut worlds = Vec::with_capacity(num);
+    ) -> impl Future<Output = Result<Vec<World>, io::Error>> {
+        let worlds = FuturesUnordered::new();
         for _ in 0..num {
             let id: i32 = self.rng.gen_range(1, 10_001);
             let w_id: i32 = self.rng.gen_range(1, 10_001);
-            worlds.push(
-                self.cl
-                    .query(&self.world, &[&w_id])
-                    .into_future()
-                    .map_err(|e| {
-                        io::Error::new(io::ErrorKind::Other, format!("{:?}", e.0))
-                    })
-                    .map(move |(row, _)| {
-                        let row = row.unwrap();
+            worlds.push(self.cl.query_one(&self.world, &[&w_id]).map(
+                move |res| match res {
+                    Err(e) => {
+                        Err(io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
+                    }
+                    Ok(row) => {
                         let mut world = World {
                             id: row.get(0),
                             randomnumber: row.get(1),
                         };
                         world.randomnumber = id;
-                        world
-                    }),
-            );
+                        Ok(world)
+                    }
+                },
+            ));
         }
 
-        let mut cl = self.cl.clone();
-        stream::futures_unordered(worlds)
-            .collect()
-            .and_then(move |worlds| {
-                let mut update = String::with_capacity(120 + 6 * num as usize);
-                update.push_str(
-                    "UPDATE world SET randomnumber = temp.randomnumber FROM (VALUES ",
-                );
-
-                for w in &worlds {
-                    let _ = write!(&mut update, "({}, {}),", w.id, w.randomnumber);
-                }
-                update.pop();
-                update.push_str(
-                    " ORDER BY 1) AS temp(id, randomnumber) WHERE temp.id = world.id",
-                );
-
-                cl.simple_query(&update)
-                    .collect()
-                    .map_err(|e| {
-                        io::Error::new(io::ErrorKind::Other, format!("{:?}", e))
-                    })
-                    .map(|_| worlds)
-            })
+        let cl = self.cl.clone();
+        async move {
+            let worlds: Vec<World> = worlds.try_collect().await?;
+
+            let mut update = String::with_capacity(120 + 6 * num as usize);
+            update.push_str(
+                "UPDATE world SET randomnumber = temp.randomnumber FROM (VALUES ",
+            );
+
+            for w in &worlds {
+                let _ = write!(&mut update, "({}, {}),", w.id, w.randomnumber);
+            }
+            update.pop();
+            update.push_str(
+                " ORDER BY 1) AS temp(id, randomnumber) WHERE temp.id = world.id",
+            );
+
+            cl.simple_query(&update)
+                .await
+                .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))?;
+
+            Ok(worlds)
+        }
     }
 
     pub fn tell_fortune(
         &mut self,
-    ) -> impl Future<Item = Vec<Fortune>, Error = io::Error> {
-        let items = vec![Fortune {
+    ) -> impl Future<Output = Result<Vec<Fortune>, io::Error>> {
+        let mut items = vec![Fortune {
             id: 0,
             message: "Additional fortune added at request time.".to_string(),
         }];
 
-        self.cl
-            .query(&self.fortune, &[])
-            .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
-            .fold(items, move |mut items, row| {
+        let fut = self.cl.query_raw(&self.fortune, &[]);
+
+        async move {
+            let mut stream = fut
+                .await
+                .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))?;
+
+            while let Some(row) = stream.next().await {
+                let row = row.map_err(|e| {
+                    io::Error::new(io::ErrorKind::Other, format!("{:?}", e))
+                })?;
                 items.push(Fortune {
                     id: row.get(0),
                     message: row.get(1),
                 });
-                Ok::<_, io::Error>(items)
-            })
-            .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{:?}", e)))
-            .map(|mut items| {
-                items.sort_by(|it, next| it.message.cmp(&next.message));
-                items
-            })
+            }
+
+            items.sort_by(|it, next| it.message.cmp(&next.message));
+            Ok(items)
+        }
     }
 }

frameworks/Rust/actix/src/main.rs

@@ -5,8 +5,8 @@ static ALLOC: jemallocator::Jemalloc = jemallocator::Jemalloc;
 extern crate serde_derive;
 
 use actix_http::{HttpService, KeepAlive};
-use actix_server::Server;
-use actix_web::dev::Body;
+use actix_service::map_config;
+use actix_web::dev::{AppConfig, Body, Server};
 use actix_web::http::header::{CONTENT_TYPE, SERVER};
 use actix_web::http::{HeaderValue, StatusCode};
 use actix_web::{web, App, HttpResponse};
@@ -15,7 +15,7 @@ use bytes::{Bytes, BytesMut};
 mod utils;
 use utils::{Message, Writer, SIZE};
 
-fn json() -> HttpResponse {
+async fn json() -> HttpResponse {
     let message = Message {
         message: "Hello, World!",
     };
@@ -30,7 +30,7 @@ fn json() -> HttpResponse {
     res
 }
 
-fn plaintext() -> HttpResponse {
+async fn plaintext() -> HttpResponse {
     let mut res = HttpResponse::with_body(
         StatusCode::OK,
         Body::Bytes(Bytes::from_static(b"Hello, World!")),
@@ -42,19 +42,24 @@ fn plaintext() -> HttpResponse {
     res
 }
 
-fn main() -> std::io::Result<()> {
-    let sys = actix_rt::System::new("techempower");
+#[actix_rt::main]
+async fn main() -> std::io::Result<()> {
+    println!("Started http server: 127.0.0.1:8080");
 
     // start http server
     Server::build()
         .backlog(1024)
         .bind("techempower", "0.0.0.0:8080", || {
-            HttpService::build().keep_alive(KeepAlive::Os).h1(App::new()
-                .service(web::resource("/json").to(json))
-                .service(web::resource("/plaintext").to(plaintext)))
+            HttpService::build()
+                .keep_alive(KeepAlive::Os)
+                .h1(map_config(
+                    App::new()
+                        .service(web::resource("/json").to(json))
+                        .service(web::resource("/plaintext").to(plaintext)),
+                    |_| AppConfig::default(),
+                ))
+                .tcp()
         })?
-        .start();
-
-    println!("Started http server: 127.0.0.1:8080");
-    sys.run()
+        .start()
+        .await
 }

frameworks/Rust/actix/src/main_diesel.rs

@@ -7,12 +7,9 @@ extern crate serde_derive;
 extern crate diesel;
 
 use actix::prelude::*;
-use actix_http::{HttpService, KeepAlive};
-use actix_server::Server;
-use actix_web::{http, web, App, Error, HttpRequest, HttpResponse};
+use actix_web::{http, web, App, Error, HttpRequest, HttpResponse, HttpServer};
 use askama::Template;
 use bytes::BytesMut;
-use futures::Future;
 
 mod db;
 mod models;
@@ -20,68 +17,63 @@ mod schema;
 mod utils;
 use utils::Writer;
 
-fn world_row(
-    db: web::Data<Addr<db::DbExecutor>>,
-) -> impl Future<Item = HttpResponse, Error = Error> {
-    db.send(db::RandomWorld)
-        .from_err()
-        .and_then(move |res| match res {
-            Ok(row) => {
-                let mut body = BytesMut::with_capacity(33);
-                serde_json::to_writer(Writer(&mut body), &row).unwrap();
-                Ok(HttpResponse::Ok()
-                    .header(http::header::SERVER, "Actix")
-                    .header(http::header::CONTENT_TYPE, "application/json")
-                    .body(body))
-            }
-            Err(_) => Ok(HttpResponse::InternalServerError().into()),
-        })
+async fn world_row(db: web::Data<Addr<db::DbExecutor>>) -> Result<HttpResponse, Error> {
+    let res = db.send(db::RandomWorld).await?;
+
+    match res {
+        Ok(row) => {
+            let mut body = BytesMut::with_capacity(33);
+            serde_json::to_writer(Writer(&mut body), &row).unwrap();
+            Ok(HttpResponse::Ok()
+                .header(http::header::SERVER, "Actix")
+                .header(http::header::CONTENT_TYPE, "application/json")
+                .body(body))
+        }
+        Err(_) => Ok(HttpResponse::InternalServerError().into()),
+    }
 }
 
-fn queries(
+async fn queries(
     req: HttpRequest,
     db: web::Data<Addr<db::DbExecutor>>,
-) -> impl Future<Item = HttpResponse, Error = Error> {
+) -> Result<HttpResponse, Error> {
     // get queries parameter
     let q = utils::get_query_param(req.query_string());
 
     // run sql queries
-    db.send(db::RandomWorlds(q))
-        .from_err()
-        .and_then(move |res| {
-            if let Ok(worlds) = res {
-                let mut body = BytesMut::with_capacity(35 * worlds.len());
-                serde_json::to_writer(Writer(&mut body), &worlds).unwrap();
-                Ok(HttpResponse::Ok()
-                    .header(http::header::SERVER, "Actix")
-                    .header(http::header::CONTENT_TYPE, "application/json")
-                    .body(body))
-            } else {
-                Ok(HttpResponse::InternalServerError().into())
-            }
-        })
+    let res = db.send(db::RandomWorlds(q)).await?;
+    if let Ok(worlds) = res {
+        let mut body = BytesMut::with_capacity(35 * worlds.len());
+        serde_json::to_writer(Writer(&mut body), &worlds).unwrap();
+        Ok(HttpResponse::Ok()
+            .header(http::header::SERVER, "Actix")
+            .header(http::header::CONTENT_TYPE, "application/json")
+            .body(body))
+    } else {
+        Ok(HttpResponse::InternalServerError().into())
+    }
 }
 
-fn updates(
+async fn updates(
     req: HttpRequest,
     db: web::Data<Addr<db::DbExecutor>>,
-) -> impl Future<Item = HttpResponse, Error = Error> {
+) -> Result<HttpResponse, Error> {
     // get queries parameter
     let q = utils::get_query_param(req.query_string());
 
     // update worlds
-    db.send(db::UpdateWorld(q)).from_err().and_then(move |res| {
-        if let Ok(worlds) = res {
-            let mut body = BytesMut::with_capacity(35 * worlds.len());
-            serde_json::to_writer(Writer(&mut body), &worlds).unwrap();
-            Ok(HttpResponse::Ok()
-                .header(http::header::SERVER, "Actix")
-                .header(http::header::CONTENT_TYPE, "application/json")
-                .body(body))
-        } else {
-            Ok(HttpResponse::InternalServerError().into())
-        }
-    })
+    let res = db.send(db::UpdateWorld(q)).await?;
+
+    if let Ok(worlds) = res {
+        let mut body = BytesMut::with_capacity(35 * worlds.len());
+        serde_json::to_writer(Writer(&mut body), &worlds).unwrap();
+        Ok(HttpResponse::Ok()
+            .header(http::header::SERVER, "Actix")
+            .header(http::header::CONTENT_TYPE, "application/json")
+            .body(body))
+    } else {
+        Ok(HttpResponse::InternalServerError().into())
+    }
 }
 
 #[derive(Template)]
@@ -90,27 +82,26 @@ struct FortuneTemplate<'a> {
     items: &'a Vec<models::Fortune>,
 }
 
-fn fortune(
-    db: web::Data<Addr<db::DbExecutor>>,
-) -> impl Future<Item = HttpResponse, Error = Error> {
-    db.send(db::TellFortune)
-        .from_err()
-        .and_then(move |res| match res {
-            Ok(rows) => {
-                let tmpl = FortuneTemplate { items: &rows };
-                let res = tmpl.render().unwrap();
-
-                Ok(HttpResponse::Ok()
-                    .header(http::header::SERVER, "Actix")
-                    .header(http::header::CONTENT_TYPE, "text/html; charset=utf-8")
-                    .body(res))
-            }
-            Err(_) => Ok(HttpResponse::InternalServerError().into()),
-        })
+async fn fortune(db: web::Data<Addr<db::DbExecutor>>) -> Result<HttpResponse, Error> {
+    let res = db.send(db::TellFortune).await?;
+    match res {
+        Ok(rows) => {
+            let tmpl = FortuneTemplate { items: &rows };
+            let res = tmpl.render().unwrap();
+
+            Ok(HttpResponse::Ok()
+                .header(http::header::SERVER, "Actix")
+                .header(http::header::CONTENT_TYPE, "text/html; charset=utf-8")
+                .body(res))
+        }
+        Err(_) => Ok(HttpResponse::InternalServerError().into()),
+    }
 }
 
-fn main() -> std::io::Result<()> {
-    let sys = actix_rt::System::builder().stop_on_panic(false).build();
+#[actix_rt::main]
+async fn main() -> std::io::Result<()> {
+    println!("Starting http server: 127.0.0.1:8080");
+
     let db_url = "postgres://benchmarkdbuser:benchmarkdbpass@tfb-database/hello_world";
 
     // Start db executor actors
@@ -118,18 +109,15 @@ fn main() -> std::io::Result<()> {
         SyncArbiter::start(num_cpus::get() * 3, move || db::DbExecutor::new(db_url));
 
     // start http server
-    Server::build()
-        .backlog(1024)
-        .bind("techempower", "0.0.0.0:8080", move || {
-            HttpService::build().keep_alive(KeepAlive::Os).h1(App::new()
-                .data(addr.clone())
-                .service(web::resource("/db").to_async(world_row))
-                .service(web::resource("/fortune").to_async(fortune))
-                .service(web::resource("/queries").to_async(queries))
-                .service(web::resource("/updates").to_async(updates)))
-        })?
-        .start();
-
-    println!("Started http server: 127.0.0.1:8080");
-    sys.run()
+    HttpServer::new(move || {
+        App::new()
+            .data(addr.clone())
+            .service(web::resource("/db").to(world_row))
+            .service(web::resource("/fortune").to(fortune))
+            .service(web::resource("/queries").to(queries))
+            .service(web::resource("/updates").to(updates))
+    })
+    .bind("0.0.0.0:8080")?
+    .run()
+    .await
 }

frameworks/Rust/actix/src/main_pg.rs

@@ -10,11 +10,11 @@ use std::io::Write;
 
 use actix::prelude::*;
 use actix_http::{HttpService, KeepAlive};
-use actix_server::Server;
+use actix_service::map_config;
+use actix_web::dev::{AppConfig, Body, Server};
 use actix_web::http::{header::CONTENT_TYPE, header::SERVER, HeaderValue, StatusCode};
-use actix_web::{dev::Body, web, App, Error, HttpRequest, HttpResponse};
+use actix_web::{web, App, Error, HttpRequest, HttpResponse};
 use bytes::BytesMut;
-use futures::Future;
 
 mod db_pg;
 mod models;
@@ -22,120 +22,119 @@ mod utils;
 use crate::db_pg::{PgConnection, RandomWorld, RandomWorlds, TellFortune, UpdateWorld};
 use crate::utils::{FortunesYarteTemplate, Writer};
 
-fn world_row(
-    db: web::Data<Addr<PgConnection>>,
-) -> impl Future<Item = HttpResponse, Error = Error> {
-    db.send(RandomWorld)
-        .from_err()
-        .and_then(move |res| match res {
-            Ok(body) => {
-                let mut res = HttpResponse::with_body(StatusCode::OK, Body::Bytes(body));
-                res.headers_mut()
-                    .insert(SERVER, HeaderValue::from_static("Actix"));
-                res.headers_mut()
-                    .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
-                Ok(res)
-            }
-            Err(_) => Ok(HttpResponse::InternalServerError().into()),
-        })
+async fn world_row(db: web::Data<Addr<PgConnection>>) -> Result<HttpResponse, Error> {
+    let res = db.send(RandomWorld).await?;
+    match res {
+        Ok(body) => {
+            let mut res = HttpResponse::with_body(StatusCode::OK, Body::Bytes(body));
+            res.headers_mut()
+                .insert(SERVER, HeaderValue::from_static("Actix"));
+            res.headers_mut()
+                .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+            Ok(res)
+        }
+        Err(_) => Ok(HttpResponse::InternalServerError().into()),
+    }
 }
 
-fn queries(
+async fn queries(
     req: HttpRequest,
     db: web::Data<Addr<PgConnection>>,
-) -> impl Future<Item = HttpResponse, Error = Error> {
+) -> Result<HttpResponse, Error> {
     // get queries parameter
     let q = utils::get_query_param(req.query_string());
 
     // run sql queries
-    db.send(RandomWorlds(q)).from_err().and_then(move |res| {
-        if let Ok(worlds) = res {
-            let mut body = BytesMut::with_capacity(35 * worlds.len());
-            serde_json::to_writer(Writer(&mut body), &worlds).unwrap();
-            let mut res =
-                HttpResponse::with_body(StatusCode::OK, Body::Bytes(body.freeze()));
-            res.headers_mut()
-                .insert(SERVER, HeaderValue::from_static("Actix"));
-            res.headers_mut()
-                .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
-            Ok(res)
-        } else {
-            Ok(HttpResponse::InternalServerError().into())
-        }
-    })
+    let res = db.send(RandomWorlds(q)).await?;
+    if let Ok(worlds) = res {
+        let mut body = BytesMut::with_capacity(35 * worlds.len());
+        serde_json::to_writer(Writer(&mut body), &worlds).unwrap();
+        let mut res =
+            HttpResponse::with_body(StatusCode::OK, Body::Bytes(body.freeze()));
+        res.headers_mut()
+            .insert(SERVER, HeaderValue::from_static("Actix"));
+        res.headers_mut()
+            .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        Ok(res)
+    } else {
+        Ok(HttpResponse::InternalServerError().into())
+    }
 }
 
-fn updates(
+async fn updates(
     req: HttpRequest,
     db: web::Data<Addr<PgConnection>>,
-) -> impl Future<Item = HttpResponse, Error = Error> {
+) -> Result<HttpResponse, Error> {
     // get queries parameter
     let q = utils::get_query_param(req.query_string());
 
     // update db
-    db.send(UpdateWorld(q)).from_err().and_then(move |res| {
-        if let Ok(worlds) = res {
-            let mut body = BytesMut::with_capacity(35 * worlds.len());
-            serde_json::to_writer(Writer(&mut body), &worlds).unwrap();
-            let mut res =
-                HttpResponse::with_body(StatusCode::OK, Body::Bytes(body.freeze()));
+    let res = db.send(UpdateWorld(q)).await?;
+    if let Ok(worlds) = res {
+        let mut body = BytesMut::with_capacity(35 * worlds.len());
+        serde_json::to_writer(Writer(&mut body), &worlds).unwrap();
+        let mut res =
+            HttpResponse::with_body(StatusCode::OK, Body::Bytes(body.freeze()));
+        res.headers_mut()
+            .insert(SERVER, HeaderValue::from_static("Actix"));
+        res.headers_mut()
+            .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        Ok(res)
+    } else {
+        Ok(HttpResponse::InternalServerError().into())
+    }
+}
+
+async fn fortune(db: web::Data<Addr<PgConnection>>) -> Result<HttpResponse, Error> {
+    let res = db.send(TellFortune).await?;
+
+    match res {
+        Ok(fortunes) => {
+            let mut body = BytesMut::with_capacity(2048);
+            let mut writer = Writer(&mut body);
+            let _ = write!(writer, "{}", FortunesYarteTemplate { fortunes });
+
+            let mut res = HttpResponse::with_body(
+                StatusCode::OK,
+                Body::Bytes(body.freeze().into()),
+            );
             res.headers_mut()
                 .insert(SERVER, HeaderValue::from_static("Actix"));
-            res.headers_mut()
-                .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+            res.headers_mut().insert(
+                CONTENT_TYPE,
+                HeaderValue::from_static("text/html; charset=utf-8"),
+            );
             Ok(res)
-        } else {
-            Ok(HttpResponse::InternalServerError().into())
         }
-    })
+        Err(_) => Ok(HttpResponse::InternalServerError().into()),
+    }
 }
 
-fn fortune(
-    db: web::Data<Addr<PgConnection>>,
-) -> impl Future<Item = HttpResponse, Error = Error> {
-    db.send(TellFortune)
-        .from_err()
-        .and_then(move |res| match res {
-            Ok(fortunes) => {
-                let mut body = BytesMut::with_capacity(2048);
-                let mut writer = Writer(&mut body);
-                let _ = write!(writer, "{}", FortunesYarteTemplate { fortunes });
-
-                let mut res = HttpResponse::with_body(
-                    StatusCode::OK,
-                    Body::Bytes(body.freeze().into()),
-                );
-                res.headers_mut()
-                    .insert(SERVER, HeaderValue::from_static("Actix"));
-                res.headers_mut().insert(
-                    CONTENT_TYPE,
-                    HeaderValue::from_static("text/html; charset=utf-8"),
-                );
-                Ok(res)
-            }
-            Err(_) => Ok(HttpResponse::InternalServerError().into()),
-        })
-}
+#[actix_rt::main]
+async fn main() -> std::io::Result<()> {
+    println!("Started http server: 127.0.0.1:8080");
 
-fn main() -> std::io::Result<()> {
-    let sys = actix_rt::System::builder().stop_on_panic(false).build();
     const DB_URL: &str =
         "postgres://benchmarkdbuser:benchmarkdbpass@tfb-database/hello_world";
 
     // start http server
     Server::build()
         .backlog(1024)
+        .workers(1)
         .bind("techempower", "0.0.0.0:8080", || {
-            let addr = PgConnection::connect(DB_URL);
-            HttpService::build().keep_alive(KeepAlive::Os).h1(App::new()
-                .data(addr)
-                .service(web::resource("/db").to_async(world_row))
-                .service(web::resource("/queries").to_async(queries))
-                .service(web::resource("/fortune").to_async(fortune))
-                .service(web::resource("/updates").to_async(updates)))
+            HttpService::build()
+                .keep_alive(KeepAlive::Os)
+                .h1(map_config(
+                    App::new()
+                        .data_factory(|| PgConnection::connect(DB_URL))
+                        .service(web::resource("/db").to(world_row))
+                        .service(web::resource("/queries").to(queries))
+                        .service(web::resource("/fortune").to(fortune))
+                        .service(web::resource("/updates").to(updates)),
+                    |_| AppConfig::default(),
+                ))
+                .tcp()
         })?
-        .start();
-
-    println!("Started http server: 127.0.0.1:8080");
-    sys.run()
+        .start()
+        .await
 }

frameworks/Rust/actix/src/main_platform.rs

@@ -6,17 +6,19 @@ extern crate serde_derive;
 #[macro_use]
 extern crate diesel;
 
+use std::future::Future;
 use std::io::Write;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 use actix_http::body::Body;
 use actix_http::http::header::{CONTENT_TYPE, SERVER};
 use actix_http::http::{HeaderValue, StatusCode};
 use actix_http::{Error, HttpService, KeepAlive, Request, Response};
-use actix_server::{Server, ServerConfig};
-use actix_service::{NewService, Service};
+use actix_server::Server;
+use actix_service::{Service, ServiceFactory};
 use bytes::BytesMut;
 use futures::future::ok;
-use futures::{Async, Future, Poll};
 use serde_json::to_writer;
 
 mod db_pg_direct;
@@ -37,11 +39,11 @@ impl Service for App {
     type Request = Request;
     type Response = Response;
     type Error = Error;
-    type Future = Box<dyn Future<Item = Response, Error = Error>>;
+    type Future = Pin<Box<dyn Future<Output = Result<Response, Error>>>>;
 
     #[inline]
-    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
-        Ok(Async::Ready(()))
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
     }
 
     fn call(&mut self, req: Request) -> Self::Future {
@@ -50,20 +52,24 @@ impl Service for App {
             "/db" => {
                 let h_srv = self.hdr_srv.clone();
                 let h_ct = self.hdr_ctjson.clone();
+                let fut = self.db.get_world();
 
-                Box::new(self.db.get_world().map(move |body| {
+                Box::pin(async move {
+                    let body = fut.await?;
                     let mut res = Response::with_body(StatusCode::OK, Body::Bytes(body));
                     let hdrs = res.headers_mut();
                     hdrs.insert(SERVER, h_srv);
                     hdrs.insert(CONTENT_TYPE, h_ct);
-                    res
-                }))
+                    Ok(res)
+                })
             }
             "/fortune" => {
                 let h_srv = self.hdr_srv.clone();
                 let h_ct = self.hdr_cthtml.clone();
+                let fut = self.db.tell_fortune();
 
-                Box::new(self.db.tell_fortune().from_err().map(move |fortunes| {
+                Box::pin(async move {
+                    let fortunes = fut.await?;
                     let mut body = BytesMut::with_capacity(2048);
                     let mut writer = Writer(&mut body);
                     let _ = write!(writer, "{}", FortunesTemplate { fortunes });
@@ -72,15 +78,17 @@ impl Service for App {
                     let hdrs = res.headers_mut();
                     hdrs.insert(SERVER, h_srv);
                     hdrs.insert(CONTENT_TYPE, h_ct);
-                    res
-                }))
+                    Ok(res)
+                })
             }
             "/queries" => {
                 let q = utils::get_query_param(req.uri().query().unwrap_or("")) as usize;
                 let h_srv = self.hdr_srv.clone();
                 let h_ct = self.hdr_ctjson.clone();
+                let fut = self.db.get_worlds(q);
 
-                Box::new(self.db.get_worlds(q).from_err().map(move |worlds| {
+                Box::pin(async move {
+                    let worlds = fut.await?;
                     let mut body = BytesMut::with_capacity(35 * worlds.len());
                     to_writer(Writer(&mut body), &worlds).unwrap();
                     let mut res =
@@ -88,15 +96,17 @@ impl Service for App {
                     let hdrs = res.headers_mut();
                     hdrs.insert(SERVER, h_srv);
                     hdrs.insert(CONTENT_TYPE, h_ct);
-                    res
-                }))
+                    Ok(res)
+                })
             }
             "/updates" => {
                 let q = utils::get_query_param(req.uri().query().unwrap_or("")) as usize;
                 let h_srv = self.hdr_srv.clone();
                 let h_ct = self.hdr_ctjson.clone();
+                let fut = self.db.update(q);
 
-                Box::new(self.db.update(q).from_err().map(move |worlds| {
+                Box::pin(async move {
+                    let worlds = fut.await?;
                     let mut body = BytesMut::with_capacity(35 * worlds.len());
                     to_writer(Writer(&mut body), &worlds).unwrap();
                     let mut res =
@@ -104,10 +114,10 @@ impl Service for App {
                     let hdrs = res.headers_mut();
                     hdrs.insert(SERVER, h_srv);
                     hdrs.insert(CONTENT_TYPE, h_ct);
-                    res
-                }))
+                    Ok(res)
+                })
             }
-            _ => Box::new(ok(Response::new(http::StatusCode::NOT_FOUND))),
+            _ => Box::pin(ok(Response::new(http::StatusCode::NOT_FOUND))),
         }
     }
 }
@@ -115,25 +125,28 @@ impl Service for App {
 #[derive(Clone)]
 struct AppFactory;
 
-impl NewService for AppFactory {
-    type Config = ServerConfig;
+impl ServiceFactory for AppFactory {
+    type Config = ();
     type Request = Request;
     type Response = Response;
     type Error = Error;
     type Service = App;
     type InitError = ();
-    type Future = Box<dyn Future<Item = Self::Service, Error = Self::InitError>>;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Service, Self::InitError>>>>;
 
-    fn new_service(&self, _: &ServerConfig) -> Self::Future {
+    fn new_service(&self, _: ()) -> Self::Future {
         const DB_URL: &str =
             "postgres://benchmarkdbuser:benchmarkdbpass@tfb-database/hello_world";
 
-        Box::new(PgConnection::connect(DB_URL).map(|db| App {
-            db,
-            hdr_srv: HeaderValue::from_static("Actix"),
-            hdr_ctjson: HeaderValue::from_static("application/json"),
-            hdr_cthtml: HeaderValue::from_static("text/html; charset=utf-8"),
-        }))
+        Box::pin(async move {
+            let db = PgConnection::connect(DB_URL).await;
+            Ok(App {
+                db,
+                hdr_srv: HeaderValue::from_static("Actix"),
+                hdr_ctjson: HeaderValue::from_static("application/json"),
+                hdr_cthtml: HeaderValue::from_static("text/html; charset=utf-8"),
+            })
+        })
     }
 }
 
@@ -146,6 +159,7 @@ fn main() -> std::io::Result<()> {
             HttpService::build()
                 .keep_alive(KeepAlive::Os)
                 .h1(AppFactory)
+                .tcp()
         })?
         .start();
 

frameworks/Rust/actix/src/main_raw.rs

@@ -6,40 +6,38 @@ extern crate serde_derive;
 #[macro_use]
 extern crate diesel;
 
+use std::future::Future;
 use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 use actix_codec::{AsyncRead, AsyncWrite, Decoder};
-use actix_http::h1;
-use actix_http::Request;
-use actix_server::{Io, Server};
-use actix_service::service_fn;
-use bytes::{BufMut, BytesMut};
-use futures::{Async, Future, Poll};
+use actix_http::{h1, Request};
+use actix_rt::net::TcpStream;
+use actix_server::Server;
+use actix_service::fn_service;
+use bytes::{Buf, BufMut, BytesMut};
 use serde_json::to_writer;
-use tokio_tcp::TcpStream;
 
 mod models;
 mod utils;
 
 use crate::utils::{Message, Writer};
 
-const HTTPOK: &[u8] = b"HTTP/1.1 200 OK\r\n";
+const JSON: &[u8] = b"HTTP/1.1 200 OK\r\nServer: Actix\r\nContent-Type: application/json\r\nContent-Length: 27\r\n";
+const PLAIN: &[u8] = b"HTTP/1.1 200 OK\r\nServer: Actix\r\nContent-Type: text/plain\r\nContent-Length: 13\r\n";
 const HTTPNFOUND: &[u8] = b"HTTP/1.1 400 OK\r\n";
 const HDR_SERVER: &[u8] = b"Server: Actix\r\n";
-const HDR_CTPLAIN: &[u8] = b"Content-Type: text/plain\r\n";
-const HDR_CTJSON: &[u8] = b"Content-Type: application/json\r\n";
-const HDR_PL_LEN: &[u8] = b"Content-Length: 13\r\n";
-const HDR_JS_LEN: &[u8] = b"Content-Length: 27\r\n";
 const BODY: &[u8] = b"Hello, World!";
 
-struct App<T> {
-    io: T,
+struct App {
+    io: TcpStream,
     read_buf: BytesMut,
     write_buf: BytesMut,
     codec: h1::Codec,
 }
 
-impl<T: AsyncRead + AsyncWrite> App<T> {
+impl App {
     fn handle_request(&mut self, req: Request) {
         let path = req.path();
         match path {
@@ -47,18 +45,12 @@ impl<T: AsyncRead + AsyncWrite> App<T> {
                 let message = Message {
                     message: "Hello, World!",
                 };
-                self.write_buf.put_slice(HTTPOK);
-                self.write_buf.put_slice(HDR_SERVER);
-                self.write_buf.put_slice(HDR_CTJSON);
-                self.write_buf.put_slice(HDR_JS_LEN);
+                self.write_buf.put_slice(JSON);
                 self.codec.config().set_date(&mut self.write_buf);
                 to_writer(Writer(&mut self.write_buf), &message).unwrap();
             }
             "/plaintext" => {
-                self.write_buf.put_slice(HTTPOK);
-                self.write_buf.put_slice(HDR_SERVER);
-                self.write_buf.put_slice(HDR_CTPLAIN);
-                self.write_buf.put_slice(HDR_PL_LEN);
+                self.write_buf.put_slice(PLAIN);
                 self.codec.config().set_date(&mut self.write_buf);
                 self.write_buf.put_slice(BODY);
             }
@@ -70,87 +62,88 @@ impl<T: AsyncRead + AsyncWrite> App<T> {
     }
 }
 
-impl<T: AsyncRead + AsyncWrite> Future for App<T> {
-    type Item = ();
-    type Error = ();
+impl Future for App {
+    type Output = Result<(), ()>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
 
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         loop {
-            if self.read_buf.remaining_mut() < 4096 {
-                self.read_buf.reserve(32_768);
+            if this.read_buf.capacity() - this.read_buf.len() < 4096 {
+                this.read_buf.reserve(32_768);
             }
-            let read = unsafe { self.io.read(self.read_buf.bytes_mut()) };
+            let read = Pin::new(&mut this.io).poll_read_buf(cx, &mut this.read_buf);
             match read {
-                Ok(0) => return Ok(Async::Ready(())),
-                Ok(n) => unsafe { self.read_buf.advance_mut(n) },
-                Err(e) => {
-                    if e.kind() == io::ErrorKind::WouldBlock {
-                        break;
-                    } else {
-                        return Err(());
+                Poll::Pending => break,
+                Poll::Ready(Ok(n)) => {
+                    if n == 0 {
+                        return Poll::Ready(Ok(()));
                     }
                 }
+                Poll::Ready(Err(_)) => return Poll::Ready(Err(())),
             }
         }
 
-        if self.write_buf.remaining_mut() < 8192 {
-            self.write_buf.reserve(32_768);
+        if this.write_buf.capacity() - this.write_buf.len() < 8192 {
+            this.write_buf.reserve(32_768);
         }
 
         loop {
-            match self.codec.decode(&mut self.read_buf) {
-                Ok(Some(h1::Message::Item(req))) => self.handle_request(req),
+            match this.codec.decode(&mut this.read_buf) {
+                Ok(Some(h1::Message::Item(req))) => this.handle_request(req),
                 Ok(None) => break,
-                _ => return Err(()),
+                _ => return Poll::Ready(Err(())),
             }
         }
 
-        if !self.write_buf.is_empty() {
-            let len = self.write_buf.len();
+        if !this.write_buf.is_empty() {
+            let len = this.write_buf.len();
             let mut written = 0;
             while written < len {
-                match self.io.write(&self.write_buf[written..]) {
-                    Ok(0) => return Ok(Async::Ready(())),
-                    Ok(n) => {
-                        written += n;
-                    }
-                    Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+                match Pin::new(&mut this.io).poll_write(cx, &this.write_buf[written..]) {
+                    Poll::Pending => {
                         if written > 0 {
-                            let _ = self.write_buf.split_to(written);
+                            this.write_buf.advance(written);
                         }
                         break;
                     }
-                    Err(_) => return Err(()),
+                    Poll::Ready(Ok(n)) => {
+                        if n == 0 {
+                            return Poll::Ready(Ok(()));
+                        } else {
+                            written += n;
+                        }
+                    }
+                    Poll::Ready(Err(_)) => return Poll::Ready(Err(())),
                 }
             }
             if written > 0 {
                 if written == len {
-                    unsafe { self.write_buf.set_len(0) }
+                    unsafe { this.write_buf.set_len(0) }
                 } else {
-                    let _ = self.write_buf.split_to(written);
+                    this.write_buf.advance(written);
                 }
             }
         }
-        Ok(Async::NotReady)
+        Poll::Pending
     }
 }
 
-fn main() -> io::Result<()> {
-    let sys = actix_rt::System::builder().stop_on_panic(false).build();
+#[actix_rt::main]
+async fn main() -> io::Result<()> {
+    println!("Started http server: 127.0.0.1:8080");
 
     // start http server
     Server::build()
         .backlog(1024)
         .bind("techempower", "0.0.0.0:8080", || {
-            service_fn(|io: Io<TcpStream, _>| App {
-                io: io.into_parts().0,
+            fn_service(|io: TcpStream| App {
+                io,
                 read_buf: BytesMut::with_capacity(32_768),
                 write_buf: BytesMut::with_capacity(32_768),
                 codec: h1::Codec::default(),
             })
         })?
-        .start();
-
-    println!("Started http server: 127.0.0.1:8080");
-    sys.run()
+        .start()
+        .await
 }

frameworks/Rust/actix/src/utils.rs

@@ -107,5 +107,5 @@ markup::define! {
 #[derive(Template)]
 #[template(path = "fortune.hbs")]
 pub struct FortunesYarteTemplate {
-    pub fortunes: Vec<Fortune>
+    pub fortunes: Vec<Fortune>,
 }