fix: use rustls and switch ConnectionHandler to async_trait

Cargo.toml

@@ -26,6 +26,7 @@ relay_rpc = { path = "./relay_rpc", optional = true }
 
 [dev-dependencies]
 anyhow = "1"
+async-trait = "0.1"
 structopt = { version = "0.3", default-features = false }
 tokio = { version = "1.22", features = ["full"] }
 

examples/basic_client.rs

@@ -1,4 +1,5 @@
 use {
+    async_trait::async_trait,
     relay_client::{
         Client,
         CloseFrame,
@@ -36,27 +37,28 @@ impl Handler {
     }
 }
 
+#[async_trait]
 impl ConnectionHandler for Handler {
-    fn connected(&mut self) {
+    async fn connected(&mut self) {
         println!("[{}] connection open", self.name);
     }
 
-    fn disconnected(&mut self, frame: Option<CloseFrame<'static>>) {
+    async fn disconnected(&mut self, frame: Option<CloseFrame<'static>>) {
         println!("[{}] connection closed: frame={frame:?}", self.name);
     }
 
-    fn message_received(&mut self, message: PublishedMessage) {
+    async fn message_received(&mut self, message: PublishedMessage) {
         println!(
             "[{}] inbound message: topic={} message={}",
             self.name, message.topic, message.message
         );
     }
 
-    fn inbound_error(&mut self, error: Error) {
+    async fn inbound_error(&mut self, error: Error) {
         println!("[{}] inbound error: {error}", self.name);
     }
 
-    fn outbound_error(&mut self, error: Error) {
+    async fn outbound_error(&mut self, error: Error) {
         println!("[{}] outbound error: {error}", self.name);
     }
 }

relay_client/Cargo.toml

@@ -4,11 +4,12 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+async-trait = "0.1"
 relay_rpc = { path = "../relay_rpc" }
 futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
 thiserror = "1.0"
 tokio = { version = "1.22", features = ["rt", "time", "sync", "macros", "rt-multi-thread"] }
-tokio-tungstenite = { version = "0.17", features = ["connect", "native-tls"] }
+tokio-tungstenite = { version = "0.18", features = ["rustls-tls-native-roots"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_qs = "0.10"

relay_client/src/client.rs

@@ -1,6 +1,7 @@
 use {
     self::connection::{connection_event_loop, ConnectionControl},
     crate::{ConnectionOptions, Error},
+    async_trait::async_trait,
     relay_rpc::{
         domain::{SubscriptionId, Topic},
         rpc::{BatchSubscribe, BatchUnsubscribe, Publish, Subscribe, Subscription, Unsubscribe},
@@ -44,23 +45,24 @@ impl PublishedMessage {
 }
 
 /// Handlers for the RPC stream events.
+#[async_trait]
 pub trait ConnectionHandler: Send + 'static {
     /// Called when a connection to the Relay is established.
-    fn connected(&mut self) {}
+    async fn connected(&mut self) {}
 
     /// Called when the Relay connection is closed.
-    fn disconnected(&mut self, _frame: Option<CloseFrame<'static>>) {}
+    async fn disconnected(&mut self, _frame: Option<CloseFrame<'static>>) {}
 
     /// Called when a message is received from the Relay.
-    fn message_received(&mut self, message: PublishedMessage);
+    async fn message_received(&mut self, message: PublishedMessage);
 
     /// Called when an inbound error occurs, such as data deserialization
     /// failure, or an unknown response message ID.
-    fn inbound_error(&mut self, _error: Error) {}
+    async fn inbound_error(&mut self, _error: Error) {}
 
     /// Called when an outbound error occurs, i.e. failed to write to the
     /// websocket stream.
-    fn outbound_error(&mut self, _error: Error) {}
+    async fn outbound_error(&mut self, _error: Error) {}
 }
 
 /// The Relay RPC client.

relay_client/src/client/connection.rs

@@ -49,7 +49,7 @@ pub(super) async fn connection_event_loop<T>(
                             let result = conn.connect(*opts).await;
 
                             if result.is_ok() {
-                                handler.connected();
+                                handler.connected().await;
                             }
 
                             tx.send(result).ok();
@@ -67,7 +67,7 @@ pub(super) async fn connection_event_loop<T>(
                     // Control TX has been dropped, shutting down.
                     None => {
                         conn.disconnect().await.ok();
-                        handler.disconnected(None);
+                        handler.disconnected(None).await;
                         break;
                     }
                 }
@@ -76,20 +76,20 @@ pub(super) async fn connection_event_loop<T>(
             event = conn.select_next_some() => {
                 match event {
                     StreamEvent::InboundSubscriptionRequest(request) => {
-                        handler.message_received(PublishedMessage::from_request(&request));
+                        handler.message_received(PublishedMessage::from_request(&request)).await;
                         request.respond(Ok(true)).ok();
                     }
 
                     StreamEvent::InboundError(error) => {
-                        handler.inbound_error(error);
+                        handler.inbound_error(error).await;
                     }
 
                     StreamEvent::OutboundError(error) => {
-                        handler.outbound_error(error);
+                        handler.outbound_error(error).await;
                     }
 
                     StreamEvent::ConnectionClosed(frame) => {
-                        handler.disconnected(frame);
+                        handler.disconnected(frame).await;
                         conn.reset();
                     }
                 }