Support ANNOUNCE_OK and ANNOUNCE_ERROR handler

moqt-core/src/modules/messages/control_messages/announce_ok.rs

@@ -17,6 +17,10 @@ impl AnnounceOk {
     pub fn new(track_namespace: Vec<String>) -> Self {
         Self { track_namespace }
     }
+
+    pub fn track_namespace(&self) -> &Vec<String> {
+        &self.track_namespace
+    }
 }
 
 impl MOQTPayload for AnnounceOk {

moqt-server/src/modules/control_message_handler.rs

@@ -4,6 +4,8 @@
 use crate::modules::handlers::{
     announce_handler::AnnounceResponse, unannounce_handler::unannounce_handler,
 };
+use crate::modules::server_processes::announce_error_message::process_announce_error_message;
+use crate::modules::server_processes::announce_ok_message::process_announce_ok_message;
 use crate::modules::server_processes::subscribe_namespace_message::process_subscribe_namespace_message;
 use crate::modules::server_processes::{
     announce_message::process_announce_message, client_setup_message::process_client_setup_message,
@@ -214,8 +216,38 @@
                 }
             }
         }
-        // ControlMessageType::AnnounceOk => {}
-        // ControlMessageType::AnnounceError => {}
+        ControlMessageType::AnnounceOk => {
+            match process_announce_ok_message(
+                &mut payload_buf,
+                client,
+                pubsub_relation_manager_repository,
+            )
+            .await
+            {
+                Ok(_) => {
+                    return MessageProcessResult::SuccessWithoutResponse;
+                }
+                Err(err) => {
+                    return MessageProcessResult::Failure(
+                        TerminationErrorCode::InternalError,
+                        err.to_string(),
+                    );
+                }
+            }
+        }
+        ControlMessageType::AnnounceError => {
+            match process_announce_error_message(&mut payload_buf).await {
+                Ok(_) => {
+                    return MessageProcessResult::SuccessWithoutResponse;
+                }
+                Err(err) => {
+                    return MessageProcessResult::Failure(
+                        TerminationErrorCode::InternalError,
+                        err.to_string(),
+                    );
+                }
+            }
+        }
         ControlMessageType::SubscribeNamespace => {
             match process_subscribe_namespace_message(
                 &mut payload_buf,

moqt-server/src/modules/handlers.rs

@@ -1,4 +1,5 @@
 pub(crate) mod announce_handler;
+pub(crate) mod announce_ok_handler;
 pub(crate) mod server_setup_handler;
 pub(crate) mod stream_track_header_handler;
 pub(crate) mod subscribe_handler;

moqt-server/src/modules/handlers/announce_ok_handler.rs

@@ -0,0 +1,71 @@
+use anyhow::Result;
+use moqt_core::pubsub_relation_manager_repository::PubSubRelationManagerRepository;
+use moqt_core::{messages::control_messages::announce_ok::AnnounceOk, MOQTClient};
+
+pub(crate) async fn announce_ok_handler(
+    announce_ok_message: AnnounceOk,
+    client: &mut MOQTClient,
+    pubsub_relation_manager_repository: &mut dyn PubSubRelationManagerRepository,
+) -> Result<()> {
+    tracing::trace!("announce_ok_handler start.");
+    tracing::debug!("announce_ok_message: {:#?}", announce_ok_message);
+
+    // Record the announced Track Namespace
+    pubsub_relation_manager_repository
+        .set_downstream_announced_namespace(
+            announce_ok_message.track_namespace().clone(),
+            client.id,
+        )
+        .await?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod success {
+    use crate::modules::handlers::announce_ok_handler::announce_ok_handler;
+    use crate::modules::pubsub_relation_manager::{
+        commands::PubSubRelationCommand, manager::pubsub_relation_manager,
+        wrapper::PubSubRelationManagerWrapper,
+    };
+    use moqt_core::messages::moqt_payload::MOQTPayload;
+    use moqt_core::pubsub_relation_manager_repository::PubSubRelationManagerRepository;
+    use moqt_core::{messages::control_messages::announce_ok::AnnounceOk, moqt_client::MOQTClient};
+    use tokio::sync::mpsc;
+
+    #[tokio::test]
+    async fn normal_case() {
+        // Generate ANNOUNCE message
+        let track_namespace = Vec::from(["test".to_string(), "test".to_string()]);
+        let announce_ok_message = AnnounceOk::new(track_namespace.clone());
+        let mut buf = bytes::BytesMut::new();
+        announce_ok_message.packetize(&mut buf);
+
+        // Generate client
+        let downstream_session_id = 0;
+        let mut client = MOQTClient::new(downstream_session_id);
+
+        // Generate PubSubRelationManagerWrapper
+        let (track_namespace_tx, mut track_namespace_rx) =
+            mpsc::channel::<PubSubRelationCommand>(1024);
+        tokio::spawn(async move { pubsub_relation_manager(&mut track_namespace_rx).await });
+        let mut pubsub_relation_manager: PubSubRelationManagerWrapper =
+            PubSubRelationManagerWrapper::new(track_namespace_tx);
+
+        let max_subscribe_id = 10;
+
+        let _ = pubsub_relation_manager
+            .setup_subscriber(max_subscribe_id, downstream_session_id)
+            .await;
+
+        // Execute announce_ok_handler and get result
+        let result = announce_ok_handler(
+            announce_ok_message,
+            &mut client,
+            &mut pubsub_relation_manager,
+        )
+        .await;
+
+        assert!(result.is_ok());
+    }
+}

moqt-server/src/modules/server_processes.rs

@@ -1,4 +1,6 @@
+pub(crate) mod announce_error_message;
 pub(crate) mod announce_message;
+pub(crate) mod announce_ok_message;
 pub(crate) mod client_setup_message;
 pub(crate) mod stream_track_header;
 pub(crate) mod subscribe_message;

moqt-server/src/modules/server_processes/announce_error_message.rs

@@ -0,0 +1,19 @@
+use anyhow::{bail, Result};
+use bytes::BytesMut;
+use moqt_core::messages::{
+    control_messages::announce_error::AnnounceError, moqt_payload::MOQTPayload,
+};
+
+pub(crate) async fn process_announce_error_message(payload_buf: &mut BytesMut) -> Result<()> {
+    let announce_error_message = match AnnounceError::depacketize(payload_buf) {
+        Ok(announce_error_message) => announce_error_message,
+        Err(err) => {
+            tracing::error!("{:#?}", err);
+            bail!(err.to_string());
+        }
+    };
+
+    tracing::warn!("announce_error_message: {:#?}", announce_error_message);
+
+    Ok(())
+}

moqt-server/src/modules/server_processes/announce_ok_message.rs

@@ -0,0 +1,29 @@
+use crate::modules::handlers::announce_ok_handler::announce_ok_handler;
+use anyhow::{bail, Result};
+use bytes::BytesMut;
+use moqt_core::pubsub_relation_manager_repository::PubSubRelationManagerRepository;
+use moqt_core::{
+    messages::{control_messages::announce_ok::AnnounceOk, moqt_payload::MOQTPayload},
+    MOQTClient,
+};
+
+pub(crate) async fn process_announce_ok_message(
+    payload_buf: &mut BytesMut,
+    client: &mut MOQTClient,
+    pubsub_relation_manager_repository: &mut dyn PubSubRelationManagerRepository,
+) -> Result<()> {
+    let announce_ok_message = match AnnounceOk::depacketize(payload_buf) {
+        Ok(announce_ok_message) => announce_ok_message,
+        Err(err) => {
+            tracing::error!("{:#?}", err);
+            bail!(err.to_string());
+        }
+    };
+
+    announce_ok_handler(
+        announce_ok_message,
+        client,
+        pubsub_relation_manager_repository,
+    )
+    .await
+}