use {
crate::{
utils::{bufio::BufIoIncoming, errorfmt::ErrorFmt},
xcon::{
Event, ExtensionData, ExtensionIdRange, Parser, XconData, XconError, consts::XGE_EVENT,
},
},
std::{mem, rc::Rc},
};
pub(super) async fn handle_incoming(xorg: Rc<XconData>, incoming: BufIoIncoming) {
let mut incoming = Incoming {
incoming,
socket: xorg,
ed: None,
};
incoming.run().await;
}
pub struct Incoming {
socket: Rc<XconData>,
incoming: BufIoIncoming,
ed: Option<Rc<ExtensionData>>,
}
impl Incoming {
async fn run(&mut self) {
loop {
if self.socket.dead.get() {
return;
}
if let Err(e) = self.handle_msg().await {
log::error!("Could not process an incoming message: {}", ErrorFmt(e));
self.socket.kill();
return;
}
}
}
#[expect(clippy::await_holding_refcell_ref)] async fn handle_msg(&mut self) -> Result<(), XconError> {
const MAX_LENGTH_UNITS: usize = 0x4000 / 4;
const MIN_MSG_SIZE: usize = 32;
let mut msg_buf = self.socket.in_bufs.pop().unwrap_or_default();
msg_buf.clear();
self.incoming
.fill_msg_buf(MIN_MSG_SIZE, &mut msg_buf)
.await?;
let mut serial = 0;
const KEYMAP_NOTIFY: u8 = 11;
let mut reply_handlers = self.socket.reply_handlers.borrow_mut();
if msg_buf[0] & 0x7f != KEYMAP_NOTIFY {
let serial_16 = u16::from_ne_bytes([msg_buf[2], msg_buf[3]]);
serial = (self.socket.last_recv_serial.get() & !0xffff) | (serial_16 as u64);
if serial < self.socket.last_recv_serial.get() {
serial += 0x10000;
}
self.socket.last_recv_serial.set(serial);
while let Some(first) = reply_handlers.front() {
if first.serial() < serial {
let handler = reply_handlers.pop_front().unwrap();
drop(reply_handlers);
handler.handle_noreply(&self.socket)?;
reply_handlers = self.socket.reply_handlers.borrow_mut();
} else {
break;
}
}
}
if self.ed.is_none() {
self.ed = self.socket.extensions.get();
}
match msg_buf[0] & 0x7f {
0 => 'handle_error: {
let code = msg_buf[1];
let (ext, code) = if code < 128 {
(None, code)
} else if let Some(ed) = &self.ed {
let r = match find_range(&ed.errors, code) {
Some(r) => r,
_ => {
log::error!("Received an out of bounds error code {}", code);
break 'handle_error;
}
};
match r.extension {
Some(e) => (Some(e), code - r.first),
None => {
log::warn!(
"Received an error from an unconfigured extension: `{}`",
r.name
);
break 'handle_error;
}
}
} else {
log::error!("Received an extension error before extension have been fetched");
break 'handle_error;
};
let e = match ext {
Some(e) => XconError::ExtensionError(e, code),
_ => XconError::CoreError(code),
};
if let Some(first) = reply_handlers.front()
&& first.serial() == serial
{
let handler = reply_handlers.pop_front().unwrap();
drop(reply_handlers);
handler.handle_error(e);
break 'handle_error;
}
log::error!(
"Received an error with no corresponding handler: {}",
ErrorFmt(e)
);
}
1 => {
if let Some(first) = reply_handlers.front()
&& first.serial() == serial
{
let handler = reply_handlers.pop_front().unwrap();
drop(reply_handlers);
let mut fds = vec![];
if handler.has_fds() {
let num_fds = msg_buf[1] as usize;
if self.incoming.fds.len() < num_fds {
return Err(XconError::MissingFds);
}
fds.extend(self.incoming.fds.drain(..num_fds));
}
let length =
u32::from_ne_bytes([msg_buf[4], msg_buf[5], msg_buf[6], msg_buf[7]])
as usize;
if length > MAX_LENGTH_UNITS {
return Err(XconError::ExcessiveMessageSize);
}
let length = length * 4;
self.incoming.fill_msg_buf(length, &mut msg_buf).await?;
let mut parser = unsafe {
let msg_buf = mem::transmute::<&[u8], &'static [u8]>(&msg_buf[..]);
Parser::new(msg_buf, fds)
};
handler.handle_result(&self.socket, &mut parser, mem::take(&mut msg_buf))?;
}
}
ev => 'handle_event: {
drop(reply_handlers);
let (ext, code) = if ev == XGE_EVENT {
let length =
u32::from_ne_bytes([msg_buf[4], msg_buf[5], msg_buf[6], msg_buf[7]])
as usize;
if length > MAX_LENGTH_UNITS {
return Err(XconError::ExcessiveMessageSize);
}
let length = length * 4;
self.incoming.fill_msg_buf(length, &mut msg_buf).await?;
let opcode = msg_buf[1];
let ext = match &self.ed {
Some(ed) => ed.ext_by_opcode.get(&opcode),
_ => {
log::error!("Received an XGE event before extension have been fetched");
break 'handle_event;
}
};
let ext = match ext {
Some(ext) => *ext,
_ => {
log::warn!(
"Received an event from an unconfigured extension: `{}`",
opcode
);
break 'handle_event;
}
};
let code = u16::from_ne_bytes([msg_buf[8], msg_buf[9]]);
(Some(ext), code)
} else if ev < 64 {
(None, ev as u16)
} else if let Some(ed) = &self.ed {
let r = match find_range(&ed.events, ev) {
Some(r) => r,
_ => {
log::error!("Received an out of bounds event {}", ev);
break 'handle_event;
}
};
match r.extension {
Some(e) => (Some(e), (ev - r.first) as u16),
None => {
log::warn!(
"Received an event from an unconfigured extension: `{}`",
r.name
);
break 'handle_event;
}
}
} else {
log::error!("Received an extension event before extension have been fetched");
break 'handle_event;
};
self.socket.events.push(Event {
socket: self.socket.clone(),
ext,
code,
buf: mem::take(&mut msg_buf),
serial,
});
}
}
if msg_buf.capacity() > 0 {
self.socket.in_bufs.push(msg_buf);
}
Ok(())
}
}
fn find_range(codes: &[ExtensionIdRange], code: u8) -> Option<&ExtensionIdRange> {
let idx = match codes.binary_search_by_key(&code, |v| v.first) {
Ok(v) => v,
Err(v) if v > 0 => v - 1,
_ => return None,
};
Some(&codes[idx])
}