[WIP] Feat/pub sub

.gitignore

@@ -26,4 +26,6 @@ cargo-flamegraph.trace/*
 key.pem
 cert.pem
 
-Cargo.lock
+Cargo.lock
+
+**.log

.vscode/settings.json

@@ -11,14 +11,15 @@
   "rust-analyzer.check.command": "clippy",
   "rust-analyzer.cargo.extraEnv": {
     "RUSTFLAGS": "--cfg web_sys_unstable_apis --cfg tokio_unstable"
+    // "RUST_BACKTRACE": "1"
   },
   "rust-analyzer.cargo.features": [
     // "web_sys_unstable_apis"
   ],
   "rust-analyzer.linkedProjects": [
-    "moqt-wasm/Cargo.toml",
-    "moqt-wasm/moqt/Cargo.toml",
-    "moqt-wasm/sample/client/Cargo.toml",
-    "moqt-wasm/sample/server/Cargo.toml"
+    "Cargo.toml",
+    "moqt/Cargo.toml",
+    "client/Cargo.toml",
+    "relay/Cargo.toml"
   ]
 }

Cargo.toml

@@ -6,6 +6,6 @@ members = [
     'moqt-server-sample',
     'moqt-client-sample',
     'moqt',
-    'sample/server',
-    'sample/client',
+    'relay',
+    'client',
 ]

sample/client/Cargo.toml → client/Cargo.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-moqt = {path = "../../moqt"}
+moqt = {path = "../moqt"}
 tokio = { version = "1.45.0", features = ["full", "tracing"] }
 tracing = "0.1.37"
 async-trait = "0.1.74"

client/src/client.rs

@@ -0,0 +1,303 @@
+use std::{
+    net::ToSocketAddrs,
+    str::FromStr,
+    sync::{
+        Arc,
+        atomic::{AtomicU64, Ordering},
+    },
+};
+
+use anyhow::bail;
+use moqt::{DatagramField, Endpoint, ObjectDatagram, QUIC, Session, SubscribeOption};
+
+use crate::stream_runner::StreamTaskRunner;
+
+pub(crate) struct Client {
+    label: String,
+    join_handle: tokio::task::JoinHandle<()>,
+    track_alias: Arc<AtomicU64>,
+    publisher: moqt::Publisher<moqt::QUIC>,
+    subscriber: moqt::Subscriber<moqt::QUIC>,
+    runner: StreamTaskRunner,
+}
+
+impl Client {
+    pub(crate) async fn new(cert_path: String, label: String) -> anyhow::Result<Self> {
+        // let endpoint = Endpoint::<QUIC>::create_client_with_custom_cert(0, &cert_path)?;
+        let config = moqt::ClientConfig::default();
+        let endpoint = Endpoint::<QUIC>::create_client(config)?;
+        // let url = url::Url::from_str("moqt://interop-relay.cloudflare.mediaoverquic.com:443")?;
+        let url = url::Url::from_str("moqt://fb.mvfst.net:9448")?;
+        // let url = url::Url::from_str("moqt://lminiero.it:9000")?;
+        // let url = url::Url::from_str("moqt://moqt.research.skyway.io:4434")?;
+        // let url = url::Url::from_str("moqt://localhost:4434")?;
+        let host = url.host_str().unwrap();
+        let remote_address = (host, url.port().unwrap_or(4433))
+            .to_socket_addrs()?
+            .next()
+            .unwrap();
+
+        tracing::info!("remote_address: {} host: {}", remote_address, host);
+
+        let session = match endpoint.connect(remote_address, host).await {
+            Ok(s) => s,
+            Err(e) => {
+                bail!("test failed: {:?}", e)
+            }
+        };
+        let track_alias = Arc::new(AtomicU64::new(0));
+        let (publisher, subscriber) = session.create_publisher_subscriber_pair();
+        let join_handle = Self::create_receiver(label.clone(), session, track_alias.clone());
+
+        Ok(Self {
+            label,
+            join_handle,
+            track_alias,
+            publisher,
+            subscriber,
+            runner: StreamTaskRunner::new(),
+        })
+    }
+
+    fn create_receiver(
+        label: String,
+        session: Session<moqt::QUIC>,
+        track_alias: Arc<AtomicU64>,
+    ) -> tokio::task::JoinHandle<()> {
+        tokio::task::Builder::new()
+            .spawn(async move {
+                let runner = StreamTaskRunner::new();
+                loop {
+                    let result = session.receive_event().await;
+                    if let Err(e) = result {
+                        tracing::error!("Failed to receive event: {}", e);
+                        break;
+                    }
+                    let event = result.unwrap();
+                    match event {
+                        moqt::SessionEvent::PublishNamespace(publish_namespace_handler) => {
+                            tracing::info!(
+                                "Received: {} Publish Namespace: {}",
+                                label,
+                                publish_namespace_handler.track_namespace
+                            );
+                            let _ = publish_namespace_handler.ok().await;
+                        }
+                        moqt::SessionEvent::SubscribeNameSpace(subscribe_namespace_handler) => {
+                            tracing::info!(
+                                "Received: {} Subscribe Namespace: {}",
+                                label,
+                                subscribe_namespace_handler.track_namespace_prefix
+                            );
+                            let _ = subscribe_namespace_handler.ok().await;
+                        }
+                        moqt::SessionEvent::Publish(publish_handler) => {
+                            tracing::info!("Received: {} Publish", label);
+                            match publish_handler
+                                .ok(128, moqt::FilterType::LatestObject)
+                                .await
+                            {
+                                Ok(h) => h,
+                                Err(_) => {
+                                    tracing::error!("failed to send");
+                                    return;
+                                }
+                            };
+                            // Self::subscribe(label.clone(), publish_handler, &runner).await;
+                        }
+                        moqt::SessionEvent::Subscribe(subscribe_handler) => {
+                            tracing::info!("Received: {} Subscribe", label);
+                            let track_alias = track_alias.load(Ordering::SeqCst);
+                            let _ = subscribe_handler
+                                .ok(track_alias, 1000000, moqt::ContentExists::False)
+                                .await;
+                            let publication = subscribe_handler.into_publication(track_alias);
+                            Self::create_stream(label.clone(), publication, &runner).await;
+                        }
+                        moqt::SessionEvent::ProtocolViolation() => {
+                            tracing::info!("Received: {} ProtocolViolation", label);
+                        }
+                    };
+                }
+            })
+            .unwrap()
+    }
+
+    pub(crate) async fn publish_namespace(&self, track_namespace: String) {
+        let result = self.publisher.publish_namespace(track_namespace).await;
+        if result.is_err() {
+            tracing::info!("{}: publish namespace error", self.label);
+        } else {
+            tracing::info!("{}: publish namespace ok", self.label);
+        }
+    }
+
+    pub(crate) async fn subscribe_namespace(&self, track_namespace_prefix: String) {
+        let result = self
+            .subscriber
+            .subscribe_namespace(track_namespace_prefix)
+            .await;
+        if result.is_err() {
+            tracing::info!("{}: subscribe namespace error", self.label);
+        } else {
+            tracing::info!("{}: subscribe namespace ok", self.label);
+        }
+    }
+
+    pub(crate) async fn publish(&self, track_namespace: String, track_name: String) {
+        let option = moqt::PublishOption::default();
+        let pub_result = self
+            .publisher
+            .publish(track_namespace, track_name, option)
+            .await;
+        if let Ok(p) = pub_result {
+            tracing::info!("{}: publish ok", self.label);
+            self.track_alias
+                .fetch_add(p.track_alias, std::sync::atomic::Ordering::SeqCst);
+        } else {
+            tracing::error!("{}: publish error", self.label);
+        }
+    }
+
+    async fn subscribe(
+        label: String,
+        publish_handler: moqt::PublishHandler<moqt::QUIC>,
+        runner: &StreamTaskRunner,
+    ) {
+        let full_name = format!(
+            "{}/{}",
+            publish_handler.track_namespace, publish_handler.track_name
+        );
+        let task = async move {
+            tracing::info!("{} :subscribe {}", label, full_name);
+            let subscription = publish_handler.into_subscription(0);
+            let acceptance = match subscription.accept_stream_or_datagram().await {
+                Ok(acceptance) => acceptance,
+                Err(_) => {
+                    tracing::error!("Failed to accept stream or datagram");
+                    return;
+                }
+            };
+            match acceptance {
+                moqt::Acceptance::Stream(stream) => todo!(),
+                moqt::Acceptance::Datagram(mut receiver, object) => {
+                    tracing::info!("{} :subscribe datagram :{:?}", label, object);
+                    loop {
+                        let result = receiver.receive().await;
+                        if let Err(e) = result {
+                            tracing::error!("Failed to receive: {}", e);
+                            break;
+                        }
+                        let object = result.unwrap();
+                        tracing::info!("{} :subscribe datagram, message: {:?}", label, object);
+                    }
+                }
+            }
+        };
+        runner.add_task(Box::pin(task)).await;
+    }
+
+    async fn create_stream(
+        label: String,
+        publication: moqt::PublishedResource<moqt::QUIC>,
+        runner: &StreamTaskRunner,
+    ) {
+        tracing::info!("{} :create stream", label);
+        let datagram = publication.create_datagram();
+        // let _ = publication
+        //     .create_stream()
+        //     .await
+        //     .inspect(|_| {
+        //         tracing::info!("{} :create stream ok", label);
+        //     })
+        //     .inspect_err(|_| {
+        //         tracing::error!("{} :create stream error", label);
+        //     });
+        let task = async move {
+            let mut id = 0;
+            tracing::info!("{} :create stream start", label);
+            loop {
+                let format_text = format!("hello from {}! id: {}", label, id);
+                let data = DatagramField::to_bytes(format_text);
+                let field = DatagramField::Payload0x00 {
+                    object_id: id,
+                    payload: data,
+                    payload: Arc::new(format_text.as_bytes().to_vec()),
+                };
+                let obj = ObjectDatagram::new(publication.track_alias, id, field);
+                match datagram.send(obj).await {
+                    Ok(_) => {
+                        tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
+                        id += 1
+                    }
+                    Err(_) => {
+                        tracing::error!("failed to send");
+                        break;
+                    }
+                }
+            }
+        };
+        runner.add_task(Box::pin(task)).await;
+    }
+
+    pub(crate) async fn active_subscribe(
+        &self,
+        label: String,
+        track_namespace: String,
+        track_name: String,
+    ) {
+        let full_name = format!("{}/{}", track_namespace, track_name);
+        let option = SubscribeOption {
+            subscriber_priority: 128,
+            group_order: moqt::GroupOrder::Ascending,
+            forward: true,
+            filter_type: moqt::FilterType::LatestObject,
+        };
+        tracing::info!("{} :subscribe {}", label, full_name);
+        let subscription = match self
+            .subscriber
+            .subscribe(track_namespace, track_name, option)
+            .await
+        {
+            Ok(s) => s,
+            Err(e) => {
+                tracing::error!("Failed to subscribe: {}", e);
+                return;
+            }
+        };
+        tracing::warn!("qqq subscribe ok");
+        let task = async move {
+            let acceptance = match subscription.accept_stream_or_datagram().await {
+                Ok(acceptance) => acceptance,
+                Err(e) => {
+                    tracing::error!("Failed to accept stream or datagram: {}", e);
+                    return;
+                }
+            };
+            tracing::warn!("qqq accept ok");
+            match acceptance {
+                moqt::Acceptance::Stream(stream) => todo!(),
+                moqt::Acceptance::Datagram(mut receiver, object) => {
+                    tracing::info!("{} :subscribe datagram: {:?}", label, object,);
+                    loop {
+                        let result = receiver.receive().await;
+                        if let Err(e) = result {
+                            tracing::error!("Failed to receive: {}", e);
+                            break;
+                        }
+                        let object = result.unwrap();
+                        tracing::info!("{} :subscribe datagram: {:?}", label, object);
+                    }
+                }
+            };
+        };
+        self.runner.add_task(Box::pin(task)).await;
+    }
+}
+
+impl Drop for Client {
+    fn drop(&mut self) {
+        tracing::info!("Client has been dropped.");
+        self.join_handle.abort();
+    }
+}