Add a method to the direct api to retrieve the sctp max message size

src/change/direct.rs

@@ -284,6 +284,13 @@ impl<'a> DirectApi<'a> {
         self.rtc.session.streams.stream_rx_by_midrid(midrid, true)
     }
 
+    /// Get sctp max-message-size
+    ///
+    /// This returns the maximum buffer size that channel.write will accept
+    pub fn sctp_max_message_size(&self) -> u32 {
+        self.rtc.sctp.remote_max_message_size()
+    }
+
     /// Declare the intention to send data using the given SSRC.
     ///
     /// * The resend RTX is optional but necessary to do resends. str0m does not do

src/lib.rs

@@ -1544,7 +1544,11 @@ impl Rtc {
     ///
     /// See [`Rtc`] instance documentation for how this is expected to be used in a loop.
     pub fn poll_output(&mut self) -> Result<Output, RtcError> {
-        let o = self.do_poll_output()?;
+        let o = loop {
+            if let Some(o) = self.do_poll_output()? {
+                break o;
+            }
+        };
 
         match &o {
             Output::Event(e) => match e {
@@ -1572,10 +1576,10 @@ impl Rtc {
         Ok(o)
     }
 
-    fn do_poll_output(&mut self) -> Result<Output, RtcError> {
+    fn do_poll_output(&mut self) -> Result<Option<Output>, RtcError> {
         if self.state == RtcState::Closed {
             self.last_timeout_reason = Reason::NotHappening;
-            return Ok(Output::Timeout(not_happening()));
+            return Ok(Some(Output::Timeout(not_happening())));
         }
 
         while let Some(e) = self.ice.poll_event() {
@@ -1584,7 +1588,7 @@ impl Rtc {
                     //
                 }
                 IceAgentEvent::IceConnectionStateChange(v) => {
-                    return Ok(Output::Event(Event::IceConnectionStateChange(v)));
+                    return Ok(Some(Output::Event(Event::IceConnectionStateChange(v))));
                 }
                 IceAgentEvent::DiscoveredRecv { proto, source } => {
                     debug!("ICE remote address: {:?}/{:?}", Pii(source), proto);
@@ -1678,7 +1682,7 @@ impl Rtc {
                 }
                 DtlsOutput::CloseNotify => {
                     self.start_close()?;
-                    return Ok(Output::Event(Event::Closed));
+                    return Ok(Some(Output::Event(Event::Closed)));
                 }
                 other => {
                     return Err(RtcError::Dtls(DtlsError::Io(std::io::Error::other(
@@ -1689,7 +1693,7 @@ impl Rtc {
         }
 
         if just_connected {
-            return Ok(Output::Event(Event::Connected));
+            return Ok(Some(Output::Event(Event::Connected)));
         }
 
         while let Some(e) = self.sctp.poll() {
@@ -1723,63 +1727,63 @@ impl Rtc {
 
                         // Run again since this would feed the DTLS subsystem
                         // to produce a packet now.
-                        return self.do_poll_output();
+                        return Ok(None);
                     }
                 }
                 SctpEvent::Open { id, label } => {
                     self.chan.ensure_channel_id_for(id);
                     let id = self.chan.channel_id_by_stream_id(id).unwrap();
-                    return Ok(Output::Event(Event::ChannelOpen(id, label)));
+                    return Ok(Some(Output::Event(Event::ChannelOpen(id, label))));
                 }
                 SctpEvent::Close { id } => {
                     let Some(id) = self.chan.channel_id_by_stream_id(id) else {
                         warn!("Drop ChannelClose event for id: {:?}", id);
                         continue;
                     };
                     self.chan.remove_channel(id, self.last_now);
-                    return Ok(Output::Event(Event::ChannelClose(id)));
+                    return Ok(Some(Output::Event(Event::ChannelClose(id))));
                 }
                 SctpEvent::AssociationLost => {
                     self.start_close()?;
-                    return Ok(Output::Event(Event::Closed));
+                    return Ok(Some(Output::Event(Event::Closed)));
                 }
                 SctpEvent::Data { id, binary, data } => {
                     let Some(id) = self.chan.channel_id_by_stream_id(id) else {
                         warn!("Drop ChannelData event for id: {:?}", id);
                         continue;
                     };
                     let cd = ChannelData { id, binary, data };
-                    return Ok(Output::Event(Event::ChannelData(cd)));
+                    return Ok(Some(Output::Event(Event::ChannelData(cd))));
                 }
                 SctpEvent::BufferedAmountLow { id } => {
                     let Some(id) = self.chan.channel_id_by_stream_id(id) else {
                         warn!("Drop BufferedAmountLow for id: {:?}", id);
                         continue;
                     };
-                    return Ok(Output::Event(Event::ChannelBufferedAmountLow(id)));
+                    return Ok(Some(Output::Event(Event::ChannelBufferedAmountLow(id))));
                 }
             }
         }
 
         if let Some(ev) = self.session.poll_event() {
-            return Ok(Output::Event(ev));
+            return Ok(Some(Output::Event(ev)));
         }
 
         // Some polling needs to bubble up errors.
         if let Some(ev) = self.session.poll_event_fallible()? {
-            return Ok(Output::Event(ev));
+            return Ok(Some(Output::Event(ev)));
         }
 
         if let Some(e) = self.stats.as_mut().and_then(|s| s.poll_output()) {
-            return Ok(match e {
+            return Ok(Some(match e {
                 StatsEvent::Peer(s) => Output::Event(Event::PeerStats(s)),
                 StatsEvent::MediaIngress(s) => Output::Event(Event::MediaIngressStats(s)),
                 StatsEvent::MediaEgress(s) => Output::Event(Event::MediaEgressStats(s)),
-            });
+            }));
         }
 
         if let Some(v) = self.ice.poll_transmit() {
-            return Ok(Output::Transmit(v));
+            return Ok(Some(Output::Transmit(v)));
         }
 
         if let Some(send) = &self.send_addr {
@@ -1795,7 +1799,7 @@ impl Rtc {
                     destination: send.destination,
                     contents,
                 };
-                return Ok(Output::Transmit(t));
+                return Ok(Some(Output::Transmit(t)));
             }
         } else {
             // Don't allow accumulated feedback to build up indefinitely
@@ -1827,11 +1831,11 @@ impl Rtc {
         if self.state == RtcState::Closing && self.close_drain_complete() {
             self.state = RtcState::Closed;
             self.last_timeout_reason = Reason::NotHappening;
-            return Ok(Output::Timeout(not_happening()));
+            return Ok(Some(Output::Timeout(not_happening())));
         }
 
         self.last_timeout_reason = reason;
-        Ok(Output::Timeout(next))
+        Ok(Some(Output::Timeout(next)))
     }
 
     /// The reason for the last [`Output::Timeout`]

src/sctp/mod.rs

@@ -27,14 +27,17 @@ mod error;
 pub use error::SctpError;
 
 /// Bytes that can be buffered inside str0m across all streams.
-const MAX_BUFFERED_ACROSS_STREAMS: usize = 128 * 1024;
+const DEFAULT_MAX_BUFFERED_ACROSS_STREAMS: usize = 128 * 1024;
 
 /// Maximum message size we advertise in SDP (what we can receive)
 pub const LOCAL_MAX_MESSAGE_SIZE: u32 = 256 * 1024;
 
 /// Default max message size if remote doesn't advertise
 pub const DEFAULT_REMOTE_MAX_MESSAGE_SIZE: u32 = 64 * 1024;
 
+/// Headroom for max_buffered_across_streams vs remote_max_message_size
+pub const BUFFER_MULTIPLIER: u32 = 2;
+
 pub(crate) struct RtcSctp {
     state: RtcSctpState,
     endpoint: Endpoint,
@@ -46,6 +49,7 @@ pub(crate) struct RtcSctp {
     last_now: Instant,
     client: bool,
     remote_max_message_size: u32,
+    max_buffered_across_streams: usize,
     snap_enabled: bool,
     snap_init: Option<SctpInitData>,
     #[cfg(test)]
@@ -279,6 +283,7 @@ impl RtcSctp {
             last_now: Instant::now(), // placeholder until init()
             client: false,
             remote_max_message_size: DEFAULT_REMOTE_MAX_MESSAGE_SIZE,
+            max_buffered_across_streams: DEFAULT_MAX_BUFFERED_ACROSS_STREAMS,
             snap_enabled: false,
             snap_init: None,
             #[cfg(test)]
@@ -323,6 +328,12 @@ impl RtcSctp {
             self.remote_max_message_size = max_msg_size;
         }
 
+        let bufsize: usize =
+            BUFFER_MULTIPLIER.saturating_mul(self.remote_max_message_size) as usize;
+        if self.max_buffered_across_streams < bufsize {
+            self.max_buffered_across_streams = bufsize;
+        }
+
         if let Some(snap_data) = sctp_init_data {
             // SNAP path: both local and remote INIT chunks must be present.
             if snap_data.local_init.is_none() || snap_data.remote_init.is_none() {
@@ -341,6 +352,7 @@ impl RtcSctp {
                 .connect(config, self.fake_addr)
                 .map_err(|e| SctpError::Proto(ProtoError::Other(e.to_string())))?;
             assoc.set_max_send_message_size(self.remote_max_message_size);
+
             self.handle = handle;
             self.assoc = Some(assoc);
 
@@ -569,7 +581,7 @@ impl RtcSctp {
             })
             .sum();
 
-        MAX_BUFFERED_ACROSS_STREAMS - total
+        self.max_buffered_across_streams - total
     }
 
     pub fn write(&mut self, id: u16, binary: bool, buf: &[u8]) -> Result<usize, SctpError> {
@@ -1008,7 +1020,6 @@ impl RtcSctp {
             .and_then(|s| s.config.as_ref())
     }
 
-    #[cfg(test)]
     pub(crate) fn remote_max_message_size(&self) -> u32 {
         self.remote_max_message_size
     }