feat(connection): add decoder configuration handling and notification…

alvr/client_core/src/connection.rs

@@ -15,8 +15,9 @@ use alvr_common::{
 };
 use alvr_packets::{
     AUDIO, ClientConnectionResult, ClientControlPacket, ClientStatistics, ConnectionAcceptedInfo,
-    HAPTICS, Haptics, STATISTICS, ServerControlPacket, StreamConfigPacket, TRACKING, TrackingData,
-    VIDEO, VideoPacketHeader, VideoStreamingCapabilities, VideoStreamingCapabilitiesExt,
+    DecoderInitializationConfig, HAPTICS, Haptics, STATISTICS, ServerControlPacket,
+    StreamConfigPacket, TRACKING, TrackingData, VIDEO, VideoPacketHeader,
+    VideoStreamingCapabilities, VideoStreamingCapabilitiesExt,
 };
 use alvr_session::{SocketProtocol, settings_schema::Switch};
 use alvr_sockets::{
@@ -64,7 +65,9 @@ pub struct ConnectionContext {
     pub tracking_sender: Mutex<Option<StreamSender<TrackingData>>>,
     pub statistics_sender: Mutex<Option<StreamSender<ClientStatistics>>>,
     pub statistics_manager: Mutex<Option<StatisticsManager>>,
+    pub decoder_config: Mutex<Option<DecoderInitializationConfig>>,
     pub decoder_callback: Mutex<Option<Box<DecoderCallback>>>,
+    pub decoder_callback_notifier: Condvar,
     pub video_frame_metadata_queue: Mutex<VecDeque<(Duration, VideoFrameMetadata)>>,
     pub max_prediction: RwLock<Duration>,
 }
@@ -82,6 +85,24 @@ fn set_hud_message(event_queue: &Mutex<VecDeque<ClientCoreEvent>>, message: &str
         .push_back(ClientCoreEvent::UpdateHudMessage(message));
 }
 
+fn notify_decoder_config(
+    ctx: &ConnectionContext,
+    event_queue: &Mutex<VecDeque<ClientCoreEvent>>,
+    config: DecoderInitializationConfig,
+) {
+    let mut current_config = ctx.decoder_config.lock();
+    if current_config.as_ref() != Some(&config) {
+        *current_config = Some(config.clone());
+        *ctx.decoder_callback.lock() = None;
+        event_queue
+            .lock()
+            .push_back(ClientCoreEvent::DecoderConfig {
+                codec: config.codec,
+                config_nal: config.config_buffer,
+            });
+    }
+}
+
 fn is_streaming(ctx: &ConnectionContext) -> bool {
     *ctx.state.read() == ConnectionState::Streaming
 }
@@ -273,9 +294,12 @@ fn connection_pipeline(
     let statistics_sender = stream_socket.request_stream(STATISTICS);
 
     ctx.video_frame_metadata_queue.lock().clear();
+    *ctx.decoder_config.lock() = None;
+    *ctx.decoder_callback.lock() = None;
 
     let video_receive_thread = thread::spawn({
         let ctx = Arc::clone(&ctx);
+        let event_queue = Arc::clone(&event_queue);
         move || {
             let mut stream_corrupted = true;
             while is_streaming(&ctx) {
@@ -292,6 +316,10 @@ fn connection_pipeline(
                     stats.report_video_packet_received(header.timestamp);
                 }
 
+                if let Some(config) = header.decoder_config {
+                    notify_decoder_config(&ctx, &event_queue, config);
+                }
+
                 if header.is_idr {
                     stream_corrupted = false;
                 } else if data.had_packet_loss() {
@@ -319,12 +347,25 @@ fn connection_pipeline(
                         }
                     }
 
-                    let submitted = ctx
-                        .decoder_callback
-                        .lock()
+                    let mut decoder_callback_guard = ctx.decoder_callback.lock();
+                    if decoder_callback_guard.is_none() && is_streaming(&ctx) {
+                        let deadline = Instant::now() + STREAMING_RECV_TIMEOUT;
+                        while decoder_callback_guard.is_none() && is_streaming(&ctx) {
+                            let now = Instant::now();
+                            if now >= deadline {
+                                break;
+                            }
+                            ctx.decoder_callback_notifier
+                                .wait_for(&mut decoder_callback_guard, deadline - now);
+                        }
+                    }
+
+                    let submitted = decoder_callback_guard
                         .as_mut()
                         .is_some_and(|callback| callback(header.timestamp, nal));
 
+                    drop(decoder_callback_guard);
+
                     if !submitted {
                         stream_corrupted = true;
                         if let Some(sender) = &mut *ctx.control_sender.lock() {
@@ -481,12 +522,7 @@ fn connection_pipeline(
 
                 match maybe_packet {
                     Ok(ServerControlPacket::DecoderConfig(config)) => {
-                        event_queue
-                            .lock()
-                            .push_back(ClientCoreEvent::DecoderConfig {
-                                codec: config.codec,
-                                config_nal: config.config_buffer,
-                            });
+                        notify_decoder_config(&ctx, &event_queue, config);
                     }
                     Ok(ServerControlPacket::Restarting) => {
                         info!("{SERVER_RESTART_MESSAGE}");
@@ -570,6 +606,9 @@ fn connection_pipeline(
     *ctx.tracking_sender.lock() = None;
     *ctx.statistics_sender.lock() = None;
     *LOG_CHANNEL_SENDER.lock() = None;
+    *ctx.decoder_config.lock() = None;
+    *ctx.decoder_callback.lock() = None;
+    ctx.decoder_callback_notifier.notify_all();
 
     event_queue
         .lock()

alvr/client_core/src/lib.rs

@@ -262,10 +262,9 @@ impl ClientCoreContext {
         dbg_client_core!("set_decoder_input_callback");
 
         *self.connection_context.decoder_callback.lock() = Some(callback);
-
-        if let Some(sender) = &mut *self.connection_context.control_sender.lock() {
-            sender.send(&ClientControlPacket::RequestIdr).ok();
-        }
+        self.connection_context
+            .decoder_callback_notifier
+            .notify_all();
     }
 
     pub fn report_frame_decoded(&self, timestamp: Duration) {

alvr/packets/src/lib.rs

@@ -149,7 +149,7 @@ impl StreamConfigPacket {
     }
 }
 
-#[derive(Serialize, Deserialize, Clone)]
+#[derive(Serialize, Deserialize, Clone, PartialEq, Eq, Debug)]
 pub struct DecoderInitializationConfig {
     pub codec: CodecType,
     pub config_buffer: Vec<u8>, // e.g. SPS + PPS NALs
@@ -247,6 +247,7 @@ pub struct VideoPacketHeader {
     /// Centers used to encode this frame, already aligned. Normally absent when FFR is disabled.
     pub foveation_center_shifts: Option<[[f32; 2]; 2]>,
     pub is_idr: bool,
+    pub decoder_config: Option<DecoderInitializationConfig>,
 }
 
 #[derive(Serialize, Deserialize)]

alvr/server_core/src/lib.rs

@@ -452,12 +452,19 @@ impl ServerCoreContext {
                     file.write_all(&nal_buffer).ok();
                 }
 
+                let decoder_config = if is_idr {
+                    self.connection_context.decoder_config.lock().clone()
+                } else {
+                    None
+                };
+
                 let sender_result = sender.try_send(VideoPacket {
                     header: VideoPacketHeader {
                         timestamp,
                         global_view_params,
                         foveation_center_shifts,
                         is_idr,
+                        decoder_config,
                     },
                     payload: nal_buffer,
                 });