WIP: aquatic http glommio

Cargo.lock

@@ -113,6 +113,7 @@ dependencies = [
  "rustls",
  "rustls-pemfile",
  "serde",
+ "slab",
  "smartstring",
  "socket2 0.4.2",
 ]

aquatic_http/Cargo.toml

@@ -17,7 +17,7 @@ path = "src/bin/main.rs"
 
 [features]
 default = ["with-mio"]
-with-glommio = ["glommio", "futures-lite", "rustls", "rustls-pemfile"]
+with-glommio = ["glommio", "futures-lite", "rustls", "rustls-pemfile", "slab"]
 with-mio = ["crossbeam-channel", "histogram", "mio", "native-tls", "socket2"]
 
 [dependencies]
@@ -51,6 +51,7 @@ futures-lite = { version = "1", optional = true }
 glommio = { git = "https://github.com/DataDog/glommio.git", rev = "4e6b14772da2f4325271fbcf12d24cf91ed466e5", optional = true }
 rustls = { version = "0.20", optional = true }
 rustls-pemfile = { version = "0.2", optional = true }
+slab = { version = "0.4", optional = true }
 
 [dev-dependencies]
 quickcheck = "1.0"

aquatic_http/src/lib/glommio/network.rs

@@ -3,27 +3,67 @@ use std::rc::Rc;
 use std::sync::Arc;
 
 use aquatic_http_protocol::request::Request;
-use futures_lite::{AsyncReadExt, StreamExt};
+use aquatic_http_protocol::response::Response;
+use futures_lite::{AsyncReadExt, AsyncWriteExt, StreamExt};
 use glommio::prelude::*;
 use glommio::net::{TcpListener, TcpStream};
+use glommio::channels::local_channel::{new_bounded, LocalReceiver, LocalSender};
+use glommio::task::JoinHandle;
 use rustls::{IoState, ServerConnection};
+use slab::Slab;
 
 use crate::config::Config;
 
+struct ConnectionReference {
+    response_sender: LocalSender<Response>,
+    handle: JoinHandle<()>,
+}
+
+struct Connection {
+    response_receiver: LocalReceiver<Response>,
+    tls: ServerConnection,
+    stream: TcpStream,
+    index: usize,
+}
+
 pub async fn run_socket_worker(
     config: Config,
 ) {
-    let tlsConfig = Arc::new(create_tls_config(&config));
+    let tls_config = Arc::new(create_tls_config(&config));
     let config = Rc::new(config);
 
     let listener = TcpListener::bind(config.network.address).expect("bind socket");
 
+    let mut connection_slab: Slab<ConnectionReference> = Slab::new();
+
     let mut incoming = listener.incoming();
 
     while let Some(stream) = incoming.next().await {
         match stream {
             Ok(stream) => {
-                spawn_local(handle_stream(config.clone(), tlsConfig.clone(), stream)).detach();
+                let (response_sender, response_receiver) = new_bounded(1);
+
+                let entry = connection_slab.vacant_entry();
+
+                let conn = Connection {
+                    response_receiver,
+                    tls: ServerConnection::new(tls_config.clone()).unwrap(),
+                    stream,
+                    index: entry.key(),
+                };
+
+                async fn handle_stream(mut conn: Connection) {
+                    conn.handle_stream().await;
+                }
+
+                let handle = spawn_local(handle_stream(conn)).detach();
+
+                let connection_reference = ConnectionReference {
+                    response_sender,
+                    handle,
+                };
+
+                entry.insert(connection_reference);
             },
             Err(err) => {
                 ::log::error!("accept connection: {:?}", err);
@@ -33,34 +73,51 @@ pub async fn run_socket_worker(
     }
 }
 
-async fn handle_stream(
-    config: Rc<Config>,
-    tlsConfig: Arc<rustls::ServerConfig>,
-    mut stream: TcpStream,
-){
+impl Connection {
+    async fn handle_stream(&mut self){
+        loop {
+            while let Some(response) = self.response_receiver.stream().next().await {
+                response.write(&mut self.tls.writer()).unwrap();
+
+                let mut buf = Vec::new();
+                let mut buf = Cursor::new(&mut buf);
+
+                while self.tls.wants_write() {
+                    self.tls.write_tls(&mut buf).unwrap();
+                }
+
+                self.stream.write_all(&buf.into_inner()).await.unwrap();
+            }
+        }
+    }
+
+    async fn handle_stream_handshake(&mut self) {
         let mut buf = [0u8; 1024];
-    let mut conn = ServerConnection::new(tlsConfig).unwrap();
 
         loop {
-        match stream.read(&mut buf).await {
+            match self.stream.read(&mut buf).await {
                 Ok(ciphertext_bytes_read) => {
                     let mut cursor = Cursor::new(&buf[..ciphertext_bytes_read]);
 
-                match conn.read_tls(&mut cursor) {
+                    match self.tls.read_tls(&mut cursor) {
                         Ok(plaintext_bytes_read) => {
-                        match conn.process_new_packets() {
+                            match self.tls.process_new_packets() {
                                 Ok(_) => {
                                     if ciphertext_bytes_read == 0 && plaintext_bytes_read == 0 {
                                         let mut request_bytes = Vec::new();
 
-                                    conn.reader().read_to_end(&mut request_bytes);
+                                        self.tls.reader().read_to_end(&mut request_bytes);
 
                                         match Request::from_bytes(&request_bytes[..]) {
                                             Ok(request) => {
-
+                                                ::log::info!("request read: {:?}", request);
                                             },
                                             Err(err) => {
-                                            // TODO: return error response, close connection
+                                                // TODO: send error response, close connection
+
+                                                ::log::info!("Request::from_bytes: {:?}", err);
+
+                                                break
                                             }
                                         }
                                     }
@@ -86,6 +143,9 @@ async fn handle_stream(
         }
     }
 
+
+
+}
 fn create_tls_config(
     config: &Config,
 ) -> rustls::ServerConfig {