feat: client connect over tcp and tls

Cargo.toml

@@ -4,27 +4,32 @@ version = "0.4.0"
 description = "Experimental but sane electrum client by @evanlinjin."
 license = "MIT OR Apache-2.0"
 edition = "2021"
-rust-version = "1.70"
+rust-version = "1.71"
 repository = "https://github.com/bitcoindevkit/electrum_streaming_client"
 documentation = "https://docs.rs/electrum_streaming_client"
 readme = "README.md"
 
+[package.metadata.docs.rs]
+all-features = true
+
 [dependencies]
 futures = "0.3"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 bitcoin = { version = "0.32", features = ["serde"] }
 
-tokio = { version = "1.44.2", features = ["io-util"],  optional = true }
+tokio = { version = "1.44.2", features = ["io-util", "net", "time"], optional = true }
 tokio-util = { version = "0.7.15", features = ["compat"], optional = true }
+tokio-rustls = { version = "0.26", optional = true }
+rustls = { version = "0.23", optional = true }
+webpki-roots = { version = "1", optional = true }
 
 [features]
 default = ["tokio"]
 tokio = ["dep:tokio", "tokio-util"]
+ssl = ["dep:rustls", "dep:webpki-roots", "dep:tokio-rustls"]
 
 [dev-dependencies]
-async-std = "1.13.0"
 bdk_testenv = "0.11"
-futures = { version = "0.3", features = ["thread-pool"] }
 tokio = { version = "1.44.2", features = ["full"] }
 anyhow = "1.0.98"

README.md

@@ -16,34 +16,34 @@ models.
 ## Example (async with Tokio)
 
 ```rust,no_run
-use electrum_streaming_client::{AsyncClient, Event};
-use tokio::net::TcpStream;
+use electrum_streaming_client::{request, AsyncClient, ConnectConfig, ServerUrl};
 use futures::StreamExt;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let stream = TcpStream::connect("127.0.0.1:50001").await?;
-    let (reader, writer) = stream.into_split();
-    let (client, mut events, worker) = AsyncClient::new_tokio(reader, writer);
+    let url: ServerUrl = "ssl://electrum.blockstream.info:50002".parse()?;
+    let (client, mut events, worker) =
+        AsyncClient::connect(&url, &ConnectConfig::default()).await?;
+    let worker = tokio::spawn(worker);
 
-    tokio::spawn(worker); // spawn the client worker task
-
-    let relay_fee = client.send_request(electrum_streaming_client::request::RelayFee).await?;
+    let relay_fee = client.send_request(request::RelayFee).await?;
     println!("Relay fee: {relay_fee:?}");
 
-    while let Some(event) = events.next().await {
-        println!("Event: {event:?}");
-    }
+    client.send_event_request(request::HeadersSubscribe)?;
+    println!("Event: {:?}", events.next().await);
+
+    drop(client);
+    worker.await??;
 
     Ok(())
 }
 ```
 
 ## Optional Features
 
-- `tokio`: Enables [`AsyncClient::new_tokio`] for use with Tokio-compatible streams.
+- `tokio` (default): Enables Tokio transport support.
+- `ssl`: Enables TLS via rustls. Async TLS additionally requires `tokio`.
 
 ## License
 
 MIT
-

src/client.rs

@@ -1,3 +1,5 @@
+use std::{future::Future, io, thread::JoinHandle, time::Duration};
+
 use crate::pending_request::{PendingRequest, RequestExt};
 use crate::*;
 
@@ -94,7 +96,7 @@ impl AsyncClient {
     ) -> (
         Self,
         AsyncEventReceiver,
-        impl std::future::Future<Output = std::io::Result<()>> + Send,
+        impl Future<Output = io::Result<()>> + Send,
     )
     where
         R: futures::AsyncRead + Send + Unpin,
@@ -123,7 +125,7 @@ impl AsyncClient {
                         Some(incoming_res) => {
                             let event_opt = state
                                 .handle_incoming(incoming_res?)
-                                .map_err(|error| std::io::Error::new(std::io::ErrorKind::Other, error))?;
+                                .map_err(|error| io::Error::new(io::ErrorKind::Other, error))?;
                             if let Some(event) = event_opt {
                                 if let Err(_err) = event_tx.unbounded_send(event) {
                                     break;
@@ -134,7 +136,7 @@ impl AsyncClient {
                     }
                 }
             }
-            std::io::Result::<()>::Ok(())
+            io::Result::<()>::Ok(())
         };
 
         (Self { tx: req_tx }, event_recv, fut)
@@ -164,7 +166,7 @@ impl AsyncClient {
     ) -> (
         Self,
         AsyncEventReceiver,
-        impl std::future::Future<Output = std::io::Result<()>> + Send,
+        impl Future<Output = io::Result<()>> + Send,
     )
     where
         R: tokio::io::AsyncRead + Send + Unpin,
@@ -179,6 +181,85 @@ impl AsyncClient {
         self.tx.close_channel();
     }
 
+    /// Creates a new [`AsyncClient`] connected to `addr` over plaintext TCP via Tokio.
+    #[cfg(feature = "tokio")]
+    pub async fn connect_tcp(
+        addr: &crate::transport::ServerAddr,
+        timeout: Option<Duration>,
+    ) -> io::Result<(
+        Self,
+        AsyncEventReceiver,
+        impl Future<Output = io::Result<()>> + Send,
+    )> {
+        let stream = crate::transport::tokio::connect_tcp(addr, timeout).await?;
+        let (reader, writer) = tokio::io::split(stream);
+        Ok(Self::new_tokio(reader, writer))
+    }
+
+    /// Creates a new [`AsyncClient`] connected to `addr` over TLS via Tokio.
+    ///
+    /// `timeout` bounds DNS, TCP connect, and the TLS handshake.
+    /// `validate_domain` requires a domain host.
+    #[cfg(all(feature = "ssl", feature = "tokio"))]
+    pub async fn connect_ssl(
+        addr: &crate::transport::ServerAddr,
+        validate_domain: bool,
+        timeout: Option<Duration>,
+    ) -> Result<
+        (
+            Self,
+            AsyncEventReceiver,
+            impl Future<Output = io::Result<()>> + Send,
+        ),
+        crate::ConnectError,
+    > {
+        let stream = crate::transport::tokio::connect_ssl(addr, validate_domain, timeout).await?;
+        let (reader, writer) = tokio::io::split(stream);
+        Ok(Self::new_tokio(reader, writer))
+    }
+
+    /// Connects to `url` using its scheme.
+    ///
+    /// A missing scheme defaults to plaintext TCP; `ssl://` requires the `ssl` feature.
+    #[cfg(feature = "tokio")]
+    pub async fn connect(
+        url: &crate::transport::ServerUrl,
+        config: &crate::transport::ConnectConfig,
+    ) -> Result<
+        (
+            Self,
+            AsyncEventReceiver,
+            impl Future<Output = io::Result<()>> + Send,
+        ),
+        crate::ConnectError,
+    > {
+        fn box_worker<F>(
+            worker: F,
+        ) -> std::pin::Pin<Box<dyn Future<Output = io::Result<()>> + Send>>
+        where
+            F: Future<Output = io::Result<()>> + Send + 'static,
+        {
+            Box::pin(worker)
+        }
+
+        match url.scheme() {
+            crate::transport::Scheme::Tcp => {
+                let (client, events, worker) =
+                    Self::connect_tcp(url.addr(), config.timeout()).await?;
+                Ok((client, events, box_worker(worker)))
+            }
+            #[cfg(feature = "ssl")]
+            crate::transport::Scheme::Ssl => {
+                let (client, events, worker) =
+                    Self::connect_ssl(url.addr(), config.validate_domain(), config.timeout())
+                        .await?;
+                Ok((client, events, box_worker(worker)))
+            }
+            #[cfg(not(feature = "ssl"))]
+            crate::transport::Scheme::Ssl => Err(crate::ConnectError::SslUnsupported),
+        }
+    }
+
     /// Sends a single tracked request to the Electrum server and awaits the response.
     ///
     /// This method is for request–response style interactions where only a single result is
@@ -304,27 +385,27 @@ impl BlockingClient {
     ) -> (
         Self,
         BlockingEventReceiver,
-        std::thread::JoinHandle<std::io::Result<()>>,
-        std::thread::JoinHandle<std::io::Result<()>>,
+        JoinHandle<io::Result<()>>,
+        JoinHandle<io::Result<()>>,
     )
     where
-        R: std::io::Read + Send + 'static,
-        W: std::io::Write + Send + 'static,
+        R: io::Read + Send + 'static,
+        W: io::Write + Send + 'static,
     {
         use std::sync::mpsc::*;
         let (event_tx, event_recv) = channel::<Event>();
         let (req_tx, req_recv) = channel::<RawOneOrMany<PendingRequest>>();
-        let incoming_stream = crate::io::ReadStreamer::new(std::io::BufReader::new(reader));
+        let incoming_stream = crate::io::ReadStreamer::new(io::BufReader::new(reader));
         let read_state = std::sync::Arc::new(std::sync::Mutex::new(RequestTracker::new()));
         let write_state = std::sync::Arc::clone(&read_state);
 
-        let read_join = std::thread::spawn(move || -> std::io::Result<()> {
+        let read_join = std::thread::spawn(move || -> io::Result<()> {
             for incoming_res in incoming_stream {
                 let event_opt = read_state
                     .lock()
                     .unwrap()
                     .handle_incoming(incoming_res?)
-                    .map_err(|error| std::io::Error::new(std::io::ErrorKind::Other, error))?;
+                    .map_err(|error| io::Error::new(io::ErrorKind::Other, error))?;
                 if let Some(event) = event_opt {
                     if let Err(_err) = event_tx.send(event) {
                         break;
@@ -333,7 +414,7 @@ impl BlockingClient {
             }
             Ok(())
         });
-        let write_join = std::thread::spawn(move || -> std::io::Result<()> {
+        let write_join = std::thread::spawn(move || -> io::Result<()> {
             let mut next_id = 0_u32;
             for req in req_recv {
                 let raw_req = write_state.lock().unwrap().track_request(&mut next_id, req);
@@ -344,6 +425,75 @@ impl BlockingClient {
         (Self { tx: req_tx }, event_recv, read_join, write_join)
     }
 
+    /// Connects to `url` using its scheme.
+    ///
+    /// A missing scheme defaults to plaintext TCP; `ssl://` requires the `ssl` feature.
+    #[allow(clippy::type_complexity)]
+    pub fn connect(
+        url: &crate::transport::ServerUrl,
+        config: &crate::transport::ConnectConfig,
+    ) -> Result<
+        (
+            Self,
+            BlockingEventReceiver,
+            JoinHandle<io::Result<()>>,
+            JoinHandle<io::Result<()>>,
+        ),
+        crate::ConnectError,
+    > {
+        match url.scheme() {
+            crate::transport::Scheme::Tcp => {
+                Self::connect_tcp(url.addr(), config.timeout()).map_err(Into::into)
+            }
+            #[cfg(feature = "ssl")]
+            crate::transport::Scheme::Ssl => {
+                Self::connect_ssl(url.addr(), config.validate_domain(), config.timeout())
+            }
+            #[cfg(not(feature = "ssl"))]
+            crate::transport::Scheme::Ssl => Err(crate::ConnectError::SslUnsupported),
+        }
+    }
+
+    /// Creates a new [`BlockingClient`] connected to `addr` over plaintext TCP.
+    #[allow(clippy::type_complexity)]
+    pub fn connect_tcp(
+        addr: &crate::transport::ServerAddr,
+        timeout: Option<Duration>,
+    ) -> io::Result<(
+        Self,
+        BlockingEventReceiver,
+        JoinHandle<io::Result<()>>,
+        JoinHandle<io::Result<()>>,
+    )> {
+        let writer = crate::transport::blocking::connect_tcp(addr, timeout)?;
+        let reader = writer.try_clone()?;
+        Ok(Self::new(reader, writer))
+    }
+
+    /// Creates a new [`BlockingClient`] connected to `addr` over TLS.
+    ///
+    /// `timeout` bounds TCP connect and the TLS handshake.
+    /// `validate_domain` requires a domain host.
+    #[cfg(feature = "ssl")]
+    #[allow(clippy::type_complexity)]
+    pub fn connect_ssl(
+        addr: &crate::transport::ServerAddr,
+        validate_domain: bool,
+        timeout: Option<Duration>,
+    ) -> Result<
+        (
+            Self,
+            BlockingEventReceiver,
+            JoinHandle<io::Result<()>>,
+            JoinHandle<io::Result<()>>,
+        ),
+        crate::ConnectError,
+    > {
+        let stream = crate::transport::blocking::connect_ssl(addr, validate_domain, timeout)?;
+        let (reader, writer) = stream.into_split();
+        Ok(Self::new(reader, writer))
+    }
+
     /// Sends a single tracked request to the Electrum server and waits for its response.
     ///
     /// This method blocks the current thread until the server replies. It is intended for

src/io.rs

@@ -178,7 +178,8 @@ where
 {
     let mut b = serde_json::to_vec(&msg.into()).expect("must serialize");
     b.push(b'\n');
-    writer.write_all(&b)
+    writer.write_all(&b)?;
+    writer.flush()
 }
 
 /// Asynchronously writes a JSON-RPC request or batch to an async writer, followed by a newline.
@@ -200,7 +201,8 @@ where
     use futures::AsyncWriteExt;
     let mut b = serde_json::to_vec(&msg.into()).expect("must serialize");
     b.push(b'\n');
-    writer.write_all(&b).await
+    writer.write_all(&b).await?;
+    writer.flush().await
 }
 
 /// Asynchronously writes a JSON-RPC request or batch to a tokio async writer, followed by a newline.
@@ -215,5 +217,6 @@ where
     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
+    writer.write_all(&b).await?;
+    writer.flush().await
 }

src/lib.rs

@@ -13,12 +13,20 @@ pub mod protocol;
 pub mod request;
 mod request_tracker;
 pub mod response;
+pub mod transport;
 pub use hash_types::*;
 pub use pending_request::*;
 pub use protocol::*;
 pub use request::Request;
 pub use request_tracker::*;
 pub use serde_json;
+pub use transport::{
+    ConnectConfig, ConnectConfigBuilder, ConnectError, Host, ParseServerAddrError, Scheme,
+    ServerAddr, ServerUrl,
+};
+
+#[cfg(feature = "ssl")]
+pub use transport::TlsError;
 
 /// An owned or borrowed static string.
 pub type CowStr = std::borrow::Cow<'static, str>;