IPC client connections

apps/magpie/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 canary = { path = "../.." }
 glium = "0.32"
+magpie-types = { path = "../../crates/magpie-types" }
 mio = { version = "0.8", features = ["net", "os-poll"] }
 mio-signals = "0.2"
 slab = "0.4"

apps/magpie/src/ipc.rs

@@ -1,9 +1,11 @@
+use std::io::Read;
 use std::ops::{Deref, DerefMut};
 use std::path::{Path, PathBuf};
+use std::str::from_utf8;
 use std::sync::mpsc::{channel, Receiver, Sender};
 use std::time::Duration;
 
-use mio::net::UnixListener;
+use mio::net::{UnixListener, UnixStream};
 use mio::{Events, Interest, Poll, Token, Waker};
 use mio_signals::{Signal, Signals};
 use slab::Slab;
@@ -48,7 +50,50 @@ impl DerefMut for Listener {
     }
 }
 
-pub struct Client {}
+pub struct Client {
+    connection: UnixStream,
+}
+
+impl Client {
+    pub fn new(connection: UnixStream) -> Self {
+        Self { connection }
+    }
+
+    pub fn on_readable(&mut self) -> std::io::Result<bool> {
+        let mut connection_closed = false;
+        let mut received_data = vec![0; 4096];
+        let mut bytes_read = 0;
+
+        loop {
+            match self.connection.read(&mut received_data[bytes_read..]) {
+                Ok(0) => {
+                    connection_closed = true;
+                    break;
+                }
+                Ok(n) => {
+                    bytes_read += n;
+                    if bytes_read >= received_data.len() {
+                        received_data.resize(received_data.len() + 1024, 0);
+                    }
+                }
+                Err(ref err) if err.kind() == std::io::ErrorKind::WouldBlock => break,
+                Err(ref err) if err.kind() == std::io::ErrorKind::Interrupted => continue,
+                Err(err) => return Err(err),
+            }
+        }
+
+        if bytes_read > 0 {
+            let received_data = &received_data[..bytes_read];
+            if let Ok(str_buf) = from_utf8(received_data) {
+                println!("Received data: {}", str_buf.trim_end());
+            } else {
+                println!("Received (non-UTF-8) data: {:?}", received_data);
+            }
+        }
+
+        Ok(connection_closed)
+    }
+}
 
 pub struct Ipc {
     pub message_recv: Receiver<IpcMessage>,
@@ -121,7 +166,19 @@ impl Ipc {
                 loop {
                     match self.listener.accept() {
                         Ok((connection, address)) => {
-                            println!("Got a connection from: {:?}", address);
+                            let token = Token(self.clients.vacant_key());
+                            println!(
+                                "Accepting connection (Client #{}) from {:?}",
+                                token.0, address
+                            );
+                            let mut client = Client::new(connection);
+                            let interest = Interest::READABLE;
+                            self.poll.registry().register(
+                                &mut client.connection,
+                                token,
+                                interest,
+                            )?;
+                            self.clients.insert(client);
                         }
                         Err(ref err) if err.kind() == std::io::ErrorKind::WouldBlock => break,
                         Err(err) => return Err(err),
@@ -134,6 +191,12 @@ impl Ipc {
                     self.quit = true;
                 }
             } else if let Some(client) = self.clients.get_mut(event.token().0) {
+                let disconnected = client.on_readable()?;
+                if disconnected {
+                    println!("Client #{} disconnected", event.token().0);
+                    let mut client = self.clients.remove(event.token().0);
+                    self.poll.registry().deregister(&mut client.connection)?;
+                }
             } else {
                 panic!("Unrecognized event token: {:?}", event);
             }

apps/magpie/src/main.rs

@@ -10,5 +10,5 @@ fn main() -> std::io::Result<()> {
     let (ipc, ipc_sender) = ipc::Ipc::new(window_sender)?;
     let _ipc_thread = std::thread::spawn(|| ipc.run());
     let window_store = window::WindowStore::new(ipc_sender);
-    window_store.run(event_loop);
+    window_store.run(event_loop)
 }