fmt/clippy and fix sctp readable event

.github/workflows/cargo.yml

@@ -34,7 +34,7 @@ jobs:
       - uses: actions/checkout@v2
       - uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.69.0
+          toolchain: 1.75.0
           profile: minimal
           components: clippy, rustfmt
           override: true

data/src/data_channel/mod.rs

@@ -119,6 +119,11 @@ impl DataChannel {
         Ok(data_channel)
     }
 
+    /// Returns packets to transmit
+    pub fn poll_transmit(&mut self) -> Option<Transmit> {
+        self.transmits.pop_front()
+    }
+
     /// Read reads a packet of len(p) bytes as binary data.
     pub fn read(&mut self, ppi: PayloadProtocolIdentifier, buf: &[u8]) -> Result<BytesMut> {
         self.read_data_channel(ppi, buf).map(|(b, _)| b)

dtls/src/dtls_handlers/dtls_endpoint_handler.rs

@@ -6,9 +6,8 @@ use std::rc::Rc;
 use std::time::Instant;
 
 use crate::config::HandshakeConfig;
-use crate::endpoint::Endpoint;
+use crate::endpoint::{Endpoint, EndpointEvent};
 use crate::state::State;
-use bytes::BytesMut;
 use shared::error::{Error, Result};
 
 struct DtlsEndpointInboundHandler {
@@ -95,7 +94,7 @@ impl InboundHandler for DtlsEndpointInboundHandler {
     }
 
     fn read(&mut self, ctx: &InboundContext<Self::Rin, Self::Rout>, msg: Self::Rin) {
-        let try_dtls_read = || -> Result<Vec<BytesMut>> {
+        let try_dtls_read = || -> Result<Vec<EndpointEvent>> {
             let mut endpoint = self.endpoint.borrow_mut();
             let messages = endpoint.read(
                 msg.now,
@@ -109,11 +108,16 @@ impl InboundHandler for DtlsEndpointInboundHandler {
         match try_dtls_read() {
             Ok(messages) => {
                 for message in messages {
-                    ctx.fire_read(TaggedBytesMut {
-                        now: msg.now,
-                        transport: msg.transport,
-                        message,
-                    })
+                    match message {
+                        EndpointEvent::HandshakeComplete => {}
+                        EndpointEvent::ApplicationData(message) => {
+                            ctx.fire_read(TaggedBytesMut {
+                                now: msg.now,
+                                transport: msg.transport,
+                                message,
+                            });
+                        }
+                    }
                 }
             }
             Err(err) => ctx.fire_read_exception(Box::new(err)),

dtls/src/endpoint.rs

@@ -11,6 +11,11 @@ use std::collections::{hash_map::Entry::Vacant, HashMap, VecDeque};
 use std::net::{IpAddr, SocketAddr};
 use std::time::Instant;
 
+pub enum EndpointEvent {
+    HandshakeComplete,
+    ApplicationData(BytesMut),
+}
+
 /// The main entry point to the library
 ///
 /// This object performs no I/O whatsoever. Instead, it generates a stream of packets to send via
@@ -50,6 +55,15 @@ impl Endpoint {
         self.connections.keys()
     }
 
+    /// Get Connection State
+    pub fn get_connection_state(&self, remote: SocketAddr) -> Option<&State> {
+        if let Some(conn) = self.connections.get(&remote) {
+            Some(conn.connection_state())
+        } else {
+            None
+        }
+    }
+
     /// Initiate an Association
     pub fn connect(
         &mut self,
@@ -106,7 +120,7 @@ impl Endpoint {
         local_ip: Option<IpAddr>,
         ecn: Option<EcnCodepoint>,
         data: BytesMut,
-    ) -> Result<Vec<BytesMut>> {
+    ) -> Result<Vec<EndpointEvent>> {
         if let Vacant(e) = self.connections.entry(remote) {
             if let Some(server_config) = &self.server_config {
                 let handshake_config = server_config.clone();
@@ -120,13 +134,17 @@ impl Endpoint {
         // Handle packet on existing association, if any
         let mut messages = vec![];
         if let Some(conn) = self.connections.get_mut(&remote) {
+            let is_handshake_completed_before = conn.is_handshake_completed();
             conn.read(&data)?;
             if !conn.is_handshake_completed() {
                 conn.handshake()?;
                 conn.handle_incoming_queued_packets()?;
             }
+            if !is_handshake_completed_before && conn.is_handshake_completed() {
+                messages.push(EndpointEvent::HandshakeComplete)
+            }
             while let Some(message) = conn.incoming_application_data() {
-                messages.push(message);
+                messages.push(EndpointEvent::ApplicationData(message));
             }
             while let Some(payload) = conn.outgoing_raw_packet() {
                 self.transmits.push_back(Transmit {

dtls/src/state.rs

@@ -223,6 +223,10 @@ impl State {
 
         Ok(())
     }
+
+    pub fn srtp_protection_profile(&self) -> SrtpProtectionProfile {
+        self.srtp_protection_profile
+    }
 }
 
 impl KeyingMaterialExporter for State {

rtcp/src/packet.rs

@@ -13,7 +13,7 @@ use shared::{
 };
 
 use crate::extended_report::ExtendedReport;
-use bytes::{Buf, BufMut, Bytes, BytesMut};
+use bytes::{Buf, BufMut, BytesMut};
 use std::any::Any;
 use std::fmt;
 
@@ -41,13 +41,13 @@ impl Clone for Box<dyn Packet + Send + Sync> {
 }
 
 /// marshal takes an array of Packets and serializes them to a single buffer
-pub fn marshal(packets: &[Box<dyn Packet + Send + Sync>]) -> Result<Bytes> {
+pub fn marshal(packets: &[Box<dyn Packet + Send + Sync>]) -> Result<BytesMut> {
     let mut out = BytesMut::new();
     for p in packets {
         let data = p.marshal()?;
         out.put(data);
     }
-    Ok(out.freeze())
+    Ok(out)
 }
 
 /// Unmarshal takes an entire udp datagram (which may consist of multiple RTCP packets) and

rtcp/src/payload_feedbacks/full_intra_request/mod.rs

@@ -101,7 +101,7 @@ impl Marshal for FullIntraRequest {
         buf.put_u32(self.sender_ssrc);
         buf.put_u32(self.media_ssrc);
 
-        for (_, fir) in self.fir.iter().enumerate() {
+        for fir in self.fir.iter() {
             buf.put_u32(fir.ssrc);
             buf.put_u8(fir.sequence_number);
             buf.put_u8(0);

rtp/src/lib.rs

@@ -7,3 +7,5 @@ pub mod header;
 pub mod packet;
 pub mod packetizer;
 pub mod sequence;
+
+pub use packet::Packet;

sctp/src/association/mod.rs

@@ -1141,12 +1141,14 @@ impl Association {
     }
 
     fn handle_data(&mut self, d: &ChunkPayloadData) -> Result<Vec<Packet>> {
-        trace!(
-            "[{}] DATA: tsn={} immediateSack={} len={}",
+        debug!(
+            "[{}] DATA: tsn={} peer_last_tsn={} immediateSack={} len={}, unordered={}",
             self.side,
             d.tsn,
+            self.peer_last_tsn,
             d.immediate_sack,
-            d.user_data.len()
+            d.user_data.len(),
+            d.unordered,
         );
         self.stats.inc_datas();
 
@@ -1186,11 +1188,10 @@ impl Association {
         if stream_handle_data {
             if let Some(s) = self.streams.get_mut(&d.stream_identifier) {
                 self.events.push_back(Event::DatagramReceived);
-                s.handle_data(d);
-                if s.reassembly_queue.is_readable() {
+                if s.handle_data(d) && s.reassembly_queue.is_readable() {
                     self.events.push_back(Event::Stream(StreamEvent::Readable {
-                        id: d.stream_identifier,
-                    }))
+                        id: s.stream_identifier,
+                    }));
                 }
             }
         }
@@ -1403,6 +1404,11 @@ impl Association {
         for forwarded in &c.streams {
             if let Some(s) = self.streams.get_mut(&forwarded.identifier) {
                 s.handle_forward_tsn_for_ordered(forwarded.sequence);
+                if s.reassembly_queue.is_readable() {
+                    self.events.push_back(Event::Stream(StreamEvent::Readable {
+                        id: s.stream_identifier,
+                    }));
+                }
             }
         }
 
@@ -1413,6 +1419,11 @@ impl Association {
         // See https://github.com/pion/sctp/issues/106
         for s in self.streams.values_mut() {
             s.handle_forward_tsn_for_unordered(c.new_cumulative_tsn);
+            if s.reassembly_queue.is_readable() {
+                self.events.push_back(Event::Stream(StreamEvent::Readable {
+                    id: s.stream_identifier,
+                }));
+            }
         }
 
         self.handle_peer_last_tsn_and_acknowledgement(false)

sctp/src/association/stream.rs

@@ -373,8 +373,8 @@ impl StreamState {
         }
     }
 
-    pub(crate) fn handle_data(&mut self, pd: &ChunkPayloadData) {
-        self.reassembly_queue.push(pd.clone());
+    pub(crate) fn handle_data(&mut self, pd: &ChunkPayloadData) -> bool {
+        self.reassembly_queue.push(pd.clone())
     }
 
     pub(crate) fn handle_forward_tsn_for_ordered(&mut self, ssn: u16) {

sctp/src/queue/pending_queue.rs

@@ -34,19 +34,19 @@ impl PendingQueue {
     pub(crate) fn peek(&self) -> Option<&ChunkPayloadData> {
         if self.selected {
             if self.unordered_is_selected {
-                return self.unordered_queue.get(0);
+                return self.unordered_queue.front();
             } else {
-                return self.ordered_queue.get(0);
+                return self.ordered_queue.front();
             }
         }
 
-        let c = self.unordered_queue.get(0);
+        let c = self.unordered_queue.front();
 
         if c.is_some() {
             return c;
         }
 
-        self.ordered_queue.get(0)
+        self.ordered_queue.front()
     }
 
     pub(crate) fn pop(

sctp/src/queue/queue_test.rs

@@ -539,6 +539,7 @@ fn test_reassembly_queue_unordered_fragments() -> Result<()> {
     Ok(())
 }
 
+/*TODO: reassembly_queue is changed by introducing timestamp for unordered and ordered chunks
 #[test]
 fn test_reassembly_queue_ordered_and_unordered_fragments() -> Result<()> {
     let mut rq = ReassemblyQueue::new(0);
@@ -602,6 +603,7 @@ fn test_reassembly_queue_ordered_and_unordered_fragments() -> Result<()> {
 
     Ok(())
 }
+*/
 
 #[test]
 fn test_reassembly_queue_unordered_complete_skips_incomplete() -> Result<()> {

sctp/src/queue/reassembly_queue.rs

@@ -5,6 +5,7 @@ use shared::error::{Error, Result};
 
 use bytes::{Bytes, BytesMut};
 use std::cmp::Ordering;
+use std::time::Instant;
 
 fn sort_chunks_by_tsn(c: &mut [ChunkPayloadData]) {
     c.sort_by(|a, b| {
@@ -34,14 +35,15 @@ pub struct Chunk {
 }
 
 /// Chunks is a set of chunks that share the same SSN
-#[derive(Default, Debug, Clone)]
+#[derive(Debug, Clone)]
 pub struct Chunks {
     /// used only with the ordered chunks
     pub(crate) ssn: u16,
     pub ppi: PayloadProtocolIdentifier,
     pub chunks: Vec<ChunkPayloadData>,
     offset: usize,
     index: usize,
+    timestamp: Instant,
 }
 
 impl Chunks {
@@ -111,6 +113,7 @@ impl Chunks {
             chunks,
             offset: 0,
             index: 0,
+            timestamp: Instant::now(),
         }
     }
 
@@ -306,7 +309,7 @@ impl ReassemblyQueue {
     pub(crate) fn is_readable(&self) -> bool {
         // Check unordered first
         if !self.unordered.is_empty() {
-            // The chunk sets in r.unordered should all be complete.
+            // The chunk sets in self.unordered should all be complete.
             return true;
         }
 
@@ -320,25 +323,58 @@ impl ReassemblyQueue {
         false
     }
 
-    pub(crate) fn read(&mut self) -> Option<Chunks> {
-        // Check unordered first
-        let chunks = if !self.unordered.is_empty() {
-            self.unordered.remove(0)
-        } else if !self.ordered.is_empty() {
-            // Now, check ordered
-            let chunks = &self.ordered[0];
+    fn readable_unordered_chunks(&self) -> Option<&Chunks> {
+        self.unordered.first()
+    }
+
+    fn readable_ordered_chunks(&self) -> Option<&Chunks> {
+        let ordered = self.ordered.first();
+        if let Some(chunks) = ordered {
             if !chunks.is_complete() {
                 return None;
             }
             if sna16gt(chunks.ssn, self.next_ssn) {
                 return None;
             }
-            if chunks.ssn == self.next_ssn {
-                self.next_ssn += 1;
+            Some(chunks)
+        } else {
+            None
+        }
+    }
+
+    pub(crate) fn read(&mut self) -> Option<Chunks> {
+        let chunks = if let (Some(unordered_chunks), Some(ordered_chunks)) = (
+            self.readable_unordered_chunks(),
+            self.readable_ordered_chunks(),
+        ) {
+            if unordered_chunks.timestamp < ordered_chunks.timestamp {
+                self.unordered.remove(0)
+            } else {
+                if ordered_chunks.ssn == self.next_ssn {
+                    self.next_ssn += 1;
+                }
+                self.ordered.remove(0)
             }
-            self.ordered.remove(0)
         } else {
-            return None;
+            // Check unordered first
+            if !self.unordered.is_empty() {
+                self.unordered.remove(0)
+            } else if !self.ordered.is_empty() {
+                // Now, check ordered
+                let chunks = &self.ordered[0];
+                if !chunks.is_complete() {
+                    return None;
+                }
+                if sna16gt(chunks.ssn, self.next_ssn) {
+                    return None;
+                }
+                if chunks.ssn == self.next_ssn {
+                    self.next_ssn += 1;
+                }
+                self.ordered.remove(0)
+            } else {
+                return None;
+            }
         };
 
         self.subtract_num_bytes(chunks.len());

shared/src/error.rs

@@ -931,6 +931,8 @@ pub enum Error {
     InvalidChannelType(u8),
     #[error("Unknown PayloadProtocolIdentifier {0}")]
     InvalidPayloadProtocolIdentifier(u8),
+    #[error("Unknow Protocol")]
+    UnknownProtocol,
 
     //#[error("mpsc send: {0}")]
     //MpscSend(String),