diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/interface.rs | 68 | ||||
| -rw-r--r-- | src/lib.rs | 3 | ||||
| -rw-r--r-- | src/main.rs | 35 | ||||
| -rw-r--r-- | src/protocol.rs | 4 |
4 files changed, 88 insertions, 22 deletions
diff --git a/src/interface.rs b/src/interface.rs new file mode 100644 index 0000000..568c450 --- /dev/null +++ b/src/interface.rs @@ -0,0 +1,68 @@ +use std::io; +use std::io::{BufRead, BufReader, ErrorKind, Write}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::thread; + +use super::protocol::{ClientboundPacket, ServerboundPacket}; +use crossbeam_channel::Sender; +use log::{debug, error, info, warn}; + +pub fn network_loop(s: Sender<(u32, ServerboundPacket, Sender<ClientboundPacket>)>) { + let listener = UnixListener::bind("/run/user/1000/calendar").unwrap(); + let mut id_counter = 0; + + loop { + let (stream, addr) = listener.accept().unwrap(); + let s = s.clone(); + let id = id_counter; + id_counter += 1; + thread::spawn(move || { + info!("client connected: {:?}", addr); + if let Err(err) = handle_connection(id, stream.try_clone().unwrap(), s.clone()) { + warn!("client dropped: {:?} ({})", addr, err); + } else { + info!("client dropped: {:?}", addr); + } + }); + } +} + +fn handle_connection( + id: u32, + mut stream: UnixStream, + s: Sender<(u32, ServerboundPacket, Sender<ClientboundPacket>)>, +) -> io::Result<()> { + let mut reader = BufReader::new(stream.try_clone()?); + let (responder, responses) = crossbeam_channel::unbounded(); + thread::spawn(move || { + for m in responses { + debug!("{id} -> {m:?}"); + match stream + .write_fmt(format_args!("{}\n", serde_json::to_string(&m).unwrap())) + .map_err(|e| e.kind()) + { + Ok(_) => (), + Err(ErrorKind::BrokenPipe) => break, + Err(e) => error!("network error: {:?}", e), + } + } + }); + let mut buf = String::new(); + loop { + if reader.read_line(&mut buf)? == 0 { + break Ok(()); + }; + match serde_json::from_str::<ServerboundPacket>(buf.as_str()) { + Ok(packet) => { + debug!("{id} <- {packet:?}"); + s.send((id, packet, responder.clone())) + .map_err(|_| io::Error::from(ErrorKind::InvalidInput))?; + } + Err(err) => responder + .send(ClientboundPacket::Error(format!("{}", &err))) + .map_err(|_| io::Error::from(ErrorKind::InvalidInput))?, + } + + buf.clear(); + } +} @@ -1,2 +1,3 @@ pub mod protocol; -pub mod occurence;
\ No newline at end of file +pub mod occurence; +pub mod interface; diff --git a/src/main.rs b/src/main.rs index a26a1dd..fb975d6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,27 +1,22 @@ +use std::time::Duration; -use std::io::{BufRead, BufReader}; -use std::os::unix::net::{UnixListener, UnixStream}; -use std::thread; - -fn handle_client(stream: UnixStream) { - let stream = BufReader::new(stream); - for line in stream.lines() { - println!("{}", line.unwrap()); - } -} +use calender_thing::{ + interface::network_loop, + protocol::{ClientboundPacket, ServerboundPacket}, +}; +use crossbeam_channel::{Receiver, Sender}; fn main() { - let listener = UnixListener::bind("/run/user/1000/calendar").unwrap(); + let (s, r) = crossbeam_channel::unbounded(); + std::thread::spawn(move || network_loop(s)); + main_loop(r) +} - for stream in listener.incoming() { - match stream { - Ok(stream) => { - thread::spawn(|| handle_client(stream)); - } - Err(err) => { - println!("Error: {}", err); - break; - } +fn main_loop(packets: Receiver<(u32, ServerboundPacket, Sender<ClientboundPacket>)>) { + loop { + for (client_id, packet, responder) in packets.try_iter() { + println!("{:?}, {:?}, {:?}", client_id, packet, responder); } + std::thread::sleep(Duration::from_secs_f64(10.0 / 30.0)); } } diff --git a/src/protocol.rs b/src/protocol.rs index fdd84d4..a09f879 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -3,12 +3,15 @@ use serde::{Deserialize, Serialize}; use crate::occurence::Condition; #[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] pub enum ClientboundPacket { Handshake { version: String }, + Error(String), DownloadResponse(Vec<Task>), } #[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] pub enum ServerboundPacket { Download, } @@ -27,4 +30,3 @@ pub struct Task { occurence: Option<Condition>, deadline: Option<Condition>, } - |