termchat/
application.rs

1use super::state::{State, CursorMovement, ChatMessage, MessageType, ScrollMovement};
2use crate::{
3    state::Window,
4    terminal_events::{TerminalEventCollector},
5};
6use crate::renderer::{Renderer};
7use crate::action::{Action, Processing};
8use crate::commands::{CommandManager};
9use crate::message::{NetMessage, Chunk};
10use crate::util::{Error, Result, Reportable};
11use crate::commands::send_file::{SendFileCommand};
12#[cfg(feature = "stream-video")]
13use crate::commands::send_stream::{SendStreamCommand, StopStreamCommand};
14use crate::config::Config;
15use crate::encoder::{self, Encoder};
16
17use crossterm::event::{Event as TermEvent, KeyCode, KeyEvent, KeyModifiers};
18
19use message_io::events::{EventReceiver};
20use message_io::network::{Endpoint, Transport};
21use message_io::node::{
22    self, StoredNodeEvent as NodeEvent, StoredNetEvent as NetEvent, NodeTask, NodeHandler,
23};
24
25use std::io::{ErrorKind};
26
27pub enum Signal {
28    Terminal(TermEvent),
29    Action(Box<dyn Action>),
30    // Close event with an optional error in case of failure
31    // Close(None) means no error happened
32    Close(Option<Error>),
33}
34
35pub struct Application<'a> {
36    config: &'a Config,
37    commands: CommandManager,
38    state: State,
39    node: NodeHandler<Signal>,
40    _task: NodeTask,
41    //read_file_ev: ReadFile,
42    _terminal_events: TerminalEventCollector,
43    receiver: EventReceiver<NodeEvent<Signal>>,
44    encoder: Encoder,
45}
46
47impl<'a> Application<'a> {
48    pub fn new(config: &'a Config) -> Result<Application<'a>> {
49        let (handler, listener) = node::split();
50
51        let terminal_handler = handler.clone(); // Collect terminal events
52        let _terminal_events = TerminalEventCollector::new(move |term_event| match term_event {
53            Ok(event) => terminal_handler.signals().send(Signal::Terminal(event)),
54            Err(e) => terminal_handler.signals().send(Signal::Close(Some(e))),
55        })?;
56
57        let (_task, receiver) = listener.enqueue();
58
59        let commands = CommandManager::default().with(SendFileCommand);
60        #[cfg(feature = "stream-video")]
61        let commands = commands.with(SendStreamCommand).with(StopStreamCommand);
62
63        Ok(Application {
64            config,
65            commands,
66            state: State::default(),
67            node: handler,
68            _task,
69            // Stored because we need its internal thread running until the Application was dropped
70            _terminal_events,
71            receiver,
72            encoder: Encoder::new(),
73        })
74    }
75
76    pub fn run(&mut self, out: impl std::io::Write) -> Result<()> {
77        let mut renderer = Renderer::new(out)?;
78        renderer.render(&self.state, &self.config.theme)?;
79
80        let server_addr = ("0.0.0.0", self.config.tcp_server_port);
81        let (_, server_addr) = self.node.network().listen(Transport::FramedTcp, server_addr)?;
82        self.node.network().listen(Transport::Udp, self.config.discovery_addr)?;
83
84        let (discovery_endpoint, _) =
85            self.node.network().connect_sync(Transport::Udp, self.config.discovery_addr)?;
86        let message = NetMessage::HelloLan(self.config.user_name.clone(), server_addr.port());
87        self.node.network().send(discovery_endpoint, self.encoder.encode(message));
88
89        loop {
90            match self.receiver.receive() {
91                NodeEvent::Network(net_event) => match net_event {
92                    NetEvent::Connected(_, _) => { /* handler in the connect call*/ }
93                    NetEvent::Message(endpoint, message) => match encoder::decode(&message) {
94                        Some(net_message) => self.process_network_message(endpoint, net_message),
95                        None => return Err("Unknown message received".into()),
96                    },
97                    NetEvent::Accepted(_endpoint, _resource_id) => (),
98                    NetEvent::Disconnected(endpoint) => {
99                        self.state.disconnected_user(endpoint);
100                        //If the endpoint was sending a stream make sure to close its window
101                        self.state.windows.remove(&endpoint);
102                        self.righ_the_bell();
103                    }
104                },
105                NodeEvent::Signal(signal) => match signal {
106                    Signal::Terminal(term_event) => {
107                        self.process_terminal_event(term_event);
108                    }
109                    Signal::Action(action) => {
110                        self.process_action(action);
111                    }
112                    Signal::Close(error) => {
113                        self.node.stop();
114                        return match error {
115                            Some(error) => Err(error),
116                            None => Ok(()),
117                        }
118                    }
119                },
120            }
121            renderer.render(&self.state, &self.config.theme)?;
122        }
123        //Renderer is destroyed here and the terminal is recovered
124    }
125
126    fn process_network_message(&mut self, endpoint: Endpoint, message: NetMessage) {
127        match message {
128            // by udp (multicast):
129            NetMessage::HelloLan(user, server_port) => {
130                let server_addr = (endpoint.addr().ip(), server_port);
131                if user != self.config.user_name {
132                    let mut try_connect = || -> Result<()> {
133                        let (user_endpoint, _) =
134                            self.node.network().connect_sync(Transport::FramedTcp, server_addr)?;
135                        let message = NetMessage::HelloUser(self.config.user_name.clone());
136                        self.node.network().send(user_endpoint, self.encoder.encode(message));
137                        self.state.connected_user(user_endpoint, &user);
138                        Ok(())
139                    };
140                    try_connect().report_if_err(&mut self.state);
141                }
142            }
143            // by tcp:
144            NetMessage::HelloUser(user) => {
145                self.state.connected_user(endpoint, &user);
146                self.righ_the_bell();
147            }
148            NetMessage::UserMessage(content) => {
149                if let Some(user) = self.state.user_name(endpoint) {
150                    let message = ChatMessage::new(user.into(), MessageType::Text(content));
151                    self.state.add_message(message);
152                    self.righ_the_bell();
153                }
154            }
155            NetMessage::UserData(file_name, chunk) => {
156                use std::io::Write;
157                if self.state.user_name(endpoint).is_some() {
158                    // safe unwrap due to check
159                    let user = self.state.user_name(endpoint).unwrap().to_owned();
160
161                    match chunk {
162                        Chunk::Error => {
163                            format!("'{}' had an error while sending '{}'", user, file_name)
164                                .report_err(&mut self.state);
165                        }
166                        Chunk::End => {
167                            format!(
168                                "Successfully received file '{}' from user '{}'!",
169                                file_name, user
170                            )
171                            .report_info(&mut self.state);
172                            self.righ_the_bell();
173                        }
174                        Chunk::Data(data) => {
175                            let try_write = || -> Result<()> {
176                                let user_path = std::env::temp_dir().join("termchat").join(&user);
177                                match std::fs::create_dir_all(&user_path) {
178                                    Ok(_) => (),
179                                    Err(ref err) if err.kind() == ErrorKind::AlreadyExists => (),
180                                    Err(e) => return Err(e.into()),
181                                }
182
183                                let file_path = user_path.join(file_name);
184                                std::fs::OpenOptions::new()
185                                    .create(true)
186                                    .append(true)
187                                    .open(file_path)?
188                                    .write_all(&data)?;
189
190                                Ok(())
191                            };
192
193                            try_write().report_if_err(&mut self.state);
194                        }
195                    }
196                }
197            }
198            NetMessage::Stream(data) => match data {
199                Some((data, width, height)) if data.len() == width * height / 2 => {
200                    self.state
201                        .windows
202                        .entry(endpoint)
203                        .or_insert_with(|| Window::new(width, height));
204                    self.state.update_window(&endpoint, data, width, height);
205                }
206                _ => {
207                    self.state.windows.remove(&endpoint);
208                }
209            },
210        }
211    }
212
213    fn process_terminal_event(&mut self, term_event: TermEvent) {
214        match term_event {
215            TermEvent::Mouse(_) => (),
216            TermEvent::Resize(_, _) => (),
217            TermEvent::Key(KeyEvent { code, modifiers }) => match code {
218                KeyCode::Esc => {
219                    self.node.signals().send_with_priority(Signal::Close(None));
220                }
221                KeyCode::Char(character) => {
222                    if character == 'c' && modifiers.contains(KeyModifiers::CONTROL) {
223                        self.node.signals().send_with_priority(Signal::Close(None));
224                    }
225                    else {
226                        self.state.input_write(character);
227                    }
228                }
229                KeyCode::Enter => {
230                    if let Some(input) = self.state.reset_input() {
231                        match self.commands.find_command_action(&input).transpose() {
232                            Ok(action) => {
233                                let message = ChatMessage::new(
234                                    format!("{} (me)", self.config.user_name),
235                                    MessageType::Text(input.clone()),
236                                );
237                                self.state.add_message(message);
238
239                                for endpoint in self.state.all_user_endpoints() {
240                                    self.node.network().send(
241                                        *endpoint,
242                                        self.encoder.encode(NetMessage::UserMessage(input.clone())),
243                                    );
244                                }
245
246                                match action {
247                                    Some(action) => self.process_action(action),
248                                    None => {
249                                        if input.starts_with('?') {
250                                            String::from("This command doesn't exists")
251                                                .report_err(&mut self.state);
252                                        }
253                                    }
254                                }
255                            }
256                            Err(error) => {
257                                error.report_err(&mut self.state);
258                            }
259                        };
260                    }
261                }
262                KeyCode::Delete => {
263                    self.state.input_remove();
264                }
265                KeyCode::Backspace => {
266                    self.state.input_remove_previous();
267                }
268                KeyCode::Left => {
269                    self.state.input_move_cursor(CursorMovement::Left);
270                }
271                KeyCode::Right => {
272                    self.state.input_move_cursor(CursorMovement::Right);
273                }
274                KeyCode::Home => {
275                    self.state.input_move_cursor(CursorMovement::Start);
276                }
277                KeyCode::End => {
278                    self.state.input_move_cursor(CursorMovement::End);
279                }
280                KeyCode::Up => {
281                    self.state.messages_scroll(ScrollMovement::Up);
282                }
283                KeyCode::Down => {
284                    self.state.messages_scroll(ScrollMovement::Down);
285                }
286                KeyCode::PageUp => {
287                    self.state.messages_scroll(ScrollMovement::Start);
288                }
289                _ => (),
290            },
291        }
292    }
293
294    fn process_action(&mut self, mut action: Box<dyn Action>) {
295        match action.process(&mut self.state, self.node.network()) {
296            Processing::Completed => (),
297            Processing::Partial(delay) => {
298                self.node.signals().send_with_timer(Signal::Action(action), delay);
299            }
300        }
301    }
302
303    pub fn node_handler(&self) -> NodeHandler<Signal> {
304        self.node.clone()
305    }
306
307    pub fn righ_the_bell(&self) {
308        if self.config.terminal_bell {
309            print!("\x07");
310        }
311    }
312}
termchat/application.rs

termchat/
application.rs
