WIP: Close when a new initiator has connected

Note that this deviates from the intention of the specification to
allow for more than one connection towards an initiator over the same
WebSocket connection.

Author: lgrahl

examples/chat/main.rs

@@ -243,8 +243,8 @@ fn main() {
 
     // Connect to server
     let (connect_future, event_channel) = saltyrtc_client::connect(
-            "localhost",
-            8765,
+            "server.saltyrtc.org",
+            443,
             Some(tls_connector),
             &core.handle(),
             salty_arc.clone(),
@@ -537,7 +537,7 @@ fn setup_logging(role: Role, log_to_stdout: bool) -> Config {
         .unwrap();
 
     // Instantiate filters
-    let info_filter = ThresholdFilter::new(LevelFilter::Info);
+    let info_filter = ThresholdFilter::new(LevelFilter::Debug);
 
     // Config builder
     let builder = Config::builder()

src/lib.rs

@@ -643,6 +643,9 @@ pub fn do_handshake(
                                 late_error = Some(e);
                             }
                         },
+                        HandleAction::Close(code) => {
+                            messages.push(OwnedMessage::Close(Some(CloseData::new(code.as_number(), "".to_string()))))
+                        },
                     }
                 }
 
@@ -769,6 +772,10 @@ pub fn task_loop(
                                 HandleAction::Reply(bbox) => out_messages.push(OwnedMessage::Binary(bbox.into_bytes())),
                                 HandleAction::TaskMessage(msg) => {
                                     if let TaskMessage::Close(_) = msg {
+                                        out_messages.push(OwnedMessage::Close(Some(CloseData {
+                                            status_code: 1000,
+                                            reason: "".to_string(),
+                                        })));
                                         close_stream = true;
                                     }
 
@@ -790,6 +797,13 @@ pub fn task_loop(
                                 HandleAction::HandshakeError(_) => return boxed!(future::err(Err(
                                     SaltyError::Crash("Got HandleAction::HandshakeError in task loop".into())
                                 ))),
+                                HandleAction::Close(code) => {
+                                    out_messages.push(OwnedMessage::Close(Some(CloseData {
+                                        status_code: code.as_number(),
+                                        reason: "".to_string(),
+                                    })));
+                                    close_stream = true;
+                                },
                             }
                         }
 
@@ -824,6 +838,8 @@ pub fn task_loop(
                             out_future
                                 .join(in_future)
                                 .and_then(move |_| if close_stream {
+                                    // TODO(dbrgn): Tear down all futures to stop the loop
+
                                     // Stop processing stream
                                     Err(Ok(()))
                                 } else {

src/protocol/mod.rs

@@ -1808,7 +1808,16 @@ impl Signaling for ResponderSignaling {
     fn handle_new_initiator(&mut self, _msg: NewInitiator) -> SignalingResult<Vec<HandleAction>> {
         debug!("--> Received new-initiator from server");
 
-        let mut actions: Vec<HandleAction> = vec![];
+        // Close when a new initiator has connected.
+        //
+        // Note: This deviates from the intention of the specification to allow
+        //       for more than one connection towards an initiator over the same
+        //       WebSocket connection.
+        let signaling_state = self.common().signaling_state();
+        if signaling_state == SignalingState::Task {
+            debug!("Received new-initiator message in state {:?}, closing", signaling_state);
+            return Ok(vec![HandleAction::Close(CloseCode::WsClosingNormal)])
+        }
 
         // A responder who receives a 'new-initiator' message MUST proceed by
         // deleting all currently cached information about and for the previous
@@ -1829,6 +1838,7 @@ impl Signaling for ResponderSignaling {
                 return Err(SignalingError::Crash("No auth provider set".into()));
             },
         }
+        let mut actions: Vec<HandleAction> = vec![];
         if send_token {
             let old_auth_provider = mem::replace(&mut self.common_mut().auth_provider, None);
             if let Some(AuthProvider::Token(token)) = old_auth_provider {

src/protocol/tests/signaling_messages.rs

@@ -531,6 +531,7 @@ mod client_auth {
             HandleAction::HandshakeError(_) => panic!("Unexpected HandshakeError"),
             HandleAction::TaskMessage(_) => panic!("Unexpected TaskMessage"),
             HandleAction::Event(_) => panic!("Unexpected Event"),
+            HandleAction::Close(_) => panic!("Unexpected Close"),
         };
 
         let decrypted = OpenBox::<Message>::decrypt(

src/protocol/types.rs

@@ -5,7 +5,7 @@ use std::result::Result as StdResult;
 use serde::ser::{Serialize, Serializer};
 use serde::de::{Deserialize, Deserializer, Visitor, Error as SerdeError};
 
-use crate::Event;
+use crate::{Event, CloseCode};
 use crate::boxes::ByteBox;
 use crate::errors::SaltyError;
 use crate::tasks::TaskMessage;
@@ -237,6 +237,8 @@ pub(crate) enum HandleAction {
     Event(Event),
     /// A task message was received and decoded.
     TaskMessage(TaskMessage),
+    /// Close the websocket with a specific close code.
+    Close(CloseCode),
 }