serialport_stream/
lib.rs

1//! Async serial port I/O as [`AsyncRead`] and [`AsyncWrite`], with optional [`Stream`] support.
2//!
3//! Async runtime agnostic: this crate implements [`futures`] traits only and does not depend on
4//! Tokio, async-std, or any other executor. Use it with any runtime that polls those futures
5//! (Tokio, async-std, [`futures_lite::future::block_on`](https://docs.rs/futures-lite/latest/futures_lite/future/fn.block_on.html), etc.).
6//!
7//! Configure and open ports with [`new`] → [`SerialPortStreamBuilder`] → [`.open()`](SerialPortStreamBuilder::open).
8//! Line settings, DTR, and buffer clearing are applied at open time.
9//!
10//! ## Features
11//!
12//! - **`stream`** (optional): [`Stream`], [`TryStreamExt`], and [`SerialPortStream::try_poll_next`]. Starts a
13//!   background receive pump into an in-memory FIFO (poll-based thread on Unix, `WaitCommEvent`
14//!   thread on Windows). With `stream` enabled, [`AsyncRead`] and [`Stream`] share that FIFO.
15//! - **`tracing`** (optional): diagnostic logs.
16//!
17//! ## Read paths
18//!
19//! | Platform | [`AsyncRead`] (no `stream`) | [`AsyncRead`] + [`Stream`] (`stream` feature) |
20//! | --- | --- | --- |
21//! | Unix | Direct [`async-io`](https://docs.rs/async-io) poll on the port fd | Shared poll-based background thread → FIFO |
22//! | Windows | Overlapped `ReadFile` + windows thread-pool | Shared background read thread → FIFO |
23//!
24//! When `stream` is enabled, [`AsyncRead`] and [`Stream`] share the same background receive FIFO
25//! (Unix and Windows).
26//!
27//! ## Write path
28//!
29//! On Unix, [`AsyncWrite`] uses [`async-io`](https://docs.rs/async-io). On Windows, overlapped
30//! `WriteFile` with a windows thread-pool completion.
31//!
32//! [`AsyncReadExt`] and [`AsyncWriteExt`] are re-exported from `futures`.
33//!
34//! ```no_run
35//! use serialport_stream::{new, AsyncReadExt, AsyncWriteExt};
36//!
37//! # async fn example() -> std::io::Result<()> {
38//! let mut stream = new("/dev/ttyUSB0", 115200).open()?;
39//! stream.write_all(b"PING\r\n").await?;
40//! let mut buf = [0u8; 256];
41//! let n = stream.read(&mut buf).await?;
42//! # let _ = n;
43//! # Ok(())
44//! # }
45//! ```
46
47use std::future::Future;
48use std::pin::Pin;
49#[cfg(feature = "stream")]
50use std::sync::{Arc, Mutex};
51use std::task::{Context, Poll};
52
53mod platform;
54mod types;
55
56pub mod line_settings;
57pub use types::{ClearBuffer, DataBits, FlowControl, Parity, StopBits};
58
59use crate::platform::PlatformStream;
60#[cfg(feature = "stream")]
61use futures::task::AtomicWaker;
62
63#[cfg(feature = "stream")]
64fn clone_io_error(err: &std::io::Error) -> std::io::Error {
65    match err.raw_os_error() {
66        Some(code) => std::io::Error::from_raw_os_error(code),
67        None => std::io::Error::new(err.kind(), err.to_string()),
68    }
69}
70
71pub use futures::io::{AsyncRead, AsyncReadExt};
72pub use futures::io::{AsyncWrite, AsyncWriteExt};
73#[cfg(feature = "stream")]
74pub use futures::stream::{Stream, TryStreamExt};
75
76#[derive(Debug)]
77#[cfg(feature = "stream")]
78pub(crate) struct EventsInnerRead {
79    pub(crate) in_buffer: Mutex<Vec<u8>>,
80    pub(crate) stream_error: Mutex<Option<std::io::Error>>,
81    pub(crate) waker: AtomicWaker,
82}
83
84#[cfg(feature = "stream")]
85impl EventsInnerRead {
86    pub(crate) fn new() -> Self {
87        Self {
88            in_buffer: Mutex::new(Vec::new()),
89            stream_error: Mutex::new(None),
90            waker: AtomicWaker::new(),
91        }
92    }
93}
94
95/// Builder for serial port path, line settings, and one-shot open options.
96///
97/// Created with [`new()`], configured with chained methods, then finalized with
98/// [`open()`](SerialPortStreamBuilder::open).
99///
100/// # Example
101///
102/// ```no_run
103/// use serialport_stream::new;
104/// use serialport_stream::{ClearBuffer, DataBits, FlowControl, Parity, StopBits};
105///
106/// # fn example() -> std::io::Result<()> {
107/// let stream = new("/dev/ttyUSB0", 115200)
108///     .data_bits(DataBits::Eight)
109///     .parity(Parity::None)
110///     .stop_bits(StopBits::One)
111///     .flow_control(FlowControl::None)
112///     .dtr_on_open(true)
113///     .clear(ClearBuffer::All)
114///     .open()?;
115/// # Ok(())
116/// # }
117/// ```
118#[derive(Debug, Clone, PartialEq, Eq)]
119pub struct SerialPortStreamBuilder {
120    pub(crate) path: String,
121    pub(crate) baud_rate: u32,
122    pub(crate) data_bits: DataBits,
123    pub(crate) flow_control: FlowControl,
124    pub(crate) parity: Parity,
125    pub(crate) stop_bits: StopBits,
126    pub(crate) dtr_on_open: bool,
127    pub(crate) clear_buffer: Option<ClearBuffer>,
128}
129
130impl SerialPortStreamBuilder {
131    /// Sets the path to the serial port device.
132    ///
133    /// # Examples
134    /// - Unix: `"/dev/ttyUSB0"`, `"/dev/ttyACM0"`
135    /// - Windows: `"COM3"`, `"COM10"`
136    #[allow(clippy::assigning_clones)]
137    #[must_use]
138    pub fn path<'a>(mut self, path: impl Into<std::borrow::Cow<'a, str>>) -> Self {
139        self.path = path.into().as_ref().to_owned();
140        self
141    }
142
143    /// Sets the baud rate (bits per second).
144    ///
145    /// Common values: 9600, 19200, 38400, 57600, 115200
146    #[must_use]
147    pub fn baud_rate(mut self, baud_rate: u32) -> Self {
148        self.baud_rate = baud_rate;
149        self
150    }
151
152    /// Sets the number of data bits per character.
153    ///
154    /// Default: `DataBits::Eight`
155    #[must_use]
156    pub fn data_bits(mut self, data_bits: DataBits) -> Self {
157        self.data_bits = data_bits;
158        self
159    }
160
161    /// Sets the flow control mode.
162    ///
163    /// Default: `FlowControl::None`
164    #[must_use]
165    pub fn flow_control(mut self, flow_control: FlowControl) -> Self {
166        self.flow_control = flow_control;
167        self
168    }
169
170    /// Sets the parity checking mode.
171    ///
172    /// Default: `Parity::None`
173    #[must_use]
174    pub fn parity(mut self, parity: Parity) -> Self {
175        self.parity = parity;
176        self
177    }
178
179    /// Sets the number of stop bits.
180    ///
181    /// Default: `StopBits::One`
182    #[must_use]
183    pub fn stop_bits(mut self, stop_bits: StopBits) -> Self {
184        self.stop_bits = stop_bits;
185        self
186    }
187
188    /// Sets the DTR (Data Terminal Ready) signal state applied when opening the port.
189    ///
190    /// Default: `false`
191    #[must_use]
192    pub fn dtr_on_open(mut self, state: bool) -> Self {
193        self.dtr_on_open = state;
194        self
195    }
196
197    /// Clears RX and/or TX driver buffers when the port is opened, before async I/O starts.
198    ///
199    /// See [`ClearBuffer`] (`Input`, `Output`, or `All`).
200    #[must_use]
201    pub fn clear(mut self, buffer: ClearBuffer) -> Self {
202        self.clear_buffer = Some(buffer);
203        self
204    }
205
206    /// Opens the serial port and returns a [`SerialPortStream`].
207    ///
208    /// Applies line settings, [`dtr_on_open`](Self::dtr_on_open), and optional
209    /// [`clear`](Self::clear) before async I/O begins.
210    pub fn open(self) -> std::io::Result<SerialPortStream> {
211        #[cfg(feature = "stream")]
212        let read_inner = Arc::new(EventsInnerRead::new());
213        Ok(SerialPortStream {
214            platform: PlatformStream::new(
215                self,
216                #[cfg(feature = "stream")]
217                read_inner.clone(),
218            )?,
219            #[cfg(feature = "stream")]
220            read_inner,
221            flush_task: None,
222            write_in_flight: false,
223        })
224    }
225}
226
227/// Creates a [`SerialPortStreamBuilder`] with default line settings (8N1, no flow control).
228///
229/// # Examples
230///
231/// Unix device path:
232///
233/// ```no_run
234/// # use serialport_stream::new;
235/// # fn example() -> std::io::Result<()> {
236/// let _stream = new("/dev/ttyUSB0", 115200).open()?;
237/// # Ok(())
238/// # }
239/// ```
240///
241/// Windows COM port:
242///
243/// ```no_run
244/// # use serialport_stream::new;
245/// # fn example() -> std::io::Result<()> {
246/// let _stream = new("COM3", 9600).open()?;
247/// # Ok(())
248/// # }
249/// ```
250pub fn new<'a>(
251    path: impl Into<std::borrow::Cow<'a, str>>,
252    baud_rate: u32,
253) -> SerialPortStreamBuilder {
254    SerialPortStreamBuilder {
255        path: path.into().into_owned(),
256        baud_rate,
257        data_bits: DataBits::Eight,
258        flow_control: FlowControl::None,
259        parity: Parity::None,
260        stop_bits: StopBits::One,
261        dtr_on_open: false,
262        clear_buffer: None,
263    }
264}
265
266/// An opened serial port for async reads and writes.
267///
268/// # Read behavior
269///
270/// - **Unix:** [`AsyncRead`] polls the port fd directly via async-io.
271/// - **Windows:** [`AsyncRead`] uses overlapped `ReadFile` with a windows thread-pool completion.
272///
273/// Enable the `stream` feature for [`Stream`] / [`SerialPortStream::try_poll_next`]. That starts a background receive
274/// pump and FIFO on both platforms. With `stream` enabled, [`AsyncRead`] and [`Stream`] both read
275/// from the same FIFO.
276///
277/// # Write behavior
278///
279/// Both platforms use direct async polling (async-io on Unix, overlapped `WriteFile` + windows thread-pool completion).
280///
281/// # Example
282///
283/// ```no_run
284/// use serialport_stream::new;
285/// use futures::io::{AsyncReadExt, AsyncWriteExt};
286///
287/// # async fn example() -> std::io::Result<()> {
288/// let mut stream = new("COM3", 115200).open()?;
289/// stream.write_all(&[0x0a, 0xC0]).await?;
290/// let mut buf = [0u8; 256];
291/// let n = stream.read(&mut buf).await?;
292/// # let _ = n;
293/// # Ok(())
294/// # }
295/// ```
296pub struct SerialPortStream {
297    platform: PlatformStream,
298    #[cfg(feature = "stream")]
299    read_inner: Arc<EventsInnerRead>,
300    flush_task: Option<blocking::Task<std::io::Result<()>>>,
301    write_in_flight: bool,
302}
303
304impl std::fmt::Debug for SerialPortStream {
305    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
306        f.debug_struct("SerialPortStream")
307            .field("platform", &self.platform)
308            .field("flush_task", &self.flush_task.as_ref().map(|_| "..."))
309            .field("write_in_flight", &self.write_in_flight)
310            .finish()
311    }
312}
313
314impl SerialPortStream {
315    #[cfg(feature = "stream")]
316    fn poll_receiver_ready(&mut self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
317        self.read_inner.waker.register(cx.waker());
318
319        if let Some(err) = self.read_inner.stream_error.lock().unwrap().as_ref() {
320            return Poll::Ready(Err(clone_io_error(err)));
321        }
322
323        if !self.platform.is_read_thread_started() {
324            self.platform.start_read_thread();
325            return Poll::Pending;
326        }
327
328        Poll::Ready(Ok(()))
329    }
330
331    #[cfg(feature = "stream")]
332    fn poll_read_fifo(
333        &mut self,
334        cx: &mut Context<'_>,
335        buf: &mut [u8],
336    ) -> Poll<std::io::Result<usize>> {
337        match self.poll_receiver_ready(cx) {
338            Poll::Pending => Poll::Pending,
339            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
340            Poll::Ready(Ok(())) => {
341                let mut buffer = self.read_inner.in_buffer.lock().unwrap();
342                if buffer.is_empty() {
343                    return Poll::Pending;
344                }
345                let n = buffer.len().min(buf.len());
346                buf[..n].copy_from_slice(&buffer[..n]);
347                buffer.drain(..n);
348                #[cfg(feature = "tracing")]
349                {
350                    if buffer.len() > 0 {
351                        tracing::info!(
352                            cached_bytes = buffer.len(),
353                            "serialport-stream FIFO after AsyncRead read"
354                        );
355                    }
356                }
357                Poll::Ready(Ok(n))
358            }
359        }
360    }
361
362    #[cfg(feature = "stream")]
363    /// Polls for the next received chunk, same as [`Stream::poll_next`].
364    ///
365    /// When ready, returns `Poll::Ready(Some(Ok(vec)))` with every byte currently buffered in the
366    /// receive FIFO, or `Poll::Pending` if the pump thread has not yet delivered data.
367    ///
368    /// Requires the `stream` Cargo feature.
369    pub fn try_poll_next(
370        &mut self,
371        cx: &mut Context<'_>,
372    ) -> Poll<Option<Result<Vec<u8>, std::io::Error>>> {
373        match self.poll_receiver_ready(cx) {
374            Poll::Pending => Poll::Pending,
375            Poll::Ready(Err(e)) => Poll::Ready(Some(Err(e))),
376            Poll::Ready(Ok(())) => {
377                let mut buffer = self.read_inner.in_buffer.lock().unwrap();
378                if !buffer.is_empty() {
379                    let data = buffer.drain(..).collect();
380                    return Poll::Ready(Some(Ok(data)));
381                }
382                Poll::Pending
383            }
384        }
385    }
386
387    /// Sets the baud rate (bits per second) on an already-open port.
388    ///
389    /// Other line settings (data bits, parity, stop bits, flow control) remain unchanged.
390    pub fn set_baudrate(&mut self, baud_rate: u32) -> std::io::Result<()> {
391        self.platform.set_baud_rate(baud_rate)
392    }
393}
394
395unsafe impl Send for SerialPortStream {}
396unsafe impl Sync for SerialPortStream {}
397
398impl AsyncRead for SerialPortStream {
399    fn poll_read(
400        mut self: Pin<&mut Self>,
401        cx: &mut Context<'_>,
402        buf: &mut [u8],
403    ) -> Poll<std::io::Result<usize>> {
404        assert!(!buf.is_empty());
405        let this = self.as_mut().get_mut();
406        #[cfg(feature = "stream")]
407        {
408            return this.poll_read_fifo(cx, buf);
409        }
410        #[cfg(not(feature = "stream"))]
411        {
412            this.platform.poll_read(cx, buf)
413        }
414    }
415}
416
417#[cfg(feature = "stream")]
418impl Stream for SerialPortStream {
419    type Item = Result<Vec<u8>, std::io::Error>;
420
421    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
422        self.try_poll_next(cx)
423    }
424}
425
426impl AsyncWrite for SerialPortStream {
427    fn poll_write(
428        mut self: Pin<&mut Self>,
429        cx: &mut Context<'_>,
430        buf: &[u8],
431    ) -> Poll<std::io::Result<usize>> {
432        assert!(!buf.is_empty());
433        let this = self.as_mut().get_mut();
434        let result = this.platform.poll_write(cx, buf);
435        this.write_in_flight = result.is_pending();
436        result
437    }
438
439    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
440        let this = self.as_mut().get_mut();
441
442        if this.write_in_flight {
443            return Poll::Pending;
444        }
445
446        if this.flush_task.is_none() {
447            this.flush_task = Some(this.platform.flush_tx_unblocked());
448        }
449
450        let task = this.flush_task.as_mut().expect("flush task");
451        match Pin::new(task).poll(cx) {
452            Poll::Ready(result) => {
453                this.flush_task = None;
454                Poll::Ready(result)
455            }
456            Poll::Pending => Poll::Pending,
457        }
458    }
459
460    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
461        self.poll_flush(cx)
462    }
463}
serialport_stream/lib.rs

serialport_stream/
lib.rs
