aen_window/
client.rs

1use crate::state::Session;
2use crate::wire::{Decoder, Message, Outgoing};
3use crate::*;
4use std::os::fd::{AsRawFd, RawFd};
5use std::os::unix::net::UnixStream;
6use std::time::Instant;
7
8#[derive(Clone, Debug, Eq, PartialEq)]
9pub enum Event {
10    Configured(Configure),
11    FrameAccepted {
12        frame_serial: u64,
13        configure_serial: u64,
14    },
15    FrameDiscarded {
16        frame_serial: u64,
17        configure_serial: u64,
18    },
19    Input(Input),
20    CloseRequested,
21}
22/// Nonblocking transport. Prefer WorkerWindow when application work can block.
23pub struct Client {
24    stream: UnixStream,
25    state: Session,
26    decoder: Decoder,
27    outgoing: Option<Outgoing>,
28    deadline: Option<Instant>,
29    partial_since: Option<Instant>,
30    failed: bool,
31    options: ClientOptions,
32    options_negotiated: bool,
33}
34impl Client {
35    pub fn connect(path: &Path, create: Create) -> io::Result<Self> {
36        Self::connect_with_options(path, create, ClientOptions::default())
37    }
38    /// Opt in explicitly. Unsupported servers close the handshake; this SDK
39    /// never reconnects or silently replays a create without the requested option.
40    pub fn connect_with_options(
41        path: &Path,
42        create: Create,
43        options: ClientOptions,
44    ) -> io::Result<Self> {
45        create.validate()?;
46        let stream = crate::platform::connect_local(path)?;
47        Ok(Self {
48            stream,
49            state: Session::default(),
50            decoder: Decoder::new(false, None),
51            outgoing: Some(Outgoing::new(if options == ClientOptions::default() {
52                Message::Create(create)
53            } else {
54                Message::CreateWithOptions(create, options)
55            })?),
56            deadline: Some(Instant::now() + TRANSFER_TIMEOUT),
57            partial_since: None,
58            failed: false,
59            options,
60            options_negotiated: false,
61        })
62    }
63    /// True only after the requested extension was echoed by a valid configure.
64    pub fn relative_pointer_enabled(&self) -> bool {
65        !self.failed && self.options_negotiated && self.options.relative_pointer
66    }
67    pub fn configuration(&self) -> Option<Configure> {
68        self.state.configuration()
69    }
70    pub fn can_submit(&self) -> bool {
71        !self.failed
72            && self.configuration().is_some()
73            && self.state.inflight.is_none()
74            && self.outgoing.is_none()
75    }
76    pub fn wants_write(&self) -> bool {
77        self.outgoing.is_some()
78    }
79    pub fn submit(&mut self, frame: Frame) -> io::Result<()> {
80        if self.failed {
81            return Err(io::ErrorKind::BrokenPipe.into());
82        }
83        if self.outgoing.is_some() {
84            return Err(io::ErrorKind::WouldBlock.into());
85        }
86        self.state.validate_submit(&frame)?;
87        let outgoing = Outgoing::new(Message::Frame(frame.clone()))?;
88        self.state.submit(&frame)?;
89        self.outgoing = Some(outgoing);
90        self.deadline = Some(Instant::now() + TRANSFER_TIMEOUT);
91        Ok(())
92    }
93    pub fn poll(&mut self) -> io::Result<Vec<Event>> {
94        if self.failed {
95            return Err(io::ErrorKind::BrokenPipe.into());
96        }
97        let result = self.poll_inner();
98        if result.is_err() {
99            self.failed = true;
100            self.outgoing = None;
101        }
102        result
103    }
104    fn poll_inner(&mut self) -> io::Result<Vec<Event>> {
105        let now = Instant::now();
106        if self.deadline.is_some_and(|d| now >= d)
107            || self
108                .partial_since
109                .is_some_and(|start| now.duration_since(start) >= TRANSFER_TIMEOUT)
110        {
111            return Err(io::ErrorKind::TimedOut.into());
112        }
113        let upload_complete = self
114            .outgoing
115            .as_mut()
116            .map(|outgoing| outgoing.flush(&mut self.stream, 1024 * 1024))
117            .transpose()?
118            .unwrap_or(false);
119        if upload_complete {
120            self.outgoing = None;
121        }
122        let mut events = Vec::new();
123        for _ in 0..64 {
124            let Some(message) = self.decoder.poll(&mut self.stream, 4096)? else {
125                break;
126            };
127            self.partial_since = None;
128            let event = match message {
129                Message::Configure(c) if self.options == ClientOptions::default() => {
130                    self.state.configure(c)?;
131                    if self.state.inflight.is_none() {
132                        self.deadline = None;
133                    }
134                    Event::Configured(c)
135                }
136                Message::ConfigureWithOptions(c, options)
137                    if self.options != ClientOptions::default() && options == self.options =>
138                {
139                    self.state.configure(c)?;
140                    self.options_negotiated = true;
141                    if self.state.inflight.is_none() {
142                        self.deadline = None;
143                    }
144                    Event::Configured(c)
145                }
146                Message::Accepted(frame_serial, configure_serial)
147                | Message::Discarded(frame_serial, configure_serial) => {
148                    if self.outgoing.is_some() {
149                        return Err(invalid("native window ACK precedes complete upload"));
150                    }
151                    let accepted = matches!(message, Message::Accepted(..));
152                    self.state
153                        .acknowledge(frame_serial, configure_serial, accepted)?;
154                    self.deadline = None;
155                    if accepted {
156                        Event::FrameAccepted {
157                            frame_serial,
158                            configure_serial,
159                        }
160                    } else {
161                        Event::FrameDiscarded {
162                            frame_serial,
163                            configure_serial,
164                        }
165                    }
166                }
167                Message::Input(input) => {
168                    if matches!(input.event, InputEvent::RelativeMotion { .. })
169                        && !self.relative_pointer_enabled()
170                    {
171                        return Err(invalid("unnegotiated native relative pointer input"));
172                    }
173                    self.state.input(input.clone())?;
174                    Event::Input(input)
175                }
176                Message::CloseRequested => Event::CloseRequested,
177                _ => return Err(invalid("unexpected native window server message")),
178            };
179            events.push(event);
180        }
181        if self.decoder.partial() {
182            self.partial_since.get_or_insert(now);
183        } else {
184            self.partial_since = None;
185        }
186        Ok(events)
187    }
188}
189impl AsRawFd for Client {
190    fn as_raw_fd(&self) -> RawFd {
191        self.stream.as_raw_fd()
192    }
193}
aen_window/client.rs

aen_window/
client.rs
