From 0b00fa9a105a5e0e210bd79864e605f71512336f Mon Sep 17 00:00:00 2001
From: aoife cassidy <aoife@livekit.io>
Date: Wed, 4 Dec 2024 17:18:46 +0200
Subject: [PATCH] feat(livekit): add fast_publish

---
 .nanpa/fast-publish.kdl               | 1 +
 livekit/src/rtc_engine/rtc_session.rs | 7 ++++++-
 2 files changed, 7 insertions(+), 1 deletion(-)
 create mode 100644 .nanpa/fast-publish.kdl

diff --git a/.nanpa/fast-publish.kdl b/.nanpa/fast-publish.kdl
new file mode 100644
index 00000000..d2c6d206
--- /dev/null
+++ b/.nanpa/fast-publish.kdl
@@ -0,0 +1 @@
+patch package="livekit" type="added" "implement fast publish"
diff --git a/livekit/src/rtc_engine/rtc_session.rs b/livekit/src/rtc_engine/rtc_session.rs
index fa01a69f..c73b2cdb 100644
--- a/livekit/src/rtc_engine/rtc_session.rs
+++ b/livekit/src/rtc_engine/rtc_session.rs
@@ -172,6 +172,8 @@ struct SessionInner {
     negotiation_debouncer: Mutex<Option<Debouncer>>,
 
     pending_requests: Mutex<HashMap<u32, oneshot::Sender<proto::RequestResponse>>>,
+
+    fast_publish: AtomicBool,
 }
 
 /// This struct holds a WebRTC session
@@ -258,6 +260,7 @@ impl RtcSession {
             options,
             negotiation_debouncer: Default::default(),
             pending_requests: Default::default(),
+            fast_publish: join_response.fast_publish.into(),
         });
 
         // Start session tasks
@@ -938,7 +941,9 @@ impl SessionInner {
         data: &proto::DataPacket,
         kind: DataPacketKind,
     ) -> Result<(), EngineError> {
-        self.ensure_publisher_connected(kind).await?;
+        if !self.fast_publish.load(Ordering::Acquire) {
+            self.ensure_publisher_connected(kind).await?;
+        }
         self.data_channel(SignalTarget::Publisher, kind)
             .unwrap()
             .send(&data.encode_to_vec(), true)