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 pointer_confinement_enabled(&self) -> bool {
68        !self.failed && self.options_negotiated && self.options.pointer_confinement
69    }
70    pub fn pointer_confined(&self) -> bool {
71        !self.failed && self.state.confinement_active.is_some()
72    }
73    /// Request/release confinement; success queues a request, never a grant.
74    /// Only a matching PointerConfinement input event reports the decision.
75    pub fn request_pointer_confinement(&mut self, confined: bool) -> io::Result<u64> {
76        if !self.pointer_confinement_enabled() {
77            return Err(invalid("native pointer confinement was not negotiated"));
78        }
79        if self.outgoing.is_some() {
80            return Err(io::ErrorKind::WouldBlock.into());
81        }
82        let request = self.state.next_confinement_request(confined)?;
83        let outgoing = Outgoing::new(Message::PointerConfinement(request))?;
84        self.state.request_confinement(request)?;
85        self.outgoing = Some(outgoing);
86        self.deadline = Some(Instant::now() + TRANSFER_TIMEOUT);
87        Ok(request.request_serial)
88    }
89    pub fn configuration(&self) -> Option<Configure> {
90        self.state.configuration()
91    }
92    pub fn can_submit(&self) -> bool {
93        !self.failed
94            && self.configuration().is_some()
95            && !self.state.credit_occupied()
96            && self.outgoing.is_none()
97    }
98    pub fn wants_write(&self) -> bool {
99        self.outgoing.is_some()
100    }
101    pub fn submit(&mut self, frame: Frame) -> io::Result<()> {
102        if self.failed {
103            return Err(io::ErrorKind::BrokenPipe.into());
104        }
105        if self.outgoing.is_some() {
106            return Err(io::ErrorKind::WouldBlock.into());
107        }
108        self.state.validate_submit(&frame)?;
109        let outgoing = Outgoing::new(Message::Frame(frame.clone()))?;
110        self.state.submit(&frame)?;
111        self.outgoing = Some(outgoing);
112        self.deadline = Some(Instant::now() + TRANSFER_TIMEOUT);
113        Ok(())
114    }
115    pub fn poll(&mut self) -> io::Result<Vec<Event>> {
116        if self.failed {
117            return Err(io::ErrorKind::BrokenPipe.into());
118        }
119        let result = self.poll_inner();
120        if result.is_err() {
121            self.failed = true;
122            self.outgoing = None;
123        }
124        result
125    }
126    fn poll_inner(&mut self) -> io::Result<Vec<Event>> {
127        let now = Instant::now();
128        if self.deadline.is_some_and(|d| now >= d)
129            || self
130                .partial_since
131                .is_some_and(|start| now.duration_since(start) >= TRANSFER_TIMEOUT)
132        {
133            return Err(io::ErrorKind::TimedOut.into());
134        }
135        let upload_complete = self
136            .outgoing
137            .as_mut()
138            .map(|outgoing| outgoing.flush(&mut self.stream, 1024 * 1024))
139            .transpose()?
140            .unwrap_or(false);
141        if upload_complete {
142            self.outgoing = None;
143        }
144        let mut events = Vec::new();
145        for _ in 0..64 {
146            let Some(message) = self.decoder.poll(&mut self.stream, 4096)? else {
147                break;
148            };
149            self.partial_since = None;
150            let event = match message {
151                Message::Configure(c) if self.options == ClientOptions::default() => {
152                    self.state.configure(c)?;
153                    if !self.state.credit_occupied() {
154                        self.deadline = None;
155                    }
156                    Event::Configured(c)
157                }
158                Message::ConfigureWithOptions(c, options)
159                    if self.options != ClientOptions::default() && options == self.options =>
160                {
161                    self.state.configure(c)?;
162                    self.options_negotiated = true;
163                    if !self.state.credit_occupied() {
164                        self.deadline = None;
165                    }
166                    Event::Configured(c)
167                }
168                Message::Accepted(frame_serial, configure_serial)
169                | Message::Discarded(frame_serial, configure_serial) => {
170                    if self.outgoing.is_some() {
171                        return Err(invalid("native window ACK precedes complete upload"));
172                    }
173                    let accepted = matches!(message, Message::Accepted(..));
174                    self.state
175                        .acknowledge(frame_serial, configure_serial, accepted)?;
176                    self.deadline = None;
177                    if accepted {
178                        Event::FrameAccepted {
179                            frame_serial,
180                            configure_serial,
181                        }
182                    } else {
183                        Event::FrameDiscarded {
184                            frame_serial,
185                            configure_serial,
186                        }
187                    }
188                }
189                Message::Input(input) => {
190                    if matches!(input.event, InputEvent::RelativeMotion { .. })
191                        && !self.relative_pointer_enabled()
192                    {
193                        return Err(invalid("unnegotiated native relative pointer input"));
194                    }
195                    let reply = if let InputEvent::PointerConfinement { request_serial, .. } =
196                        input.event
197                    {
198                        if !self.pointer_confinement_enabled() {
199                            return Err(invalid("unnegotiated native pointer confinement input"));
200                        }
201                        self.state
202                            .confinement_pending
203                            .is_some_and(|request| request.request_serial == request_serial)
204                    } else {
205                        false
206                    };
207                    if reply && self.outgoing.is_some() {
208                        return Err(invalid("native pointer reply precedes complete request"));
209                    }
210                    self.state.input(input.clone())?;
211                    if reply {
212                        self.deadline = None;
213                    }
214                    Event::Input(input)
215                }
216                Message::CloseRequested => Event::CloseRequested,
217                _ => return Err(invalid("unexpected native window server message")),
218            };
219            events.push(event);
220        }
221        if self.decoder.partial() {
222            self.partial_since.get_or_insert(now);
223        } else {
224            self.partial_since = None;
225        }
226        Ok(events)
227    }
228}
229impl AsRawFd for Client {
230    fn as_raw_fd(&self) -> RawFd {
231        self.stream.as_raw_fd()
232    }
233}
aen_window/client.rs

aen_window/
client.rs
