Rust work

Cargo.toml

@@ -2,17 +2,30 @@
 name = "iguagile_engine"
 version = "0.1.0"
 edition = "2021"
-
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-tonic = "0.8"
+quiche = "0.17.1"
+tonic = "0.9.1"
 prost = "0.11"
-redis = "0.22.3"
+redis = "0.23.0"
 anyhow = "1.0"
 serde_json = "1.0"
 bitmaps = "3.2.0"
 tokio = { version = "1.0", features = ["macros", "rt-multi-thread"] }
+log = { version = "0.4", features = ["std"] }
+libc = "0.2"
+libm = "0.2"
+ring = "0.16"
+slab = "0.4"
+lazy_static = "1"
+boring = { version = "2.0.0", optional = true }
+foreign-types-shared = { version = "0.3.0", optional = true }
+sfv = { version = "0.9", optional = true }
+smallvec = { version = "1.10", features = ["serde", "union"] }
 
 [build-dependencies]
-tonic-build = "0.8"
+tonic-build = "0.9.0"
+
+[dev-dependencies]
+mio = { version = "0.8", features = ["net", "os-poll"] }

iguagile/engine.go

@@ -33,7 +33,7 @@ type Engine struct {
 	idGenerator          *idGenerator
 	logger               *log.Logger
 	serverProto          *pb.Server
-	RoomUpdateDuration   time.Duration
+	room_update_duration   time.Duration
 	ServerUpdateDuration time.Duration
 }
 
@@ -44,7 +44,7 @@ func New(factory RoomServiceFactory, store Store) *Engine {
 		factory:              factory,
 		store:                store,
 		logger:               log.New(os.Stdout, "iguagile-engine ", log.Lshortfile),
-		RoomUpdateDuration:   time.Minute * 3,
+		room_update_duration:   time.Minute * 3,
 		ServerUpdateDuration: time.Minute * 3,
 	}
 }
@@ -124,7 +124,7 @@ func (e *Engine) Start(ctx context.Context, address, apiAddress string, tlsConf
 
 	go func(ctx context.Context) {
 		serverTicker := time.NewTicker(e.ServerUpdateDuration)
-		roomTicker := time.NewTicker(e.RoomUpdateDuration)
+		roomTicker := time.NewTicker(e.room_update_duration)
 		defer serverTicker.Stop()
 		defer roomTicker.Stop()
 		for {

src/client.rs

@@ -0,0 +1,35 @@
+use std::net::{ToSocketAddrs, UdpSocket};
+use std::sync::Arc;
+
+pub trait Client {
+    fn get_id(&self) -> u16;
+    fn read_loop(&mut self);
+}
+
+pub struct QUICClient {
+    id: u16,
+    sock: Arc<UdpSocket>,
+}
+
+impl QUICClient {
+    pub fn new(id: u16, addr: impl ToSocketAddrs) -> Result<Self, anyhow::Error> {
+        let sock = UdpSocket::bind(addr)?;
+        let sock = Arc::new(sock);
+        Ok(QUICClient { id, sock })
+    }
+}
+
+impl Client for QUICClient {
+    fn get_id(&self) -> u16 {
+        self.id
+    }
+
+    fn read_loop(&mut self) {
+        let mut buf = [0; 1024];
+        loop {
+            let (amt, src) = self.sock.recv_from(&mut buf).unwrap();
+            println!("Received {} bytes from {}", amt, src);
+            self.sock.send_to(&buf, src).unwrap();
+        }
+    }
+}

src/engine.rs

@@ -0,0 +1,124 @@
+use std::collections::HashMap;
+use std::net::{SocketAddr, UdpSocket};
+use std::rc::Rc;
+use std::sync::Arc;
+use std::time;
+
+use mio::{net, Events, Interest, Poll, Token};
+use ring::rand::SystemRandom;
+
+use crate::id::IdPool;
+use crate::room::Room;
+use crate::store::{iguagile, Store};
+
+const MAX_DATAGRAM_SIZE: usize = 1350;
+
+pub struct Engine {
+    server_id: i64,
+    rooms: Rc<HashMap<i64, Room>>,
+    // factory: RoomServiceFactory,
+    store: Box<dyn Store>,
+    id_pool: Box<dyn IdPool>,
+    server_proto: iguagile::Server,
+    room_update_duration: time::Duration,
+    server_update_duration: time::Duration,
+}
+
+impl Engine {
+    pub fn new(
+        server_id: i64,
+        store: Box<dyn Store>,
+        id_pool: Box<dyn IdPool>,
+        server_proto: iguagile::Server,
+        room_update_duration: time::Duration,
+        server_update_duration: time::Duration,
+    ) -> Self {
+        Engine {
+            server_id: server_id,
+            rooms: Rc::new(HashMap::new()),
+            store: store,
+            id_pool: id_pool,
+            server_proto: server_proto,
+            room_update_duration: room_update_duration,
+            server_update_duration: server_update_duration,
+        }
+    }
+
+    pub fn serve(&self, socket: Arc<UdpSocket>) -> Result<(), anyhow::Error> {
+        // let mut room_update_timer = time::Instant::now();
+        // let mut server_update_timer = time::Instant::now();
+        // loop {
+        //     if room_update_timer.elapsed() > self.room_update_duration {
+        //         self.update_rooms()?;
+        //         room_update_timer = time::Instant::now();
+        //     }
+        //     if server_update_timer.elapsed() > self.server_update_duration {
+        //         self.update_server()?;
+        //         server_update_timer = time::Instant::now();
+        //     }
+        // }
+        todo!()
+
+        // self.rooms.get(&self.server_id).unwrap().serve(socket)
+    }
+
+    pub fn start(
+        self,
+        serverAddress: SocketAddr,
+        apiAddress: SocketAddr,
+    ) -> Result<(), anyhow::Error> {
+        let mut poll = Poll::new().unwrap();
+        let mut socket = net::UdpSocket::bind(serverAddress).unwrap();
+
+        poll.registry()
+            .register(&mut socket, Token(0), Interest::READABLE)
+            .unwrap();
+
+        let mut config = quiche::Config::new(quiche::PROTOCOL_VERSION).unwrap();
+        config
+            .load_cert_chain_from_pem_file("examples/cert.crt")
+            .unwrap();
+        config
+            .load_priv_key_from_pem_file("examples/cert.key")
+            .unwrap();
+        config
+            .set_application_protos(&[b"hq-interop", b"hq-29", b"hq-28", b"hq-27", b"http/0.9"])
+            .unwrap();
+
+        config.set_max_idle_timeout(5000);
+        config.set_max_recv_udp_payload_size(MAX_DATAGRAM_SIZE);
+        config.set_max_send_udp_payload_size(MAX_DATAGRAM_SIZE);
+        config.set_initial_max_data(10_000_000);
+        config.set_initial_max_stream_data_bidi_local(1_000_000);
+        config.set_initial_max_stream_data_bidi_remote(1_000_000);
+        config.set_initial_max_stream_data_uni(1_000_000);
+        config.set_initial_max_streams_bidi(100);
+        config.set_initial_max_streams_uni(100);
+        config.set_disable_active_migration(true);
+        config.enable_early_data();
+
+        let rng = SystemRandom::new();
+        let conn_id_seed = ring::hmac::Key::generate(ring::hmac::HMAC_SHA256, &rng).unwrap();
+        let mut events = Events::with_capacity(1024);
+        let local_addr = socket.local_addr().unwrap();
+
+        loop {
+            // Find the shorter timeout from all the active connections.
+            //
+            // TODO: use event loop that properly supports timers
+            // let timeout = clients.values().filter_map(|c| c.conn.timeout()).min()a
+
+            let timeout = None;
+            poll.poll(&mut events, timeout).unwrap();
+
+            // Read incoming UDP packets from the socket and feed them to quiche,
+            // until there are no more packets to read.
+            'read: loop {}
+        }
+        return Ok(());
+    }
+}
+
+// test
+#[cfg(test)]
+mod tests {}

src/main.rs

@@ -1,15 +1,23 @@
+use std::process::exit;
+
 use crate::store::MemoryStore;
 
+mod client;
+mod engine;
 mod id;
+mod relay;
+mod room;
 mod store;
 
 fn main() {
     let redis = redis::Client::open("redis://127.0.0.1");
     if let Err(e) = redis {
         println!("redis error: {}", e);
-        return;
+        exit(1);
     }
+
     let m = MemoryStore::new(redis.unwrap());
+
     let _ = m;
     println!("Hello, world!");
 }

src/relay.rs

@@ -0,0 +1,39 @@
+use crate::room::Room;
+
+type ReceiveFunc = dyn Fn(u16, &[u8]) -> Result<(), anyhow::Error>;
+
+trait RelayServiceTrait {
+    fn receive_func(&self, stream_name: String) -> Result<&ReceiveFunc, anyhow::Error>;
+    fn on_register_client(&self, client_id: u16) -> Result<(), anyhow::Error>;
+    fn on_unregister_client(&self, client_id: u16) -> Result<(), anyhow::Error>;
+    fn on_change_host(&self, client_id: u16) -> Result<(), anyhow::Error>;
+    fn destroy(&self) -> Result<(), anyhow::Error>;
+}
+
+pub struct RelayService {
+    room: Room,
+}
+
+impl RelayService {
+    pub fn new(room: Room) -> Self {
+        RelayService { room }
+    }
+}
+
+impl RelayServiceTrait for RelayService {
+    fn receive_func(&self, stream_name: String) -> Result<&ReceiveFunc, anyhow::Error> {
+        anyhow::bail!("Not implemented");
+    }
+    fn on_register_client(&self, client_id: u16) -> Result<(), anyhow::Error> {
+        anyhow::bail!("Not implemented");
+    }
+    fn on_unregister_client(&self, client_id: u16) -> Result<(), anyhow::Error> {
+        anyhow::bail!("Not implemented");
+    }
+    fn on_change_host(&self, client_id: u16) -> Result<(), anyhow::Error> {
+        anyhow::bail!("Not implemented");
+    }
+    fn destroy(&self) -> Result<(), anyhow::Error> {
+        anyhow::bail!("Not implemented");
+    }
+}