diff options
| author | Kunal Mohan <kunalmohan99@gmail.com> | 2021-04-30 20:40:32 +0530 |
|---|---|---|
| committer | Kunal Mohan <kunalmohan99@gmail.com> | 2021-05-04 20:48:17 +0530 |
| commit | c6f93ba0d231c12d528e33315a4fdb5d46a1355f (patch) | |
| tree | ac97b38a24b024068d2664bc5ca55319ddd57ab4 /src | |
| parent | Place socket file in runtime directory or cache directory and use names crate... (diff) | |
| download | sl0ppij-c6f93ba0d231c12d528e33315a4fdb5d46a1355f.tar.gz sl0ppij-c6f93ba0d231c12d528e33315a4fdb5d46a1355f.tar.xz sl0ppij-c6f93ba0d231c12d528e33315a4fdb5d46a1355f.zip | |
use better names for senders, receivers and threads
Diffstat (limited to 'src')
| -rw-r--r-- | src/client/mod.rs | 2 | ||||
| -rw-r--r-- | src/common/mod.rs | 3 | ||||
| -rw-r--r-- | src/common/os_input_output.rs | 62 | ||||
| -rw-r--r-- | src/server/mod.rs | 48 | ||||
| -rw-r--r-- | src/tests/fakes.rs | 40 |
5 files changed, 91 insertions, 64 deletions
diff --git a/src/client/mod.rs b/src/client/mod.rs index bbde8c35..2bf814c2 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -105,7 +105,7 @@ pub fn start_client(mut os_input: Box<dyn ClientOsApi>, opts: CliArgs) { let os_input = os_input.clone(); move || { loop { - let (instruction, mut err_ctx) = os_input.client_recv(); + let (instruction, mut err_ctx) = os_input.recv_from_server(); err_ctx.add_call(ContextType::Client(ClientContext::from(&instruction))); if let ClientInstruction::Exit = instruction { break; diff --git a/src/common/mod.rs b/src/common/mod.rs index 10147a2c..bf4c3ae6 100644 --- a/src/common/mod.rs +++ b/src/common/mod.rs @@ -85,7 +85,6 @@ lazy_static! { .unwrap_or_else(|| project_dir.cache_dir()); std::fs::create_dir_all(ipc_dir).unwrap(); let session_name = names::Generator::default().next().unwrap(); - let x = ipc_dir.join(session_name); - x + ipc_dir.join(session_name) }; } diff --git a/src/common/os_input_output.rs b/src/common/os_input_output.rs index 5dfda264..9a124011 100644 --- a/src/common/os_input_output.rs +++ b/src/common/os_input_output.rs @@ -186,8 +186,8 @@ impl<T: Serialize> IpcSenderWithContext<T> { #[derive(Clone)] pub struct ServerOsInputOutput { orig_termios: Arc<Mutex<termios::Termios>>, - recv_socket: Option<Arc<Mutex<io::BufReader<LocalSocketStream>>>>, - sender_socket: Arc<Mutex<Option<IpcSenderWithContext<ClientInstruction>>>>, + receive_instructions_from_client: Option<Arc<Mutex<io::BufReader<LocalSocketStream>>>>, + send_instructions_to_client: Arc<Mutex<Option<IpcSenderWithContext<ClientInstruction>>>>, } /// The `ServerOsApi` trait represents an abstract interface to the features of an operating system that @@ -211,7 +211,7 @@ pub trait ServerOsApi: Send + Sync { /// Returns a [`Box`] pointer to this [`ServerOsApi`] struct. fn box_clone(&self) -> Box<dyn ServerOsApi>; /// Receives a message on server-side IPC channel - fn server_recv(&self) -> (ServerInstruction, ErrorContext); + fn recv_from_client(&self) -> (ServerInstruction, ErrorContext); /// Sends a message to client fn send_to_client(&self, msg: ClientInstruction); /// Adds a sender to client @@ -251,11 +251,19 @@ impl ServerOsApi for ServerOsInputOutput { waitpid(Pid::from_raw(pid), None).unwrap(); Ok(()) } - fn server_recv(&self) -> (ServerInstruction, ErrorContext) { - bincode::deserialize_from(&mut *self.recv_socket.as_ref().unwrap().lock().unwrap()).unwrap() + fn recv_from_client(&self) -> (ServerInstruction, ErrorContext) { + bincode::deserialize_from( + &mut *self + .receive_instructions_from_client + .as_ref() + .unwrap() + .lock() + .unwrap(), + ) + .unwrap() } fn send_to_client(&self, msg: ClientInstruction) { - self.sender_socket + self.send_instructions_to_client .lock() .unwrap() .as_mut() @@ -264,9 +272,9 @@ impl ServerOsApi for ServerOsInputOutput { .unwrap(); } fn add_client_sender(&mut self) { - assert!(self.sender_socket.lock().unwrap().is_none()); + assert!(self.send_instructions_to_client.lock().unwrap().is_none()); let sock_fd = self - .recv_socket + .receive_instructions_from_client .as_ref() .unwrap() .lock() @@ -275,10 +283,12 @@ impl ServerOsApi for ServerOsInputOutput { .as_raw_fd(); let dup_fd = unistd::dup(sock_fd).unwrap(); let dup_sock = unsafe { LocalSocketStream::from_raw_fd(dup_fd) }; - *self.sender_socket.lock().unwrap() = Some(IpcSenderWithContext::new(dup_sock)); + *self.send_instructions_to_client.lock().unwrap() = + Some(IpcSenderWithContext::new(dup_sock)); } fn update_receiver(&mut self, stream: LocalSocketStream) { - self.recv_socket = Some(Arc::new(Mutex::new(io::BufReader::new(stream)))); + self.receive_instructions_from_client = + Some(Arc::new(Mutex::new(io::BufReader::new(stream)))); } } @@ -293,16 +303,16 @@ pub fn get_server_os_input() -> ServerOsInputOutput { let orig_termios = Arc::new(Mutex::new(current_termios)); ServerOsInputOutput { orig_termios, - recv_socket: None, - sender_socket: Arc::new(Mutex::new(None)), + receive_instructions_from_client: None, + send_instructions_to_client: Arc::new(Mutex::new(None)), } } #[derive(Clone)] pub struct ClientOsInputOutput { orig_termios: Arc<Mutex<termios::Termios>>, - server_sender: Arc<Mutex<Option<IpcSenderWithContext<ServerInstruction>>>>, - receiver: Arc<Mutex<Option<io::BufReader<LocalSocketStream>>>>, + send_instructions_to_server: Arc<Mutex<Option<IpcSenderWithContext<ServerInstruction>>>>, + receive_instructions_from_server: Arc<Mutex<Option<io::BufReader<LocalSocketStream>>>>, } /// The `ClientOsApi` trait represents an abstract interface to the features of an operating system that @@ -326,7 +336,7 @@ pub trait ClientOsApi: Send + Sync { fn send_to_server(&self, msg: ServerInstruction); /// Receives a message on client-side IPC channel // This should be called from the client-side router thread only. - fn client_recv(&self) -> (ClientInstruction, ErrorContext); + fn recv_from_server(&self) -> (ClientInstruction, ErrorContext); fn receive_sigwinch(&self, cb: Box<dyn Fn()>); /// Establish a connection with the server socket. fn connect_to_server(&self); @@ -360,7 +370,7 @@ impl ClientOsApi for ClientOsInputOutput { Box::new(stdout) } fn send_to_server(&self, msg: ServerInstruction) { - self.server_sender + self.send_instructions_to_server .lock() .unwrap() .as_mut() @@ -368,8 +378,16 @@ impl ClientOsApi for ClientOsInputOutput { .send(msg) .unwrap(); } - fn client_recv(&self) -> (ClientInstruction, ErrorContext) { - bincode::deserialize_from(&mut self.receiver.lock().unwrap().as_mut().unwrap()).unwrap() + fn recv_from_server(&self) -> (ClientInstruction, ErrorContext) { + bincode::deserialize_from( + &mut self + .receive_instructions_from_server + .lock() + .unwrap() + .as_mut() + .unwrap(), + ) + .unwrap() } fn receive_sigwinch(&self, cb: Box<dyn Fn()>) { let mut signals = Signals::new(&[SIGWINCH, SIGTERM, SIGINT, SIGQUIT]).unwrap(); @@ -397,8 +415,8 @@ impl ClientOsApi for ClientOsInputOutput { let dup_fd = unistd::dup(sock_fd).unwrap(); let receiver = unsafe { LocalSocketStream::from_raw_fd(dup_fd) }; let sender = IpcSenderWithContext::new(socket); - *self.server_sender.lock().unwrap() = Some(sender); - *self.receiver.lock().unwrap() = Some(io::BufReader::new(receiver)); + *self.send_instructions_to_server.lock().unwrap() = Some(sender); + *self.receive_instructions_from_server.lock().unwrap() = Some(io::BufReader::new(receiver)); } } @@ -413,7 +431,7 @@ pub fn get_client_os_input() -> ClientOsInputOutput { let orig_termios = Arc::new(Mutex::new(current_termios)); ClientOsInputOutput { orig_termios, - server_sender: Arc::new(Mutex::new(None)), - receiver: Arc::new(Mutex::new(None)), + send_instructions_to_server: Arc::new(Mutex::new(None)), + receive_instructions_from_server: Arc::new(Mutex::new(None)), } } diff --git a/src/server/mod.rs b/src/server/mod.rs index bcfbd091..c1f93d5b 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -78,32 +78,34 @@ pub fn start_server(os_input: Box<dyn ServerOsApi>, opts: CliArgs) -> thread::Jo send_server_instructions.clone(), ); #[cfg(not(test))] - let _ = thread::Builder::new().name("listener".to_string()).spawn({ - let os_input = os_input.clone(); - let sessions = sessions.clone(); - let send_server_instructions = send_server_instructions.clone(); - move || { - drop(std::fs::remove_file(ZELLIJ_IPC_PIPE.clone())); - let listener = LocalSocketListener::bind(ZELLIJ_IPC_PIPE.clone()).unwrap(); - for stream in listener.incoming() { - match stream { - Ok(stream) => { - let mut os_input = os_input.clone(); - os_input.update_receiver(stream); - let sessions = sessions.clone(); - let send_server_instructions = send_server_instructions.clone(); - handle_client(sessions, os_input, send_server_instructions); - } - Err(err) => { - panic!("err {:?}", err); + let _ = thread::Builder::new() + .name("server_listener".to_string()) + .spawn({ + let os_input = os_input.clone(); + let sessions = sessions.clone(); + let send_server_instructions = send_server_instructions.clone(); + move || { + drop(std::fs::remove_file(ZELLIJ_IPC_PIPE.clone())); + let listener = LocalSocketListener::bind(ZELLIJ_IPC_PIPE.clone()).unwrap(); + for stream in listener.incoming() { + match stream { + Ok(stream) => { + let mut os_input = os_input.clone(); + os_input.update_receiver(stream); + let sessions = sessions.clone(); + let send_server_instructions = send_server_instructions.clone(); + handle_client(sessions, os_input, send_server_instructions); + } + Err(err) => { + panic!("err {:?}", err); + } } } } - } - }); + }); thread::Builder::new() - .name("ipc_server".to_string()) + .name("server_thread".to_string()) .spawn({ move || loop { let (instruction, mut err_ctx) = receive_server_instructions.recv().unwrap(); @@ -151,9 +153,9 @@ fn handle_client( send_server_instructions: SenderWithContext<ServerInstruction>, ) { thread::Builder::new() - .name("router".to_string()) + .name("server_router".to_string()) .spawn(move || loop { - let (instruction, mut err_ctx) = os_input.server_recv(); + let (instruction, mut err_ctx) = os_input.recv_from_client(); err_ctx.add_call(ContextType::IPCServer(ServerContext::from(&instruction))); let rlocked_sessions = sessions.read().unwrap(); match instruction { diff --git a/src/tests/fakes.rs b/src/tests/fakes.rs index 99bd008a..bc18d6fd 100644 --- a/src/tests/fakes.rs +++ b/src/tests/fakes.rs @@ -75,10 +75,10 @@ pub struct FakeInputOutput { win_sizes: Arc<Mutex<HashMap<RawFd, PositionAndSize>>>, possible_tty_inputs: HashMap<u16, Bytes>, last_snapshot_time: Arc<Mutex<Instant>>, - client_sender: SenderWithContext<ClientInstruction>, - client_receiver: Arc<Mutex<mpsc::Receiver<(ClientInstruction, ErrorContext)>>>, - server_sender: SenderWithContext<ServerInstruction>, - server_receiver: Arc<Mutex<mpsc::Receiver<(ServerInstruction, ErrorContext)>>>, + send_instructions_to_client: SenderWithContext<ClientInstruction>, + receive_instructions_from_server: Arc<Mutex<mpsc::Receiver<(ClientInstruction, ErrorContext)>>>, + send_instructions_to_server: SenderWithContext<ServerInstruction>, + receive_instructions_from_client: Arc<Mutex<mpsc::Receiver<(ServerInstruction, ErrorContext)>>>, should_trigger_sigwinch: Arc<(Mutex<bool>, Condvar)>, sigwinch_event: Option<PositionAndSize>, } @@ -90,10 +90,10 @@ impl FakeInputOutput { let stdout_writer = FakeStdoutWriter::new(last_snapshot_time.clone()); let (client_sender, client_receiver): ChannelWithContext<ClientInstruction> = mpsc::channel(); - let client_sender = SenderWithContext::new(SenderType::Sender(client_sender)); + let send_instructions_to_client = SenderWithContext::new(SenderType::Sender(client_sender)); let (server_sender, server_receiver): ChannelWithContext<ServerInstruction> = mpsc::channel(); - let server_sender = SenderWithContext::new(SenderType::Sender(server_sender)); + let send_instructions_to_server = SenderWithContext::new(SenderType::Sender(server_sender)); win_sizes.insert(0, winsize); // 0 is the current terminal FakeInputOutput { @@ -106,10 +106,10 @@ impl FakeInputOutput { io_events: Arc::new(Mutex::new(vec![])), win_sizes: Arc::new(Mutex::new(win_sizes)), possible_tty_inputs: get_possible_tty_inputs(), - server_receiver: Arc::new(Mutex::new(server_receiver)), - server_sender, - client_receiver: Arc::new(Mutex::new(client_receiver)), - client_sender, + receive_instructions_from_client: Arc::new(Mutex::new(server_receiver)), + send_instructions_to_server, + receive_instructions_from_server: Arc::new(Mutex::new(client_receiver)), + send_instructions_to_client, should_trigger_sigwinch: Arc::new((Mutex::new(false), Condvar::new())), sigwinch_event: None, } @@ -195,10 +195,14 @@ impl ClientOsApi for FakeInputOutput { Box::new(self.stdout_writer.clone()) } fn send_to_server(&self, msg: ServerInstruction) { - self.server_sender.send(msg).unwrap(); + self.send_instructions_to_server.send(msg).unwrap(); } - fn client_recv(&self) -> (ClientInstruction, ErrorContext) { - self.client_receiver.lock().unwrap().recv().unwrap() + fn recv_from_server(&self) -> (ClientInstruction, ErrorContext) { + self.receive_instructions_from_server + .lock() + .unwrap() + .recv() + .unwrap() } fn receive_sigwinch(&self, cb: Box<dyn Fn()>) { if self.sigwinch_event.is_some() { @@ -273,11 +277,15 @@ impl ServerOsApi for FakeInputOutput { self.io_events.lock().unwrap().push(IoEvent::Kill(fd)); Ok(()) } - fn server_recv(&self) -> (ServerInstruction, ErrorContext) { - self.server_receiver.lock().unwrap().recv().unwrap() + fn recv_from_client(&self) -> (ServerInstruction, ErrorContext) { + self.receive_instructions_from_client + .lock() + .unwrap() + .recv() + .unwrap() } fn send_to_client(&self, msg: ClientInstruction) { - self.client_sender.send(msg).unwrap(); + self.send_instructions_to_client.send(msg).unwrap(); } fn add_client_sender(&mut self) {} fn update_receiver(&mut self, _stream: LocalSocketStream) {} |
