Wip: use 'push' for send_bytes and send_text

Author: ladvoc

examples/send_bytes/src/main.rs

@@ -55,18 +55,15 @@ async fn run_receiver(
     println!("Waiting for LED control packets…");
     while let Some(event) = rx.recv().await {
         match event {
-            RoomEvent::ByteStreamOpened { reader, topic, participant_identity: _ } => {
-                if topic != LED_CONTROL_TOPIC {
+            RoomEvent::BytesReceived { bytes, info, participant_identity: _ } => {
+                if info.topic != LED_CONTROL_TOPIC {
                     continue;
                 };
-                let Some(reader) = reader.take() else { continue };
-
-                let Ok(be_bytes) = reader.read_all().await?[..4].try_into() else {
+                let Ok(be_bytes) = bytes[..4].try_into() else {
                     log::warn!("Unexpected packet length");
                     continue;
                 };
                 let packet = LedControlPacket::from(u32::from_be_bytes(be_bytes));
-
                 println!("[rx] {}", packet);
             }
             _ => {}

livekit/src/room/data_stream/outgoing.rs

@@ -348,7 +348,7 @@ impl OutgoingStreamManager {
     pub async fn send_text(
         &self,
         text: &str,
-        options: StreamTextOptions,
+        mut options: StreamTextOptions,
     ) -> StreamResult<TextStreamInfo> {
         let text_header = proto::data_stream::TextHeader {
             operation_type: options.operation_type.unwrap_or_default() as i32,
@@ -357,6 +357,8 @@ impl OutgoingStreamManager {
             attached_stream_ids: options.attached_stream_ids,
             generated: options.generated.unwrap_or_default(),
         };
+
+        options.attributes.insert("__push".to_owned(), "1".into());
         let header = proto::data_stream::Header {
             stream_id: options.id.unwrap_or_else(|| create_random_uuid()),
             timestamp: Utc::now().timestamp_millis(),
@@ -397,13 +399,13 @@ impl OutgoingStreamManager {
     pub async fn send_bytes(
         &self,
         data: impl AsRef<[u8]>,
-        options: StreamByteOptions,
+        mut options: StreamByteOptions,
     ) -> StreamResult<ByteStreamInfo> {
         let bytes = data.as_ref();
 
-        let byte_header = proto::data_stream::ByteHeader {
-            name: options.name.unwrap_or_default(),
-        };
+        let byte_header = proto::data_stream::ByteHeader { name: options.name.unwrap_or_default() };
+
+        options.attributes.insert("__push".to_owned(), "1".into());
         let header = proto::data_stream::Header {
             stream_id: options.id.unwrap_or_else(|| create_random_uuid()),
             timestamp: Utc::now().timestamp_millis(),

livekit/src/room/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use bmrng::unbounded::UnboundedRequestReceiver;
+use bytes::Bytes;
 use libwebrtc::{
     native::frame_cryptor::EncryptionState,
     prelude::{
@@ -186,6 +187,20 @@ pub enum RoomEvent {
         topic: String,
         participant_identity: ParticipantIdentity,
     },
+    /// A byte buffer has been fully received from a remove participant
+    /// sent using [`LocalParticipant::send_bytes`].
+    BytesReceived {
+        bytes: Bytes,
+        info: ByteStreamInfo,
+        participant_identity: ParticipantIdentity,
+    },
+    /// A string has been fully received from a remove participant
+    /// sent using [`LocalParticipant::send_text`].
+    TextReceived {
+        text: String,
+        info: TextStreamInfo,
+        participant_identity: ParticipantIdentity,
+    },
     #[deprecated(note = "Use high-level data streams API instead.")]
     StreamHeaderReceived {
         header: proto::data_stream::Header,
@@ -1701,17 +1716,8 @@ async fn incoming_data_stream_task(
     loop {
         tokio::select! {
             Some((reader, identity)) = open_rx.recv() => {
-                match reader {
-                    AnyStreamReader::Byte(reader) => dispatcher.dispatch(&RoomEvent::ByteStreamOpened {
-                        topic: reader.info().topic.clone(),
-                        reader: TakeCell::new(reader),
-                        participant_identity: ParticipantIdentity(identity)
-                    }),
-                    AnyStreamReader::Text(reader) => dispatcher.dispatch(&RoomEvent::TextStreamOpened {
-                        topic: reader.info().topic.clone(),
-                        reader: TakeCell::new(reader),
-                        participant_identity: ParticipantIdentity(identity)
-                    }),
+                if let Some(event) = event_for_incoming_stream(reader, identity).await {
+                    dispatcher.dispatch(&event)
                 }
             },
             _ = close_rx.recv() => {
@@ -1721,6 +1727,44 @@ async fn incoming_data_stream_task(
     }
 }
 
+async fn event_for_incoming_stream(reader: AnyStreamReader, identity: String) -> Option<RoomEvent> {
+    let participant_identity = ParticipantIdentity(identity);
+    match reader {
+        AnyStreamReader::Byte(reader) => {
+            if reader.info().attributes.contains_key("__push") {
+                // TODO: avoid cloning info
+                let info = reader.info().clone();
+                let Ok(bytes) = reader.read_all().await else {
+                    log::error!("Failed to read pushed bytes for stream '{}'", info.id);
+                    return None;
+                };
+                return Some(RoomEvent::BytesReceived { bytes, info, participant_identity });
+            }
+            Some(RoomEvent::ByteStreamOpened {
+                topic: reader.info().topic.clone(),
+                reader: TakeCell::new(reader),
+                participant_identity,
+            })
+        }
+        AnyStreamReader::Text(reader) => {
+            if reader.info().attributes.contains_key("__push") {
+                // TODO: avoid cloning info
+                let info = reader.info().clone();
+                let Ok(text) = reader.read_all().await else {
+                    log::error!("Failed to read pushed text for stream '{}'", info.id);
+                    return None;
+                };
+                return Some(RoomEvent::TextReceived { text, info, participant_identity });
+            }
+            Some(RoomEvent::TextStreamOpened {
+                topic: reader.info().topic.clone(),
+                reader: TakeCell::new(reader),
+                participant_identity,
+            })
+        }
+    }
+}
+
 /// Receives packets from the outgoing stream manager and send them.
 async fn outgoing_data_stream_task(
     mut packet_rx: UnboundedRequestReceiver<proto::DataPacket, Result<(), EngineError>>,

livekit/tests/data_stream_test.rs

@@ -40,16 +40,12 @@ async fn test_send_bytes() -> Result<()> {
     };
     let receive_text = async move {
         while let Some(event) = receiving_event_rx.recv().await {
-            let RoomEvent::ByteStreamOpened { reader, topic, participant_identity } = event else {
+            let RoomEvent::BytesReceived { bytes, info, participant_identity } = event else {
                 continue;
             };
-            assert_eq!(topic, "some-topic");
+            assert_eq!(info.topic, "some-topic");
             assert_eq!(participant_identity, sender_identity);
-
-            let Some(reader) = reader.take() else {
-                return Err(anyhow!("Failed to take reader"));
-            };
-            assert_eq!(reader.read_all().await?, BYTES_TO_SEND);
+            assert_eq!(bytes, BYTES_TO_SEND);
             break;
         }
         Ok(())
@@ -85,16 +81,12 @@ async fn test_send_text() -> Result<()> {
     };
     let receive_text = async move {
         while let Some(event) = receiving_event_rx.recv().await {
-            let RoomEvent::TextStreamOpened { reader, topic, participant_identity } = event else {
+            let RoomEvent::TextReceived { text, info, participant_identity } = event else {
                 continue;
             };
-            assert_eq!(topic, "some-topic");
+            assert_eq!(info.topic, "some-topic");
             assert_eq!(participant_identity, sender_identity);
-
-            let Some(reader) = reader.take() else {
-                return Err(anyhow!("Failed to take reader"));
-            };
-            assert_eq!(reader.read_all().await?, TEXT_TO_SEND);
+            assert_eq!(text, TEXT_TO_SEND);
             break;
         }
         Ok(())