Respond with ParamOutgoingResetRequest

Author: Marcel-G

examples/examples/data-channels/data-channels.rs

@@ -126,6 +126,11 @@ async fn main() -> Result<()> {
                 let d2 = Arc::clone(&d);
                 let d_label2 = d_label.clone();
                 let d_id2 = d_id;
+                d.on_close(Box::new(move || {
+                    println!("Data channel closed");
+                    Box::pin(async {})
+                }));
+
                 d.on_open(Box::new(move || {
                     println!("Data channel '{d_label2}'-'{d_id2}' open. Random messages will now be sent to any connected DataChannels every 5 seconds");
 

sctp/src/association/association_internal.rs

@@ -975,9 +975,7 @@ impl AssociationInternal {
             let rst_reqs: Vec<ParamOutgoingResetRequest> =
                 self.reconfig_requests.values().cloned().collect();
             for rst_req in rst_reqs {
-                let resp = self.reset_streams_if_any(&rst_req);
-                log::debug!("[{}] RESET RESPONSE: {}", self.name, resp);
-                reply.push(resp);
+                self.reset_streams_if_any(&rst_req, false, &mut reply)?;
             }
         }
 
@@ -1630,15 +1628,11 @@ impl AssociationInternal {
         let mut pp = vec![];
 
         if let Some(param_a) = &c.param_a {
-            if let Some(p) = self.handle_reconfig_param(param_a).await? {
-                pp.push(p);
-            }
+            self.handle_reconfig_param(param_a, &mut pp).await?;
         }
 
         if let Some(param_b) = &c.param_b {
-            if let Some(p) = self.handle_reconfig_param(param_b).await? {
-                pp.push(p);
-            }
+            self.handle_reconfig_param(param_b, &mut pp).await?;
         }
 
         Ok(pp)
@@ -1751,26 +1745,35 @@ impl AssociationInternal {
     async fn handle_reconfig_param(
         &mut self,
         raw: &Box<dyn Param + Send + Sync>,
-    ) -> Result<Option<Packet>> {
+        reply: &mut Vec<Packet>,
+    ) -> Result<()> {
         if let Some(p) = raw.as_any().downcast_ref::<ParamOutgoingResetRequest>() {
             self.reconfig_requests
                 .insert(p.reconfig_request_sequence_number, p.clone());
-            Ok(Some(self.reset_streams_if_any(p)))
+            self.reset_streams_if_any(p, true, reply)?;
+            Ok(())
         } else if let Some(p) = raw.as_any().downcast_ref::<ParamReconfigResponse>() {
             self.reconfigs.remove(&p.reconfig_response_sequence_number);
             if self.reconfigs.is_empty() {
                 if let Some(treconfig) = &self.treconfig {
                     treconfig.stop().await;
                 }
             }
-            Ok(None)
+            Ok(())
         } else {
             Err(Error::ErrParamterType)
         }
     }
 
-    fn reset_streams_if_any(&mut self, p: &ParamOutgoingResetRequest) -> Packet {
+    fn reset_streams_if_any(
+        &mut self,
+        p: &ParamOutgoingResetRequest,
+        respond: bool,
+        reply: &mut Vec<Packet>,
+    ) -> Result<()> {
         let mut result = ReconfigResult::SuccessPerformed;
+        let mut sis_to_reset = vec![];
+
         if sna32lte(p.sender_last_tsn, self.peer_last_tsn) {
             log::debug!(
                 "[{}] resetStream(): senderLastTSN={} <= peer_last_tsn={}",
@@ -1781,6 +1784,11 @@ impl AssociationInternal {
             for id in &p.stream_identifiers {
                 if let Some(s) = self.streams.get(id) {
                     let stream_identifier = s.stream_identifier;
+                    if respond {
+                        sis_to_reset.push(*id);
+                    }
+                    // Someon
+                    // https://github.com/pion/sctp/blob/master/association.go#L2045
                     self.unregister_stream(stream_identifier);
                 }
             }
@@ -1796,13 +1804,42 @@ impl AssociationInternal {
             result = ReconfigResult::InProgress;
         }
 
-        self.create_packet(vec![Box::new(ChunkReconfig {
+        // Answer incoming reset requests with the same reset request, but with
+        // reconfig_response_sequence_number.
+        if !sis_to_reset.is_empty() {
+            let rsn = self.generate_next_rsn();
+            let tsn = self.my_next_tsn - 1;
+
+            let c = ChunkReconfig {
+                param_a: Some(Box::new(ParamOutgoingResetRequest {
+                    reconfig_request_sequence_number: rsn,
+                    reconfig_response_sequence_number: p.reconfig_request_sequence_number,
+                    sender_last_tsn: tsn,
+                    stream_identifiers: sis_to_reset,
+                    ..Default::default()
+                })),
+                ..Default::default()
+            };
+
+            self.reconfigs.insert(rsn, c.clone()); // store in the map for retransmission
+
+            let p = self.create_packet(vec![Box::new(c)]);
+            reply.push(p);
+        }
+
+        let packet = self.create_packet(vec![Box::new(ChunkReconfig {
             param_a: Some(Box::new(ParamReconfigResponse {
                 reconfig_response_sequence_number: p.reconfig_request_sequence_number,
                 result,
             })),
             param_b: None,
-        })])
+        })]);
+
+        log::debug!("[{}] RESET RESPONSE: {}", self.name, packet);
+
+        reply.push(packet);
+
+        Ok(())
     }
 
     /// Move the chunk peeked with self.pending_queue.peek() to the inflight_queue.