Repository: 1c3t3a/rust-socketio (language: rust)

Issue:
Add multiple transport layers
The documentation talks about different 'layers' for transporting which are:
- websockets
- long-polling

I currently implemented long-polling, but feel free to extend the transport enum in engineio/transport.rs.


Discussion / hints:

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index c978609c..5a7b0235 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -22,3 +22,4 @@ serde_json = "1.0"
 regex = "1.4.2"
 if_chain = "1.0.1"
 byte = "0.2.4"
+websocket = "0.26.2"
diff --git a/README.md b/README.md
index 824fcd4e..de92399c 100644
--- a/README.md
+++ b/README.md
@@ -1,9 +1,10 @@
 [![Latest Version](https://img.shields.io/crates/v/rust_socketio)](https://crates.io/crates/rust_socketio)
+
 ![tests](https://github.com/1c3t3a/rust-socketio/workflows/Rust/badge.svg)
 
 # Rust-socketio-client
 
-An implementation of a socket.io client written in the Rust programming language. This implementation currently supports revision 5 of the socket.io protocol and therefore revision 4 of the engine.io protocol.
+An implementation of a socket.io client written in the Rust programming language. This implementation currently supports revision 5 of the socket.io protocol and therefore revision 4 of the engine.io protocol. If you have any connection issues with this client, make sure the server uses at least revision 4 of the engine.io protocol.
 
 ## Example usage
 
@@ -12,13 +13,14 @@ use rust_socketio::Socket;
 use serde_json::json;
 
 fn main() {
-    let mut socket = Socket::new("http://localhost:80", Some("/admin"));
-
-    // callback for the "foo" event
-    socket.on("foo", |message| println!("{}", message)).unwrap();
-
-    // connect to the server
-    socket.connect().expect("Connection failed");
+    // connect to a server on localhost with the /admin namespace and
+    // a foo event handler
+    let mut socket = SocketBuilder::new("http://localhost:80")
+          .set_namespace("/admin")
+          .expect("illegal namespace")
+          .on("test", |str| println!("Received: {}", str))
+          .connect()
+          .expect("Connection failed");
 
     // emit to the "foo" event
     let payload = json!({"token": 123});
@@ -37,13 +39,15 @@ fn main() {
  }
 ```
 
+The main entry point for using this crate is the `SocketBuilder` which provides a way to easily configure a socket in the needed way. When the `connect` method is called on the builder, it returns a connected client which then could be used to emit messages to certain events. One client can only be connected to one namespace. If you need to listen to the messages in different namespaces you need to allocate multiple sockets.
+
 ## Documentation
 
 Documentation of this crate can be found up on [docs.rs](https://docs.rs/rust_socketio/0.1.0/rust_socketio/).
 
 ## Current features
 
-This is the first released version of the client, so it still lacks some features that the normal client would provide. First of all the underlying engine.io protocol still uses long-polling instead of websockets. This will be resolved as soon as both the reqwest libary as well as tungstenite-websockets will bump their tokio version to 1.0.0. At the moment only reqwest is used for long polling. In general the full engine-io protocol is implemented and most of the features concerning the 'normal' socket.io protocol work as well.
+This implementation support most of the features of the socket.io protocol. In general the full engine-io protocol is implemented, and concerning the socket.io part only binary events and binary acks are not yet implemented. This implementation generally tries to make use of websockets as often as possible. This means most times only the opening request uses http and as soon as the server mentions that he is able to use websockets, an upgrade is performed. But if this upgrade is not successful or the server does not mention an upgrade possibilty, http-long polling is used (as specified in the protocol specs).
 
 Here's an overview of possible use-cases:
 
@@ -57,10 +61,6 @@ Here's an overview of possible use-cases:
 * send json-data to the server (recommended to use serde_json as it provides safe handling of json data).
 * send json-data to the server and receive an ack with a possible message.
 
-What's currently missing is the emitting of binary data - I aim to implement this as soon as possible.
-
-The whole crate is written in rust and it's necessary to use [tokio](https://docs.rs/tokio/1.0.1/tokio/), or other executors with this libary to resolve the futures.
-
 ## Content of this repository
 
 This repository contains a rust implementation of the socket.io protocol as well as the underlying engine.io protocol.
diff --git a/src/engineio/socket.rs b/src/engineio/socket.rs
index ffaefa3b..a85de98d 100644
--- a/src/engineio/socket.rs
+++ b/src/engineio/socket.rs
@@ -33,7 +33,7 @@ impl EngineSocket {
 
         let cl = Arc::clone(&self.transport_client);
         thread::spawn(move || {
-            let s = cl.read().unwrap().clone();
+            let mut s = cl.read().unwrap().clone();
             // tries to restart a poll cycle whenever a 'normal' error occurs,
             // it just panics on network errors, in case the poll cycle returned
             // `Result::Ok`, the server receives a close frame so it's safe to
diff --git a/src/engineio/transport.rs b/src/engineio/transport.rs
index 6ada5c5c..e750735e 100644
--- a/src/engineio/transport.rs
+++ b/src/engineio/transport.rs
@@ -9,11 +9,16 @@ use std::{
     sync::{atomic::AtomicBool, Arc, Mutex, RwLock},
     time::{Duration, Instant},
 };
+use websocket::{
+    dataframe::DataFrame as WsDataFrame, receiver::Reader, sync::Writer, ws::dataframe::DataFrame,
+};
+use websocket::{sync::stream::TcpStream, ClientBuilder};
 
 /// The different types of transport used for transmitting a payload.
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 enum TransportType {
     Polling(Arc<Mutex<Client>>),
+    Websocket(Arc<Mutex<Reader<TcpStream>>>, Arc<Mutex<Writer<TcpStream>>>),
 }
 
 /// Type of a `Callback` function. (Normal closures can be passed in here).
@@ -24,7 +29,7 @@ type Callback<I> = Arc<RwLock<Option<Box<dyn Fn(I) + 'static + Sync + Send>>>>;
 /// callback functions.
 #[derive(Clone)]
 pub struct TransportClient {
-    transport: TransportType,
+    transport: Arc<TransportType>,
     pub on_error: Callback<String>,
     pub on_open: Callback<()>,
     pub on_close: Callback<()>,
@@ -34,7 +39,7 @@ pub struct TransportClient {
     last_ping: Arc<Mutex<Instant>>,
     last_pong: Arc<Mutex<Instant>>,
     host_address: Arc<Mutex<Option<String>>>,
-    connection_data: Arc<Option<HandshakeData>>,
+    connection_data: Arc<RwLock<Option<HandshakeData>>>,
     engine_io_mode: Arc<AtomicBool>,
 }
 
@@ -53,7 +58,7 @@ impl TransportClient {
     /// Creates an instance of `TransportClient`.
     pub fn new(engine_io_mode: bool) -> Self {
         TransportClient {
-            transport: TransportType::Polling(Arc::new(Mutex::new(Client::new()))),
+            transport: Arc::new(TransportType::Polling(Arc::new(Mutex::new(Client::new())))),
             on_error: Arc::new(RwLock::new(None)),
             on_open: Arc::new(RwLock::new(None)),
             on_close: Arc::new(RwLock::new(None)),
@@ -63,7 +68,7 @@ impl TransportClient {
             last_ping: Arc::new(Mutex::new(Instant::now())),
             last_pong: Arc::new(Mutex::new(Instant::now())),
             host_address: Arc::new(Mutex::new(None)),
-            connection_data: Arc::new(None),
+            connection_data: Arc::new(RwLock::new(None)),
             engine_io_mode: Arc::new(AtomicBool::from(engine_io_mode)),
         }
     }
@@ -125,8 +130,9 @@ impl TransportClient {
 
     /// Opens the connection to a specified server. Includes an opening `GET`
     /// request to the server, the server passes back the handshake data in the
-    /// response. Afterwards a first Pong packet is sent to the server to
-    /// trigger the Ping-cycle.
+    /// response. If the handshake data mentions a websocket upgrade possibility,
+    /// we try to upgrade the connection. Afterwards a first Pong packet is sent
+    /// to the server to trigger the Ping-cycle.
     pub fn open<T: Into<String> + Clone>(&mut self, address: T) -> Result<()> {
         // TODO: Check if Relaxed is appropiate -> change all occurences if not
         if self.connected.load(Ordering::Relaxed) {
@@ -134,9 +140,9 @@ impl TransportClient {
         }
 
         // build the query path, random_t is used to prevent browser caching
-        let query_path = self.get_query_path();
+        let query_path = self.get_query_path()?;
 
-        match &mut self.transport {
+        match &mut self.transport.as_ref() {
             TransportType::Polling(client) => {
                 if let Ok(full_address) = Url::parse(&(address.to_owned().into() + &query_path)) {
                     self.host_address = Arc::new(Mutex::new(Some(address.into())));
@@ -144,8 +150,27 @@ impl TransportClient {
                     let response = client.lock()?.get(full_address).send()?.text()?;
 
                     // the response contains the handshake data
-                    if let Ok(conn_data) = serde_json::from_str(&response[1..]) {
-                        self.connection_data = Arc::new(conn_data);
+                    if let Ok(conn_data) = serde_json::from_str::<HandshakeData>(&response[1..]) {
+                        *Arc::get_mut(&mut self.connected).unwrap() = AtomicBool::from(true);
+
+                        // check if we could upgrade to websockets
+                        let websocket_upgrade = conn_data
+                            .upgrades
+                            .iter()
+                            .any(|upgrade| upgrade.to_lowercase() == *"websocket");
+
+                        // if we have an upgrade option, send the corresponding request, if this doesnt work
+                        // for some reason, proceed via polling
+                        if websocket_upgrade {
+                            if let Err(error) = self.upgrade_connection(&conn_data.sid) {
+                                eprintln!("upgrading to websockets was not succesfull because of [{}], proceeding via polling", error);
+                            }
+                        }
+
+                        // set the connection data
+                        let mut connection_data = self.connection_data.write()?;
+                        *connection_data = Some(conn_data);
+                        drop(connection_data);
 
                         // call the on_open callback
                         let function = self.on_open.read()?;
@@ -156,7 +181,6 @@ impl TransportClient {
 
                         // set the last ping to now and set the connected state
                         *self.last_ping.lock()? = Instant::now();
-                        *Arc::get_mut(&mut self.connected).unwrap() = AtomicBool::from(true);
 
                         // emit a pong packet to keep trigger the ping cycle on the server
                         self.emit(Packet::new(PacketId::Pong, Vec::new()))?;
@@ -173,9 +197,83 @@ impl TransportClient {
                 self.call_error_callback(format!("{}", error))?;
                 Err(error)
             }
+            _ => unreachable!("Can't happen as we initialize with polling and upgrade later"),
         }
     }
 
+    /// This handles the upgrade from polling to websocket transport. Looking at the protocol
+    /// specs, this needs the following preconditions:
+    /// - the handshake must have already happened
+    /// - the handshake `upgrade` field needs to contain the value `websocket`.
+    /// If those precondionts are valid, it is save to call the method. The protocol then
+    /// requires the following procedure:
+    /// - the client starts to connect to the server via websocket, mentioning the related `sid` received
+    ///   in the handshake.
+    /// - to test out the connection, the client sends a `ping` packet with the payload `probe`.
+    /// - if the server receives this correctly, it responses by sending a `pong` packet with the payload `probe`.
+    /// - if the client receives that both sides can be sure that the upgrade is save and the client requests
+    ///   the final updgrade by sending another `update` packet over `websocket`.
+    /// If this procedure is alright, the new transport is established.
+    fn upgrade_connection(&mut self, new_sid: &str) -> Result<()> {
+        let host_address = self.host_address.lock()?;
+        if let Some(address) = &*host_address {
+            // unwrapping here is infact save as we only set this url if it gets orderly parsed
+            let mut address = Url::parse(&address).unwrap();
+
+            if self.engine_io_mode.load(Ordering::Relaxed) {
+                address.set_path(&(address.path().to_owned() + "engine.io/"));
+            } else {
+                address.set_path(&(address.path().to_owned() + "socket.io/"));
+            }
+
+            let full_address = address
+                .query_pairs_mut()
+                .clear()
+                .append_pair("EIO", "4")
+                .append_pair("transport", "websocket")
+                .append_pair("sid", new_sid)
+                .finish();
+
+            // connect to the server via websockets
+            let client = ClientBuilder::new(full_address.as_ref())?.connect_insecure()?;
+            client.set_nonblocking(false)?;
+
+            let (mut receiver, mut sender) = client.split()?;
+
+            // send the probe packet
+            let probe_packet = Packet::new(PacketId::Ping, b"probe".to_vec());
+            sender.send_dataframe(&WsDataFrame::new(
+                true,
+                websocket::dataframe::Opcode::Text,
+                encode_payload(vec![probe_packet]),
+            ))?;
+
+            // expect to receive a probe packet
+            let message = receiver.recv_dataframe()?;
+            if message.take_payload() != b"3probe" {
+                eprintln!("Error while handshaking ws");
+                return Err(Error::HandshakeError("Error".to_owned()));
+            }
+
+            let upgrade_packet = Packet::new(PacketId::Upgrade, Vec::new());
+            // finally send the upgrade request
+            sender.send_dataframe(&WsDataFrame::new(
+                true,
+                websocket::dataframe::Opcode::Text,
+                encode_payload(vec![upgrade_packet]),
+            ))?;
+
+            // upgrade the transport layer
+            self.transport = Arc::new(TransportType::Websocket(
+                Arc::new(Mutex::new(receiver)),
+                Arc::new(Mutex::new(sender)),
+            ));
+
+            return Ok(());
+        }
+        Err(Error::HandshakeError("Error - invalid url".to_owned()))
+    }
+
     /// Sends a packet to the server
     pub fn emit(&self, packet: Packet) -> Result<()> {
         if !self.connected.load(Ordering::Relaxed) {
@@ -183,19 +281,18 @@ impl TransportClient {
             self.call_error_callback(format!("{}", error))?;
             return Err(error);
         }
+        // build the target address
+        let query_path = self.get_query_path()?;
 
-        match &self.transport {
-            TransportType::Polling(client) => {
-                let query_path = self.get_query_path();
+        let host = self.host_address.lock()?;
+        let address = Url::parse(&(host.as_ref().unwrap().to_owned() + &query_path)[..]).unwrap();
+        drop(host);
 
-                // build the target address
-                let host = self.host_address.lock()?;
-                let address =
-                    Url::parse(&(host.as_ref().unwrap().to_owned() + &query_path)[..]).unwrap();
-                drop(host);
+        // send a post request with the encoded payload as body
+        let data = encode_payload(vec![packet]);
 
-                // send a post request with the encoded payload as body
-                let data = encode_payload(vec![packet]);
+        match &self.transport.as_ref() {
+            TransportType::Polling(client) => {
                 let client = client.lock()?;
                 let status = client.post(address).body(data).send()?.status().as_u16();
                 drop(client);
@@ -206,6 +303,14 @@ impl TransportClient {
                     return Err(error);
                 }
 
+                Ok(())
+            }
+            TransportType::Websocket(_, writer) => {
+                let mut writer = writer.lock()?;
+
+                let dataframe = WsDataFrame::new(true, websocket::dataframe::Opcode::Text, data);
+                writer.send_dataframe(&dataframe).unwrap();
+
                 Ok(())
             }
         }
@@ -214,7 +319,7 @@ impl TransportClient {
     /// Performs the server long polling procedure as long as the client is
     /// connected. This should run separately at all time to ensure proper
     /// response handling from the server.
-    pub fn poll_cycle(&self) -> Result<()> {
+    pub fn poll_cycle(&mut self) -> Result<()> {
         if !self.connected.load(Ordering::Relaxed) {
             let error = Error::ActionBeforeOpen;
             self.call_error_callback(format!("{}", error))?;
@@ -227,21 +332,23 @@ impl TransportClient {
         // the time after we assume the server to be timed out
         let server_timeout = Duration::from_millis(
             Arc::as_ref(&self.connection_data)
+                .read()?
                 .as_ref()
                 .unwrap()
                 .ping_timeout
                 + Arc::as_ref(&self.connection_data)
+                    .read()?
                     .as_ref()
                     .unwrap()
                     .ping_interval,
         );
 
         while self.connected.load(Ordering::Relaxed) {
-            match &self.transport {
+            let data = match &self.transport.as_ref() {
                 // we wont't use the shared client as this blocks the ressource
                 // in the long polling requests
                 TransportType::Polling(_) => {
-                    let query_path = self.get_query_path();
+                    let query_path = self.get_query_path()?;
 
                     let host = self.host_address.lock()?;
                     let address =
@@ -249,67 +356,74 @@ impl TransportClient {
                     drop(host);
 
                     // TODO: check if to_vec is inefficient here
-                    let response = client.get(address).send()?.bytes()?.to_vec();
-                    let packets = decode_payload(response)?;
-
-                    for packet in packets {
-                        {
-                            let on_packet = self.on_packet.read()?;
-                            // call the packet callback
-                            if let Some(function) = on_packet.as_ref() {
-                                spawn_scoped!(function(packet.clone()));
-                            }
-                        }
-                        // check for the appropiate action or callback
-                        match packet.packet_id {
-                            PacketId::Message => {
-                                let on_data = self.on_data.read()?;
-                                if let Some(function) = on_data.as_ref() {
-                                    spawn_scoped!(function(packet.data));
-                                }
-                                drop(on_data);
-                            }
+                    client.get(address).send()?.bytes()?.to_vec()
+                }
+                TransportType::Websocket(receiver, _) => {
+                    let mut receiver = receiver.lock()?;
 
-                            PacketId::Close => {
-                                let on_close = self.on_close.read()?;
-                                if let Some(function) = on_close.as_ref() {
-                                    spawn_scoped!(function(()));
-                                }
-                                drop(on_close);
-                                // set current state to not connected and stop polling
-                                self.connected
-                                    .compare_and_swap(true, false, Ordering::Acquire);
-                            }
-                            PacketId::Open => {
-                                // this will never happen as the client connects
-                                // to the server and the open packet is already
-                                // received in the 'open' method
-                                unreachable!();
-                            }
-                            PacketId::Upgrade => {
-                                todo!("Upgrade the connection, but only if possible");
-                            }
-                            PacketId::Ping => {
-                                last_ping = Instant::now();
-                                self.emit(Packet::new(PacketId::Pong, Vec::new()))?;
-                            }
-                            PacketId::Pong => {
-                                // this will never happen as the pong packet is
-                                // only sent by the client
-                                unreachable!();
-                            }
-                            PacketId::Noop => (),
+                    receiver.recv_message().unwrap().take_payload()
+                }
+            };
+
+            if data.is_empty() {
+                return Ok(());
+            }
+
+            let packets = decode_payload(data)?;
+
+            for packet in packets {
+                {
+                    let on_packet = self.on_packet.read()?;
+                    // call the packet callback
+                    if let Some(function) = on_packet.as_ref() {
+                        spawn_scoped!(function(packet.clone()));
+                    }
+                }
+                // check for the appropiate action or callback
+                match packet.packet_id {
+                    PacketId::Message => {
+                        let on_data = self.on_data.read()?;
+                        if let Some(function) = on_data.as_ref() {
+                            spawn_scoped!(function(packet.data));
                         }
+                        drop(on_data);
                     }
 
-                    if server_timeout < last_ping.elapsed() {
-                        // the server is unreachable
+                    PacketId::Close => {
+                        let on_close = self.on_close.read()?;
+                        if let Some(function) = on_close.as_ref() {
+                            spawn_scoped!(function(()));
+                        }
+                        drop(on_close);
                         // set current state to not connected and stop polling
                         self.connected
                             .compare_and_swap(true, false, Ordering::Acquire);
                     }
+                    PacketId::Open => {
+                        unreachable!("Won't happen as we open the connection beforehand");
+                    }
+                    PacketId::Upgrade => {
+                        todo!("Upgrade the connection, but only if possible");
+                    }
+                    PacketId::Ping => {
+                        last_ping = Instant::now();
+                        self.emit(Packet::new(PacketId::Pong, Vec::new()))?;
+                    }
+                    PacketId::Pong => {
+                        // this will never happen as the pong packet is
+                        // only sent by the client
+                        unreachable!();
+                    }
+                    PacketId::Noop => (),
                 }
             }
+
+            if server_timeout < last_ping.elapsed() {
+                // the server is unreachable
+                // set current state to not connected and stop polling
+                self.connected
+                    .compare_and_swap(true, false, Ordering::Acquire);
+            }
         }
         Ok(())
     }
@@ -336,7 +450,7 @@ impl TransportClient {
 
     // Constructs the path for a request, depending on the different situations.
     #[inline]
-    fn get_query_path(&self) -> String {
+    fn get_query_path(&self) -> Result<String> {
         // build the base path
         let mut path = format!(
             "/{}/?EIO=4&transport={}&t={}",
@@ -345,20 +459,38 @@ impl TransportClient {
             } else {
                 "socket.io"
             },
-            match self.transport {
+            match self.transport.as_ref() {
                 TransportType::Polling(_) => "polling",
+                TransportType::Websocket(_, _) => "websocket",
             },
             TransportClient::get_random_t(),
         );
+
         // append a session id if the socket is connected
         if self.connected.load(Ordering::Relaxed) {
             path.push_str(&format!(
                 "&sid={}",
-                Arc::as_ref(&self.connection_data).as_ref().unwrap().sid
+                Arc::as_ref(&self.connection_data)
+                    .read()?
+                    .as_ref()
+                    .unwrap()
+                    .sid
             ));
         }
 
-        path
+        Ok(path)
+    }
+}
+
+impl Debug for TransportType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_fmt(format_args!(
+            "TransportType({})",
+            match self {
+                Self::Polling(_) => "Polling",
+                Self::Websocket(_, _) => "Websocket",
+            }
+        ))
     }
 }
 
diff --git a/src/error.rs b/src/error.rs
index 4e1ddcc8..cd2e9fe1 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -4,6 +4,7 @@ use std::{
     fmt::{self, Display, Formatter},
     num::ParseIntError,
 };
+use websocket::{client::ParseError, WebSocketError};
 
 /// Enumeration of all possible errors in the `socket.io` context.
 #[derive(Debug)]
@@ -22,7 +23,11 @@ pub enum Error {
     InvalidJson(String),
     DidNotReceiveProperAck(i32),
     IllegalActionAfterOpen,
+    IllegalNamespace(String),
     PoisonedLockError,
+    FromWebsocketError(WebSocketError),
+    FromWebsocketParseError(ParseError),
+    FromIoError(std::io::Error),
 }
 
 impl std::error::Error for Error {}
@@ -60,6 +65,24 @@ impl From<ParseIntError> for Error {
     }
 }
 
+impl From<WebSocketError> for Error {
+    fn from(error: WebSocketError) -> Self {
+        Self::FromWebsocketError(error)
+    }
+}
+
+impl From<ParseError> for Error {
+    fn from(error: ParseError) -> Self {
+        Self::FromWebsocketParseError(error)
+    }
+}
+
+impl From<std::io::Error> for Error {
+    fn from(error: std::io::Error) -> Self {
+        Self::FromIoError(error)
+    }
+}
+
 impl Display for Error {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         match &self {
@@ -83,6 +106,7 @@ impl Display for Error {
             Self::ActionBeforeOpen => {
                 write!(f, "Called an action before the connection was established")
             }
+            Self::IllegalNamespace(nsp) => write!(f, "Specified namespace {} is not valid", nsp),
             Self::HttpError(status_code) => write!(
                 f,
                 "Network request returned with status code: {}",
@@ -93,6 +117,9 @@ impl Display for Error {
             Self::IllegalActionAfterOpen => write!(f, "An illegal action (such as setting a callback after being connected) was triggered"),
             Self::PoisonedLockError => write!(f, "A lock was poisoned"),
             Self::InvalidPacket => write!(f, "Got an invalid packet which did not follow the protocol format"),
+            Self::FromWebsocketError(error) => write!(f, "Got a websocket error: {}", error),
+            Self::FromWebsocketParseError(error) => write!(f, "Error while parsing the url for the websocket connection: {}", error),
+            Self::FromIoError(error) => write!(f, "Got an IoError: {}", error),
         }
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index cd332eb5..14c0929f 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -2,18 +2,18 @@
 //! ## Example usage
 //!
 //! ``` rust
-//! use rust_socketio::Socket;
+//! use rust_socketio::SocketBuilder;
 //! use serde_json::json;
-//! use std::thread::sleep;
 //! use std::time::Duration;
 //!
-//! let mut socket = Socket::new(String::from("http://localhost:4200"), Some("/admin"));
-//!
-//! // callback for the "foo" event
-//! socket.on("foo", |message| println!("{}", message)).unwrap();
-//!
-//! // connect to the server
-//! socket.connect().expect("Connection failed");
+//! // get a socket that is connected to the admin namespace
+//! let mut socket = SocketBuilder::new("http://localhost:4200")
+//!      .set_namespace("/admin")
+//!      .expect("illegal namespace")
+//!      .on("test", |str| println!("Received: {}", str))
+//!      .on("error", |err| eprintln!("Error: {}", err))
+//!      .connect()
+//!      .expect("Connection failed");
 //!
 //! // emit to the "foo" event
 //! let payload = json!({"token": 123});
@@ -25,23 +25,28 @@
 //!     println!("Ack data: {}", message);
 //! };
 //!
-//! sleep(Duration::from_secs(2));
-//!
 //! // emit with an ack
 //! let ack = socket
 //!     .emit_with_ack("test", &payload.to_string(), Duration::from_secs(2), ack_callback)
 //!     .expect("Server unreachable");
 //! ```
 //!
+//! The main entry point for using this crate is the `SocketBuilder` which provides
+//! a way to easily configure a socket in the needed way. When the `connect` method
+//! is called on the builder, it returns a connected client which then could be used
+//! to emit messages to certain events. One client can only be connected to one namespace.
+//! If you need to listen to the messages in different namespaces you need to
+//! allocate multiple sockets.
+//!
 //! ## Current features
 //!
-//! This version of the client lacks some features that the reference client
-//! would provide. The underlying `engine.io` protocol still uses long-polling
-//! instead of websockets. This will be resolved as soon as both the reqwest
-//! libary as well as `tungsenite-websockets` will bump their `tokio` version to
-//! 1.0.0. At the moment only `reqwest` is used for long-polling. The full
-//! `engine-io` protocol is implemented and most of the features concerning the
-//! 'normal' `socket.io` protocol are working.
+//! This implementation support most of the features of the socket.io protocol. In general
+//! the full engine-io protocol is implemented, and concerning the socket.io part only binary
+//! events and binary acks are not yet implemented. This implementation generally tries to
+//! make use of websockets as often as possible. This means most times only the opening request
+//! uses http and as soon as the server mentions that he is able to use websockets, an upgrade
+//! is performed. But if this upgrade is not successful or the server does not mention an upgrade
+//! possibilty, http-long polling is used (as specified in the protocol specs).
 //!
 //! Here's an overview of possible use-cases:
 //!
@@ -56,10 +61,6 @@
 //! handling).
 //! - send JSON data to the server and receive an `ack`.
 //!
-//! The whole crate is written in asynchronous Rust and it's necessary to use
-//! [tokio](https://docs.rs/tokio/1.0.1/tokio/), or other executors with this
-//! library to resolve the futures.
-//!
 #![allow(clippy::rc_buffer)]
 #![warn(clippy::complexity)]
 #![warn(clippy::style)]
@@ -85,8 +86,10 @@ pub mod socketio;
 /// crate. Handles all kinds of errors.
 pub mod error;
 
+use error::Error;
+
 use crate::error::Result;
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 use crate::socketio::transport::TransportClient;
 
@@ -99,64 +102,133 @@ pub struct Socket {
     transport: TransportClient,
 }
 
-impl Socket {
-    /// Creates a socket with a certain adress to connect to as well as a
-    /// namespace. If `None` is passed in as namespace, the default namespace
-    /// `"/"` is taken.
-    ///
+/// A builder class for a `socket.io` socket. This handles setting up the client and
+/// configuring the callback, the namespace and metadata of the socket. If no
+/// namespace is specified, the default namespace `/` is taken. The `connect` method
+/// acts the `build` method and returns a connected [`Socket`].
+pub struct SocketBuilder {
+    socket: Socket,
+}
+
+impl SocketBuilder {
+    /// Create as client builder from a URL. URLs must be in the form
+    /// `[ws or wss or http or https]://[domain]:[port]/[path]`. The
+    /// path of the URL is optional and if no port is given, port 80
+    /// will be used.
     /// # Example
     /// ```rust
-    /// use rust_socketio::Socket;
+    /// use rust_socketio::SocketBuilder;
+    /// use serde_json::json;
     ///
-    /// // this connects the socket to the given url as well as the default
-    /// // namespace "/"
-    /// let socket = Socket::new("http://localhost:80", None);
+    /// let mut socket = SocketBuilder::new("http://localhost:4200")
+    ///     .set_namespace("/admin")
+    ///     .expect("illegal namespace")
+    ///     .on("test", |str| println!("Received: {}", str))
+    ///     .connect()
+    ///     .expect("error while connecting");
     ///
-    /// // this connects the socket to the given url as well as the namespace
-    /// // "/admin"
-    /// let socket = Socket::new("http://localhost:80", Some("/admin"));
+    /// // use the socket
+    /// let payload = json!({"token": 123});
+    /// let result = socket.emit("foo", &payload.to_string());
+    ///
+    /// assert!(result.is_ok());
     /// ```
-    pub fn new<T: Into<String>>(address: T, namespace: Option<&str>) -> Self {
-        Socket {
-            transport: TransportClient::new(address, namespace.map(String::from)),
+    pub fn new<T: Into<String>>(address: T) -> Self {
+        Self {
+            socket: Socket::new(address, Some("/")),
         }
     }
 
-    /// Registers a new callback for a certain event. This returns an
-    /// `Error::IllegalActionAfterOpen` error if the callback is registered
-    /// after a call to the `connect` method.
+    /// Sets the target namespace of the client. The namespace must start
+    /// with a leading `/`. Valid examples are e.g. `/admin`, `/foo`.
+    pub fn set_namespace<T: Into<String>>(mut self, namespace: T) -> Result<Self> {
+        let nsp = namespace.into();
+        if !nsp.starts_with('/') {
+            return Err(Error::IllegalNamespace(nsp));
+        }
+        self.socket.set_namespace(nsp);
+        Ok(self)
+    }
+
+    /// Registers a new callback for a certain [`socketio::event::Event`]. The event could either be
+    /// one of the common events like `message`, `error`, `connect`, `close` or a custom
+    /// event defined by a string, e.g. `onPayment` or `foo`.
     /// # Example
     /// ```rust
-    /// use rust_socketio::Socket;
+    /// use rust_socketio::SocketBuilder;
+    ///
+    /// let socket = SocketBuilder::new("http://localhost:4200")
+    ///     .set_namespace("/admin")
+    ///     .expect("illegal namespace")
+    ///     .on("test", |str| println!("Received: {}", str))
+    ///     .on("error", |err| eprintln!("Error: {}", err))
+    ///     .connect();
     ///
-    /// let mut socket = Socket::new("http://localhost:4200", None);
-    /// let result = socket.on("foo", |message| println!("{}", message));
     ///
-    /// assert!(result.is_ok());
     /// ```
-    pub fn on<F>(&mut self, event: &str, callback: F) -> Result<()>
+    pub fn on<F>(mut self, event: &str, callback: F) -> Self
     where
         F: FnMut(String) + 'static + Sync + Send,
     {
-        self.transport.on(event.into(), callback)
+        // unwrapping here is safe as this only returns an error
+        // when the client is already connected, which is
+        // impossible here
+        self.socket.on(event, callback).unwrap();
+        self
     }
 
-    /// Connects the client to a server. Afterwards the `emit_*` methods can be
-    /// called to interact with the server. Attention: it's not allowed to add a
-    /// callback after a call to this method.
-    ///
+    /// Connects the socket to a certain endpoint. This returns a connected
+    /// [`Socket`] instance. This method returns an [`std::result::Result::Err`]
+    /// value if something goes wrong during connection.
     /// # Example
     /// ```rust
-    /// use rust_socketio::Socket;
+    /// use rust_socketio::SocketBuilder;
+    /// use serde_json::json;
     ///
-    /// let mut socket = Socket::new("http://localhost:4200", None);
+    /// let mut socket = SocketBuilder::new("http://localhost:4200")
+    ///     .set_namespace("/admin")
+    ///     .expect("illegal namespace")
+    ///     .on("test", |str| println!("Received: {}", str))
+    ///     .connect()
+    ///     .expect("connection failed");
     ///
-    /// socket.on("foo", |message| println!("{}", message)).unwrap();
-    /// let result = socket.connect();
+    /// // use the socket
+    /// let payload = json!({"token": 123});
+    /// let result = socket.emit("foo", &payload.to_string());
     ///
     /// assert!(result.is_ok());
     /// ```
-    pub fn connect(&mut self) -> Result<()> {
+    pub fn connect(mut self) -> Result<Socket> {
+        self.socket.connect()?;
+        Ok(self.socket)
+    }
+}
+
+impl Socket {
+    /// Creates a socket with a certain adress to connect to as well as a
+    /// namespace. If `None` is passed in as namespace, the default namespace
+    /// `"/"` is taken.
+    /// ```
+    pub(crate) fn new<T: Into<String>>(address: T, namespace: Option<&str>) -> Self {
+        Socket {
+            transport: TransportClient::new(address, namespace.map(String::from)),
+        }
+    }
+
+    /// Registers a new callback for a certain event. This returns an
+    /// `Error::IllegalActionAfterOpen` error if the callback is registered
+    /// after a call to the `connect` method.
+    pub(crate) fn on<F>(&mut self, event: &str, callback: F) -> Result<()>
+    where
+        F: FnMut(String) + 'static + Sync + Send,
+    {
+        self.transport.on(event.into(), callback)
+    }
+
+    /// Connects the client to a server. Afterwards the `emit_*` methods can be
+    /// called to interact with the server. Attention: it's not allowed to add a
+    /// callback after a call to this method.
+    pub(crate) fn connect(&mut self) -> Result<()> {
         self.transport.connect()
     }
 
@@ -168,13 +240,13 @@ impl Socket {
     ///
     /// # Example
     /// ```
-    /// use rust_socketio::Socket;
+    /// use rust_socketio::SocketBuilder;
     /// use serde_json::json;
     ///
-    /// let mut socket = Socket::new("http://localhost:4200", None);
-    ///
-    /// socket.on("foo", |message| println!("{}", message)).unwrap();
-    /// socket.connect().expect("Connection failed");
+    /// let mut socket = SocketBuilder::new("http://localhost:4200")
+    ///     .on("test", |str| println!("Received: {}", str))
+    ///     .connect()
+    ///     .expect("connection failed");
     ///
     /// let payload = json!({"token": 123});
     /// let result = socket.emit("foo", &payload.to_string());
@@ -201,15 +273,16 @@ impl Socket {
     ///
     /// # Example
     /// ```
-    /// use rust_socketio::Socket;
+    /// use rust_socketio::SocketBuilder;
     /// use serde_json::json;
     /// use std::time::Duration;
     /// use std::thread::sleep;
     ///
-    /// let mut socket = Socket::new("http://localhost:4200", None);
+    /// let mut socket = SocketBuilder::new("http://localhost:4200")
+    ///     .on("foo", |str| println!("Received: {}", str))
+    ///     .connect()
+    ///     .expect("connection failed");
     ///
-    /// socket.on("foo", |message| println!("{}", message)).unwrap();
-    /// socket.connect().expect("Connection failed");
     ///
     /// let payload = json!({"token": 123});
     /// let ack_callback = |message| { println!("{}", message) };
@@ -233,6 +306,11 @@ impl Socket {
         self.transport
             .emit_with_ack(event.into(), data, timeout, callback)
     }
+
+    /// Sets the namespace attribute on a client (used by the builder class)
+    pub(crate) fn set_namespace<T: Into<String>>(&mut self, namespace: T) {
+        *Arc::get_mut(&mut self.transport.nsp).unwrap() = Some(namespace.into());
+    }
 }
 
 #[cfg(test)]
@@ -278,4 +356,20 @@ mod test {
 
         sleep(Duration::from_secs(2));
     }
+
+    #[test]
+    fn test_builder() {
+        let socket_builder = SocketBuilder::new(SERVER_URL).set_namespace("/admin");
+
+        assert!(socket_builder.is_ok());
+
+        let socket = socket_builder
+            .unwrap()
+            .on("error", |err| eprintln!("Error!!: {}", err))
+            .on("test", |str| println!("Received: {}", str))
+            .on("message", |msg| println!("Received: {}", msg))
+            .connect();
+
+        assert!(socket.is_ok());
+    }
 }
diff --git a/src/socketio/transport.rs b/src/socketio/transport.rs
index 8c28af15..8f69888e 100644
--- a/src/socketio/transport.rs
+++ b/src/socketio/transport.rs
@@ -41,8 +41,8 @@ pub struct TransportClient {
     engineio_connected: Arc<AtomicBool>,
     on: Arc<Vec<(Event, Callback<String>)>>,
     outstanding_acks: Arc<RwLock<Vec<Ack>>>,
-    // Namespace, for multiplexing messages
-    nsp: Arc<Option<String>>,
+    // namespace, for multiplexing messages
+    pub(crate) nsp: Arc<Option<String>>,
 }
 
 impl TransportClient {


Test patch: