diff --git a/src/client.rs b/src/client.rs index 3d68f22..91b1e7e 100644 --- a/src/client.rs +++ b/src/client.rs @@ -101,7 +101,7 @@ impl AsyncClient { R: futures::AsyncRead + Send + Unpin, W: futures::AsyncWrite + Send + Unpin, { - use futures::{channel::mpsc, StreamExt}; + use futures::{channel::mpsc, AsyncWriteExt, StreamExt}; let (event_tx, event_recv) = mpsc::unbounded::(); let (req_tx, mut req_recv) = mpsc::unbounded::>(); @@ -116,7 +116,7 @@ impl AsyncClient { req_opt = req_recv.next() => match req_opt { Some(req) => { let raw_req = state.track_request(&mut next_id, req); - crate::io::async_write(&mut writer, raw_req).await?; + writer.write_all(&crate::io::encode(raw_req)).await?; }, None => break, }, @@ -338,7 +338,7 @@ impl BlockingClient { let mut next_id = 0_u32; for req in req_recv { let raw_req = write_state.lock().unwrap().track_request(&mut next_id, req); - crate::io::blocking_write(&mut writer, raw_req)?; + std::io::Write::write_all(&mut writer, &crate::io::encode(raw_req))?; } Ok(()) }); diff --git a/src/io.rs b/src/io.rs index 58a83ec..9008e63 100644 --- a/src/io.rs +++ b/src/io.rs @@ -2,6 +2,8 @@ //! //! This module provides core types and functions for serializing and deserializing Electrum //! messages—including requests, responses, and notifications—over arbitrary transports. +//! +//! [`Decoder`] and [`encode`] do no I/O, so a custom transport only needs to move bytes. use std::{ collections::VecDeque, @@ -11,11 +13,78 @@ use std::{ use crate::{RawIncoming, RawOneOrMany, RawRequest}; +/// A sans-io decoder for Electrum's newline-delimited JSON-RPC messages. +/// +/// Feed it bytes as they arrive with [`Decoder::feed`], then iterate it for the [`RawIncoming`]s +/// in the complete lines received so far. A batch yields one item per message. A line that isn't +/// valid JSON-RPC yields an error and is skipped. +/// +/// ```rust +/// use electrum_streaming_client::io::Decoder; +/// +/// let mut decoder = Decoder::new(); +/// decoder.feed(b"{\"jsonrpc\":\"2.0\",\"method\":\"blockchain.headers.subscribe\","); +/// assert!(decoder.next().is_none()); +/// decoder.feed(b"\"params\":[]}\n"); +/// assert!(decoder.next().unwrap().is_ok()); +/// ``` +#[derive(Debug, Default)] +pub struct Decoder { + buf: Vec, + // Length of the start of `buf` known to hold no newline. + scanned: usize, + queue: VecDeque>, +} + +impl Decoder { + /// Creates an empty decoder. + pub fn new() -> Self { + Self::default() + } + + /// Appends bytes received from the server. + pub fn feed(&mut self, bytes: &[u8]) { + self.buf.extend_from_slice(bytes); + } + + /// Returns true if part of a line is buffered (a message cut off, if the input ended). + pub fn has_partial_line(&self) -> bool { + !self.buf.iter().all(u8::is_ascii_whitespace) + } +} + +impl Iterator for Decoder { + type Item = Result; + + fn next(&mut self) -> Option { + if self.queue.is_empty() { + // Decode every complete line at once, so the buffer is drained once per feed. + let Some(i) = self.buf[self.scanned..].iter().rposition(|&b| b == b'\n') else { + self.scanned = self.buf.len(); + return None; + }; + let end = self.scanned + i + 1; + for line in self.buf[..end].split(|&b| b == b'\n') { + // Skip empty lines, including the empty slice after the final `\n`. + if line.iter().all(u8::is_ascii_whitespace) { + continue; + } + match serde_json::from_slice::>(line) { + Ok(msgs) => self.queue.extend(msgs.into_vec().into_iter().map(Ok)), + Err(err) => self.queue.push_back(Err(err)), + } + } + self.buf.drain(..end); + self.scanned = 0; + } + self.queue.pop_front() + } +} + /// A streaming parser for Electrum JSON-RPC messages from an input reader. /// -/// `ReadStreamer` incrementally reads from a source implementing [`std::io::BufRead`] or -/// [`futures::io::AsyncBufRead`] (depending on the API used), parses incoming JSON-RPC payloads, and -/// queues deserialized [`RawIncoming`] items for consumption. +/// `ReadStreamer` reads from a source implementing [`std::io::BufRead`] or +/// [`futures::io::AsyncBufRead`] (depending on the API used) and decodes it with a [`Decoder`]. /// /// ### Behavior /// @@ -24,6 +93,9 @@ use crate::{RawIncoming, RawOneOrMany, RawRequest}; /// - For **async transports**, `ReadStreamer` implements [`futures::Stream`], with the same item /// type. /// +/// It ends when the reader ends or fails. If the reader ends partway through a line, the last item +/// is an [`UnexpectedEof`](std::io::ErrorKind::UnexpectedEof) error. +/// /// ### Examples /// /// **Blocking I/O** @@ -62,9 +134,7 @@ use crate::{RawIncoming, RawOneOrMany, RawRequest}; #[derive(Debug)] pub struct ReadStreamer { reader: Option, - buf: Vec, - queue: VecDeque, - err: Option, + decoder: Decoder, } impl ReadStreamer { @@ -75,27 +145,19 @@ impl ReadStreamer { pub fn new(reader: R) -> Self { Self { reader: Some(reader), - buf: Vec::new(), - queue: VecDeque::new(), - err: None, + decoder: Decoder::new(), } } - fn _enqueue_from_buf(&mut self) -> bool { - match self.buf.pop() { - Some(b) => assert_eq!(b, b'\n'), - None => return false, + fn _end(&mut self, err: Option) -> Option> { + self.reader = None; + if err.is_none() && !self.decoder.has_partial_line() { + return None; } - match serde_json::from_slice::>(&self.buf) { - Ok(RawOneOrMany::Single(t)) => self.queue.push_back(t), - Ok(RawOneOrMany::Batch(v)) => self.queue.extend(v), - Err(err) => { - self.err = Some(err.into()); - return false; - } - }; - self.buf.clear(); - true + self.decoder = Decoder::new(); + Some(Err( + err.unwrap_or_else(|| std::io::ErrorKind::UnexpectedEof.into()) + )) } } @@ -104,19 +166,19 @@ impl Iterator for ReadStreamer { fn next(&mut self) -> Option { loop { - if let Some(item) = self.queue.pop_front() { - return Some(Ok(item)); - } - if let Some(err) = self.err.take() { - return Some(Err(err)); + if let Some(item) = self.decoder.next() { + return Some(item.map_err(Into::into)); } - let mut reader = self.reader.take()?; - if let Err(err) = reader.read_until(b'\n', &mut self.buf) { - self.err = Some(err); - continue; - } - if self._enqueue_from_buf() { - self.reader = Some(reader); + let reader = self.reader.as_mut()?; + match reader.fill_buf() { + Ok([]) => return self._end(None), + Ok(bytes) => { + let n = bytes.len(); + self.decoder.feed(bytes); + reader.consume(n); + } + Err(err) if err.kind() == std::io::ErrorKind::Interrupted => {} + Err(err) => return self._end(Some(err)), } } } @@ -125,95 +187,91 @@ impl Iterator for ReadStreamer { impl futures::Stream for ReadStreamer { type Item = std::io::Result; - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - use futures::AsyncBufReadExt; - use futures::FutureExt; - Poll::Ready(loop { - if let Some(item) = self.queue.pop_front() { - break Some(Ok(item)); - } - if let Some(err) = self.err.take() { - break Some(Err(err)); + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.get_mut(); + loop { + if let Some(item) = this.decoder.next() { + return Poll::Ready(Some(item.map_err(Into::into))); } - let mut reader = match self.reader.take() { - Some(r) => r, - None => break None, + let reader = match this.reader.as_mut() { + Some(reader) => reader, + None => return Poll::Ready(None), }; - match reader.read_until(b'\n', &mut self.buf).poll_unpin(cx) { - Poll::Ready(Err(err)) => { - self.err = Some(err); - continue; - } - Poll::Ready(Ok(_)) => { - if self._enqueue_from_buf() { - self.reader = Some(reader); - } - } - Poll::Pending => { - self.reader = Some(reader); - return Poll::Pending; + match futures::ready!(Pin::new(&mut *reader).poll_fill_buf(cx)) { + Ok([]) => return Poll::Ready(this._end(None)), + Ok(bytes) => { + let n = bytes.len(); + this.decoder.feed(bytes); + Pin::new(reader).consume(n); } + Err(err) if err.kind() == std::io::ErrorKind::Interrupted => {} + Err(err) => return Poll::Ready(this._end(Some(err))), } - }) + } } } -/// Writes a JSON-RPC request or batch to a blocking writer, followed by a newline. -/// -/// The message is serialized using `serde_json` and written as a single line, -/// terminated by `\n`, to comply with Electrum's line-delimited JSON-RPC protocol. -/// -/// Returns an error if writing to the underlying writer fails. -/// -/// # Parameters -/// - `writer`: A blocking writer implementing [`std::io::Write`]. -/// - `msg`: A single or batched [`RawRequest`] to be serialized. -/// -/// # Errors -/// Returns a [`std::io::Error`] if the write operation fails. -pub fn blocking_write(mut writer: W, msg: T) -> std::io::Result<()> -where - T: Into>, - W: std::io::Write, -{ +/// Encodes a JSON-RPC request or batch as a newline-terminated line, as Electrum's +/// line-delimited JSON-RPC protocol expects. +pub fn encode>>(msg: T) -> Vec { let mut b = serde_json::to_vec(&msg.into()).expect("must serialize"); b.push(b'\n'); - writer.write_all(&b) + b } -/// Asynchronously writes a JSON-RPC request or batch to an async writer, followed by a newline. -/// -/// The message is serialized using `serde_json` and written as a single line terminated by `\n`, -/// following Electrum's line-delimited JSON-RPC protocol. -/// -/// # Parameters -/// - `writer`: An async writer implementing [`futures::io::AsyncWrite`] + [`Unpin`]. -/// - `msg`: A single or batched [`RawRequest`] to be serialized. -/// -/// # Errors -/// Returns a [`std::io::Error`] if the async write operation fails. -pub async fn async_write(mut writer: W, msg: T) -> std::io::Result<()> -where - T: Into>, - W: futures::AsyncWrite + Unpin, -{ - use futures::AsyncWriteExt; - let mut b = serde_json::to_vec(&msg.into()).expect("must serialize"); - b.push(b'\n'); - writer.write_all(&b).await -} +#[cfg(test)] +mod tests { + use super::*; -/// Asynchronously writes a JSON-RPC request or batch to a tokio async writer, followed by a newline. -/// -/// This is the `"tokio"` version of [`async_write`]. -#[cfg(feature = "tokio")] -pub async fn tokio_write(mut writer: W, msg: T) -> std::io::Result<()> -where - T: Into>, - W: tokio::io::AsyncWrite + Unpin, -{ - use tokio::io::AsyncWriteExt; - let mut b = serde_json::to_vec(&msg.into()).expect("must serialize"); - b.push(b'\n'); - writer.write_all(&b).await + const NOTIFICATION: &[u8] = + b"{\"jsonrpc\":\"2.0\",\"method\":\"blockchain.headers.subscribe\",\"params\":[]}\n"; + + #[test] + fn decoder_splits_lines_across_feeds() { + let mut input = NOTIFICATION.to_vec(); + input.extend_from_slice(b"\r\n[{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":5},"); + input.extend_from_slice(b"{\"jsonrpc\":\"2.0\",\"id\":2,\"error\":\"no\"}]\nnot json\n"); + input.extend_from_slice(NOTIFICATION); + + // One byte at a time, and all lines in one feed. + for chunk_len in [1, input.len()] { + let mut decoder = Decoder::new(); + let mut out = Vec::new(); + for chunk in input.chunks(chunk_len) { + decoder.feed(chunk); + out.extend(&mut decoder); + } + assert!(!decoder.has_partial_line()); + assert!(matches!(out[0], Ok(RawIncoming::Notification(_)))); + assert!(matches!(out[1], Ok(RawIncoming::Response(ref r)) if r.id == 1)); + assert!(matches!(out[2], Ok(RawIncoming::Response(ref r)) if r.id == 2)); + assert!(out[3].is_err()); + assert!(matches!(out[4], Ok(RawIncoming::Notification(_)))); + assert_eq!(out.len(), 5); + } + } + + #[test] + fn read_streamer_errors_on_cut_off_line() { + let input = &NOTIFICATION[..NOTIFICATION.len() - 1]; + let out = ReadStreamer::new(input).collect::>(); + assert_eq!(out.len(), 1); + assert_eq!( + out[0].as_ref().unwrap_err().kind(), + std::io::ErrorKind::UnexpectedEof + ); + } + + #[test] + fn read_streamer_async_matches_blocking() { + use futures::StreamExt; + let mut input = NOTIFICATION.repeat(2); + input.extend_from_slice(b"{"); + let out = futures::executor::block_on( + ReadStreamer::new(futures::io::Cursor::new(input)).collect::>(), + ); + assert_eq!(out.len(), 3); + assert!(out[..2].iter().all(|r| r.is_ok())); + assert!(out[2].is_err()); + } }