Add Options::additional_headers and subprotocols

README.md

@@ -13,27 +13,29 @@ This is a simple [WebSocket](https://en.wikipedia.org/wiki/WebSocket) library fo
 ## Usage
 
 ``` rust
-let (mut sender, receiver) = ewebsock::connect("ws://example.com").unwrap();
+let options = ewebsock::Options::default();
+// add header and subprotocol to options
+let (mut sender, receiver) = ewebsock::connect("ws://example.com", options).unwrap();
 sender.send(ewebsock::WsMessage::Text("Hello!".into()));
 while let Some(event) = receiver.try_recv() {
     println!("Received {:?}", event);
 }
 ```
 
 ## Testing
+
 First start the example echo server with:
+
 ```sh
 cargo r -p echo_server
 ```
 
-Then test the native library with:
+Then test the library with:
 
 ```sh
+# native mode
 cargo run -p example_app
-```
 
-And the web library with:
-```sh
-./example_app/start_server.sh &
-./example_app/build_web.sh --open
+# web mode
+cd example_app/ && trunk serve
 ```

echo_server/Cargo.toml

@@ -10,4 +10,4 @@ publish = false
 
 
 [dependencies]
-tungstenite = { version = ">=0.17, <=0.21" }
+tungstenite = { version = "0.23" }

echo_server/src/main.rs

@@ -1,4 +1,3 @@
-#![allow(deprecated)] // TODO(emilk): Remove when we update tungstenite
 #![allow(clippy::unwrap_used, clippy::disallowed_methods)] // We are just testing here.
 
 use std::{net::TcpListener, thread::spawn};
@@ -11,14 +10,19 @@ fn main() {
         spawn(move || {
             let mut websocket = tungstenite::accept(stream.unwrap()).unwrap();
             eprintln!("New client connected");
-            while let Ok(msg) = websocket.read_message() {
+            loop {
+                let msg = websocket.read().unwrap();
+
                 // We do not want to send back ping/pong messages.
                 if msg.is_binary() || msg.is_text() {
-                    if websocket.write_message(msg).is_ok() {
-                        eprintln!("Responded.");
-                    } else {
+                    if let Err(err) = websocket.send(msg) {
+                        eprintln!("Error sending message: {err}");
                         break;
+                    } else {
+                        eprintln!("Responded.");
                     }
+                } else {
+                    eprintln!("Message received not text or binary.");
                 }
             }
             eprintln!("Client left.");

ewebsock/Cargo.toml

@@ -47,7 +47,7 @@ log = "0.4"
 
 # native:
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
-tungstenite = { version = ">=0.17, <=0.21" }
+tungstenite = { version = "0.23" }
 
 # Optional dependencies for feature "tokio":
 async-stream = { version = "0.3", optional = true }
@@ -57,7 +57,7 @@ futures-util = { version = "0.3", optional = true, default-features = false, fea
   "std",
 ] }
 tokio = { version = "1.16", features = ["rt", "sync"], optional = true }
-tokio-tungstenite = { version = ">=0.17, <=0.21", optional = true }
+tokio-tungstenite = { version = "0.23", optional = true }
 
 # web:
 [target.'cfg(target_arch = "wasm32")'.dependencies]

ewebsock/src/lib.rs

@@ -124,7 +124,7 @@ pub type Result<T> = std::result::Result<T, Error>;
 pub(crate) type EventHandler = Box<dyn Send + Fn(WsEvent) -> ControlFlow<()>>;
 
 /// Options for a connection.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Options {
     /// The maximum size of a single incoming message frame, in bytes.
     ///
@@ -133,12 +133,24 @@ pub struct Options {
     ///
     /// Ignored on Web.
     pub max_incoming_frame_size: usize,
+
+    /// Additional Request headers
+    pub additional_headers: Option<Vec<(String, String)>>,
+
+    /// Additional subprotocols
+    pub subprotocols: Option<Vec<String>>,
+
+    /// Delay blocking in ms - default 10ms
+    pub delay_blocking: Option<std::time::Duration>,
 }
 
 impl Default for Options {
     fn default() -> Self {
         Self {
             max_incoming_frame_size: 64 * 1024 * 1024,
+            additional_headers: None,
+            subprotocols: None,
+            delay_blocking: None,
         }
     }
 }

ewebsock/src/native_tungstenite.rs

@@ -1,12 +1,29 @@
-#![allow(deprecated)] // TODO(emilk): Remove when we update tungstenite
-
 use std::{
     ops::ControlFlow,
     sync::mpsc::{Receiver, TryRecvError},
 };
 
 use crate::{EventHandler, Options, Result, WsEvent, WsMessage};
 
+/// transform uri and options into a request builder
+pub fn into_requester(
+    uri: tungstenite::http::Uri,
+    options: Options,
+) -> tungstenite::client::ClientRequestBuilder {
+    let mut client_request = tungstenite::client::ClientRequestBuilder::new(uri);
+    if let Some(headers) = options.additional_headers {
+        for (key, value) in headers {
+            client_request = client_request.with_header(key, value);
+        }
+    }
+    if let Some(subprotocols) = options.subprotocols {
+        for subprotocol in subprotocols {
+            client_request = client_request.with_sub_protocol(subprotocol);
+        }
+    }
+    client_request
+}
+
 /// This is how you send [`WsMessage`]s to the server.
 ///
 /// When the last clone of this is dropped, the connection is closed.
@@ -64,21 +81,28 @@ pub(crate) fn ws_receive_impl(url: String, options: Options, on_event: EventHand
 
 /// Connect and call the given event handler on each received event.
 ///
-/// Blocking version of [`ws_receive`], only available on native.
+/// Blocking version of [`crate::ws_receive`], only available on native.
 ///
 /// # Errors
 /// All errors are returned to the caller, and NOT reported via `on_event`.
 pub fn ws_receiver_blocking(url: &str, options: Options, on_event: &EventHandler) -> Result<()> {
-    let config = tungstenite::protocol::WebSocketConfig::from(options);
+    let uri: tungstenite::http::Uri = match url.parse() {
+        Ok(uri) => uri,
+        Err(err) => return Err(format!("Failed to parse URI: {err}")),
+    };
+    let config = tungstenite::protocol::WebSocketConfig::from(options.clone());
     let max_redirects = 3; // tungstenite default
 
-    let (mut socket, response) =
-        match tungstenite::client::connect_with_config(url, Some(config), max_redirects) {
-            Ok(result) => result,
-            Err(err) => {
-                return Err(format!("Connect: {err}"));
-            }
-        };
+    let (mut socket, response) = match tungstenite::client::connect_with_config(
+        crate::into_requester(uri, options),
+        Some(config),
+        max_redirects,
+    ) {
+        Ok(result) => result,
+        Err(err) => {
+            return Err(format!("Connect: {err}"));
+        }
+    };
 
     log::debug!("WebSocket HTTP response code: {}", response.status());
     log::trace!(
@@ -95,7 +119,7 @@ pub fn ws_receiver_blocking(url: &str, options: Options, on_event: &EventHandler
     }
 
     loop {
-        let control = match socket.read_message() {
+        let control = match socket.read() {
             Ok(incoming_msg) => match incoming_msg {
                 tungstenite::protocol::Message::Text(text) => {
                     on_event(WsEvent::Message(WsMessage::Text(text)))
@@ -155,7 +179,7 @@ pub(crate) fn ws_connect_impl(
 
 /// Connect and call the given event handler on each received event.
 ///
-/// This is a blocking variant of [`ws_connect`], only available on native.
+/// This is a blocking variant of [`crate::ws_connect`], only available on native.
 ///
 /// # Errors
 /// All errors are returned to the caller, and NOT reported via `on_event`.
@@ -165,15 +189,25 @@ pub fn ws_connect_blocking(
     on_event: &EventHandler,
     rx: &Receiver<WsMessage>,
 ) -> Result<()> {
-    let config = tungstenite::protocol::WebSocketConfig::from(options);
+    let delay = options
+        .delay_blocking
+        .unwrap_or(std::time::Duration::from_millis(10)); // default value 10ms
+    let config = tungstenite::protocol::WebSocketConfig::from(options.clone());
     let max_redirects = 3; // tungstenite default
-    let (mut socket, response) =
-        match tungstenite::client::connect_with_config(url, Some(config), max_redirects) {
-            Ok(result) => result,
-            Err(err) => {
-                return Err(format!("Connect: {err}"));
-            }
-        };
+    let uri: tungstenite::http::Uri = match url.parse() {
+        Ok(uri) => uri,
+        Err(err) => return Err(format!("Failed to parse URI: {err}")),
+    };
+    let (mut socket, response) = match tungstenite::client::connect_with_config(
+        crate::into_requester(uri, options),
+        Some(config),
+        max_redirects,
+    ) {
+        Ok(result) => result,
+        Err(err) => {
+            return Err(format!("Connect: {err}"));
+        }
+    };
 
     log::debug!("WebSocket HTTP response code: {}", response.status());
     log::trace!(
@@ -216,22 +250,22 @@ pub fn ws_connect_blocking(
                     WsMessage::Pong(data) => tungstenite::protocol::Message::Pong(data),
                     WsMessage::Unknown(_) => panic!("You cannot send WsMessage::Unknown"),
                 };
-                if let Err(err) = socket.write_message(outgoing_message) {
+                if let Err(err) = socket.send(outgoing_message) {
                     socket.close(None).ok();
-                    socket.write_pending().ok();
+                    socket.flush().ok();
                     return Err(format!("send: {err}"));
                 }
             }
             Err(TryRecvError::Disconnected) => {
                 log::debug!("WsSender dropped - closing connection.");
                 socket.close(None).ok();
-                socket.write_pending().ok();
+                socket.flush().ok();
                 return Ok(());
             }
             Err(TryRecvError::Empty) => {}
         };
 
-        let control = match socket.read_message() {
+        let control = match socket.read() {
             Ok(incoming_msg) => {
                 did_work = true;
                 match incoming_msg {
@@ -273,7 +307,7 @@ pub fn ws_connect_blocking(
         }
 
         if !did_work {
-            std::thread::sleep(std::time::Duration::from_millis(10)); // TODO(emilk): make configurable
+            std::thread::sleep(delay);
         }
     }
 }

ewebsock/src/tungstenite_common.rs

@@ -2,6 +2,7 @@ impl From<crate::Options> for tungstenite::protocol::WebSocketConfig {
     fn from(options: crate::Options) -> Self {
         let crate::Options {
             max_incoming_frame_size,
+            ..
         } = options;
 
         Self {

example_app/.gitignore

@@ -0,0 +1 @@
+dist/

example_app/Cargo.toml

@@ -9,10 +9,6 @@ include = ["../LICENSE-APACHE", "../LICENSE-MIT", "**/*.rs", "Cargo.toml"]
 publish = false
 
 
-[lib]
-crate-type = ["cdylib", "rlib"]
-
-
 [features]
 default = []
 
@@ -25,6 +21,7 @@ ewebsock = { path = "../ewebsock", features = ["tls"] }
 
 eframe = "0.26.2" # Gives us egui, epi and web+native backends
 log = "0.4"
+env_logger = "0.10"
 
 # native:
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]