Fix/compatibility fixes

CHANGELOG.md

@@ -12,8 +12,85 @@ for a staging or local environment must set both fields. Code that builds
 `..ClientConfig::default()`. `DEFAULT_COORDINATOR_WS_URL` moved from
 `rtc::coordinator::ws` to the crate root.
 
+### Call events come in one stream for each source
+
+`Call`, `RtcCall` and `RtcCore` replace `subscribe()`, `on()`, `off()` and the
+`CallEvent` enum with three streams:
+
+- `sfu_events()` gives `SfuCallEvent`: the events from the SFU, including
+  `CallEnded { reason }` for the SFU `call_ended`.
+- `coordinator_events()` gives `CoordinatorEvent`: the call-scoped coordinator
+  events, including `call.ended`.
+- `client_events()` gives `ClientCallEvent`: `CallingStateChanged`.
+
+The SDK leaves the call on the SFU `call_ended` or the coordinator
+`call.ended`; the other one may then not arrive. `CallingStateChanged(Left)` is
+the reliable end of the call. Each stream has its own buffer, so a lagging
+receiver loses events only from its own stream.
+
+### Track events carry the SFU data
+
+`SfuCallEvent::TrackPublished` and `SfuCallEvent::TrackUnpublished` give
+`track_type` as a `TrackType`, not an `i32`, and add `participant`.
+`TrackUnpublished` also adds `cause`. Patterns that match these variants must
+use the new fields or `..`.
+
+### Decoded audio frames carry their RTP timestamp
+
+`PcmFrame` adds `pts: Option<u32>`: the RTP timestamp of the first sample, in
+units of 1/48000 s, wrapping like RTP. `RemoteTrack::next_pcm` sets it; a frame
+rebuilt for a lost packet continues from the frame before it. Frames that the
+application or a conversion builds have `None`, and `write_pcm` ignores the
+field. Code that builds `PcmFrame` with a struct literal must set `pts` or use
+`PcmFrame::new` / `PcmFrame::mono`.
+
+### Subscription config has the stream-py shape
+
+`SubscriptionConfig` replaces `audio`, `video`, `screen_share` and
+`video_dimension` with the fields of the stream-py `SubscriptionConfig`:
+
+- `default: TrackSubscriptionConfig` gives `track_types`, `video_dimension`
+  and `screenshare_dimension`. Screen-share video and screen-share audio are
+  now separate track types, and screen share has its own dimension.
+- `role_filters` gives a rule by participant role. The first role of the
+  participant that has a rule selects it; other participants use `default`.
+- `max_subscriptions` limits the number of tracks. The tracks of the
+  participants that the call learned about first are kept.
+
+`SubscriptionConfig::default()` now subscribes to nothing, and
+`SubscriptionConfig::matches` is removed. The presets `audio_all`,
+`audio_video`, `all` and `none` stay. The default video and screen-share
+dimension is now 1920×1080 (it was 1280×720), also for a `SubscriptionTarget`
+without a dimension. `Call::participants` gives the participants in the order
+the call learned about them. `set_incoming_video_enabled` keeps the configured
+video dimension.
+
 ## New Features
 
+### A token-only client can prepare a call before the join
+
+`RtcClient::call` returns an `RtcCall` that is not joined yet, and
+`RtcCall::join` joins it. Register `on_track` and subscribe before the join to
+get the join events and tracks. `Call::rtc` gives the same `RtcCall` type for a
+client with an API secret; both handles share one session. `RtcCall` also adds
+`update_publish_options` and `set_disconnection_timeout`.
+
+### Stable call event names
+
+`SfuCallEvent::name` gives the stable `SfuEvent` field name of the source event
+(for example `participant_joined` or `call_ended`), and
+`participant_count_changed`. `ClientCallEvent::name` gives
+`calling_state_changed`. A `CoordinatorEvent` has its coordinator `event_type`
+(for example `call.created`).
+
+### Configurable call event buffer
+
+`ClientConfig::call_event_capacity` sets how many events each call event
+stream keeps for a slow receiver. The default stays 256. A larger value makes a
+lag less likely, but each call allocates all slots of its three streams. Code
+that builds `ClientConfig` with a struct literal must set the new field or use
+`..ClientConfig::default()`.
+
 ### Video REST: advanced call statistics and reporting
 
 Application-level stats on `VideoClient` (`get_active_calls_status`,

Cargo.toml

@@ -56,6 +56,7 @@ base64 = "0.23.1"
 bytes = "1.12.1"
 futures-util = "0.3.34"
 hmac = "0.13.0"
+indexmap = "2.14.2"
 prost = "0.14.4"
 prost-types = "0.14.4"
 reqwest = { version = "0.13.4", default-features = false, features = ["rustls", "json", "charset", "http2", "stream"] }

README.md

@@ -31,8 +31,8 @@ remote audio and video, transform it, and publish media back into the call.
   feedback, and client call-event reporting.
 - Join a call as a server-side SFU participant with retry, reconnect, and
   migration handling.
-- Subscribe globally or by participant session to remote audio, video, and
-  screen-share tracks.
+- Subscribe globally, by participant role, or by participant session to remote
+  audio, video, and screen-share tracks.
 - Observe typed participant, connection-quality, pin, grant, and inbound-pause
   state from the SFU.
 - Read Opus audio as PCM, decode VP8/VP9/H264 video as I420, or work with raw
@@ -252,8 +252,9 @@ async fn run_media_bridge(call: &Call) -> Result<(), Box<dyn std::error::Error>>
 For a complete bridge with cancellation, barge-in, audio and video processing,
 and deterministic cleanup, see [`gpt_realtime_bot`](https://github.com/GetStream/stream-video-rust/blob/main/examples/gpt_realtime_bot.rs).
 
-For selective agents, use `Call::update_subscription_targets` with
-`SubscriptionTarget` values instead of subscribing to every participant. A
+For selective agents, set `role_filters` and `max_subscriptions` in
+`SubscriptionConfig`, or use `Call::update_subscription_targets` with
+`SubscriptionTarget` values for an exact list of participant sessions. A
 temporary `mute_track` / `unmute_track` preserves the same local track and
 sender; `stop_publish` remains terminal for that local track handle. The latest
 SFU view is available synchronously through `Call::call_state`.

examples/gpt_realtime_bot.rs

@@ -52,7 +52,7 @@ use getstream::models::{CallRequest, GetOrCreateCallRequest, MemberRequest, User
 use getstream::rtc::proto::models::TrackType;
 use getstream::rtc::{
     JoinCallData, LocalAudioTrack, LocalVideoTrack, RemoteTrack, RtcError, SubscriptionConfig,
-    VideoFrame,
+    TrackSubscriptionConfig, VideoFrame,
 };
 use getstream::video::Call;
 use getstream::{Stream, TokenOptions};
@@ -328,6 +328,17 @@ async fn configure_openai(
             .await
             .context("add audio track to the OpenAI PeerConnection")?,
     );
+    let paced_mic = mic.clone();
+    pc.on_peer_connection_state_change(Box::new(move |state| {
+        let paced_mic = paced_mic.clone();
+        Box::pin(async move {
+            if state == RTCPeerConnectionState::Connected {
+                paced_mic.start_pacing().await;
+            } else {
+                paced_mic.pause_pacing();
+            }
+        })
+    }));
 
     let camera = LocalVideoTrack::h264().context("H264 track for OpenAI")?;
     spawn_rtcp_drain(
@@ -733,8 +744,12 @@ pub async fn start_bot(
 
     if let Err(error) = call
         .update_subscriptions(SubscriptionConfig {
-            video_dimension: Some((640, 360)),
-            ..SubscriptionConfig::audio_video()
+            default: TrackSubscriptionConfig {
+                track_types: vec![TrackType::Audio, TrackType::Video],
+                video_dimension: (640, 360),
+                ..Default::default()
+            },
+            ..Default::default()
         })
         .await
         .context("update_subscriptions")

src/client.rs

@@ -32,6 +32,7 @@ const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
 const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
 const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(55);
 const DEFAULT_MAX_CONNS_PER_HOST: usize = 5;
+const DEFAULT_CALL_EVENT_CAPACITY: usize = 256;
 /// Default maximum body accepted from coordinator HTTP endpoints (16 MiB).
 pub const DEFAULT_MAX_RESPONSE_BODY_BYTES: usize = 16 * 1024 * 1024;
 /// Default maximum inbound SFU/coordinator WebSocket frame and message size (4 MiB).
@@ -120,6 +121,11 @@ pub struct ClientConfig {
     /// and known secret keys are redacted regardless. Other fields, including
     /// PII and secrets stored under custom keys, remain visible when enabled.
     pub log_bodies: bool,
+    /// Events that each call event stream (`sfu_events`, `coordinator_events`,
+    /// `client_events`) keeps for a slow receiver, rounded up to a power of two.
+    /// A receiver that falls further behind loses the oldest events. Each call
+    /// allocates all slots. Must be at least 1. Default 256.
+    pub call_event_capacity: usize,
 }
 
 impl Default for ClientConfig {
@@ -133,6 +139,7 @@ impl Default for ClientConfig {
             max_conns_per_host: DEFAULT_MAX_CONNS_PER_HOST,
             retry: RetryConfig::default(),
             log_bodies: false,
+            call_event_capacity: DEFAULT_CALL_EVENT_CAPACITY,
         }
     }
 }
@@ -148,6 +155,7 @@ pub(crate) struct Client {
     http: reqwest::Client,
     retry: RetryConfig,
     log_bodies: bool,
+    call_event_capacity: usize,
     max_response_body_bytes: usize,
     max_websocket_message_bytes: usize,
     stream_client_header: String,
@@ -206,6 +214,14 @@ impl Client {
                 config.coordinator_ws_url
             )));
         }
+        // tokio broadcast channels panic outside this range.
+        if config.call_event_capacity == 0 || config.call_event_capacity > usize::MAX >> 1 {
+            return Err(Error::Config(format!(
+                "call event capacity {} must be between 1 and {}",
+                config.call_event_capacity,
+                usize::MAX >> 1
+            )));
+        }
 
         let http = reqwest::Client::builder()
             .pool_max_idle_per_host(config.max_conns_per_host)
@@ -227,6 +243,7 @@ impl Client {
             http,
             retry: config.retry,
             log_bodies: config.log_bodies,
+            call_event_capacity: config.call_event_capacity,
             max_response_body_bytes: limits.max_response_body_bytes,
             max_websocket_message_bytes: limits.max_websocket_message_bytes,
             stream_client_header: format!("stream-rust-{}", env!("CARGO_PKG_VERSION")),
@@ -245,6 +262,10 @@ impl Client {
         &self.coordinator_ws_url
     }
 
+    pub(crate) fn call_event_capacity(&self) -> usize {
+        self.call_event_capacity
+    }
+
     /// The shared `reqwest` client (connection pool). Used by the RTC layer to
     /// reuse the pool for coordinator join + SFU Twirp calls.
     pub(crate) fn http(&self) -> &reqwest::Client {
@@ -652,6 +673,17 @@ mod tests {
         assert!(matches!(error, Error::Config(_)));
     }
 
+    #[test]
+    fn call_event_capacity_must_be_at_least_one() {
+        let config = ClientConfig {
+            call_event_capacity: 0,
+            ..ClientConfig::default()
+        };
+        let error = Client::new("key".to_owned(), "secret".to_owned(), config)
+            .expect_err("zero call event capacity");
+        assert!(matches!(error, Error::Config(_)));
+    }
+
     #[test]
     fn compatibility_limits_are_conservative_and_configurable() {
         let limits = NetworkLimits::default();