room_session wip

crates/livekit-core/src/room/internal.rs

@@ -0,0 +1,317 @@
+use parking_lot::{Mutex, RwLock};
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::Arc;
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+
+use crate::events::{ParticipantConnectedEvent, ParticipantDisconnectedEvent, RoomEvents};
+use crate::proto::{self, participant_info};
+use crate::room::ConnectionState;
+use crate::rtc_engine::{EngineEvent, EngineEvents, RTCEngine};
+use crate::signal_client::SignalOptions;
+
+use super::id::{ParticipantIdentity, ParticipantSid};
+use super::participant::local_participant::LocalParticipant;
+use super::participant::remote_participant::RemoteParticipant;
+use super::participant::{ParticipantInternalTrait, ParticipantTrait};
+use super::{RoomError, RoomHandle, RoomResult};
+use tracing::{error, instrument, Level};
+
+#[derive(Debug)]
+pub struct SessionInner {
+    pub state: AtomicU8, // ConnectionState
+    pub sid: Mutex<String>,
+    pub name: Mutex<String>,
+    pub participants: RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>>,
+    pub rtc_engine: Arc<RTCEngine>,
+    pub local_participant: Arc<LocalParticipant>,
+    pub room_events: Arc<RoomEvents>,
+}
+
+#[derive(Debug)]
+pub struct RoomSession {
+    inner: Arc<SessionInner>,
+    session_task: JoinHandle<()>,
+    close_emitter: oneshot::Sender<()>,
+}
+
+impl RoomSession {
+    pub async fn connect(
+        room_events: Arc<RoomEvents>,
+        url: &str,
+        token: &str,
+    ) -> RoomResult<Self> {
+        let (rtc_engine, engine_events) = RTCEngine::new();
+        let rtc_engine = Arc::new(rtc_engine);
+        rtc_engine
+            .connect(url, token, SignalOptions::default())
+            .await?;
+
+        let join_response = rtc_engine.join_response().unwrap();
+        let pi = join_response.participant.unwrap().clone();
+        let local_participant = Arc::new(LocalParticipant::new(
+            rtc_engine.clone(),
+            pi.sid.into(),
+            pi.identity.into(),
+            pi.name,
+            pi.metadata,
+        ));
+        let room_info = join_response.room.unwrap();
+        let inner = Arc::new(SessionInner {
+            state: AtomicU8::new(ConnectionState::Connecting as u8),
+            sid: Mutex::new(room_info.sid),
+            name: Mutex::new(room_info.name),
+            participants: Default::default(),
+            rtc_engine,
+            local_participant,
+            room_events,
+        });
+
+        for pi in join_response.other_participants {
+            let participant = {
+                let pi = pi.clone();
+                inner.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
+            };
+            participant.update_info(pi.clone());
+            participant
+                .update_tracks(RoomHandle::from(inner.clone()), pi.tracks)
+                .await;
+        }
+
+        let (close_emitter, close_receiver) = oneshot::channel();
+        let session_task = tokio::spawn(inner.room_task(engine_events, close_receiver));
+
+        let session = Self {
+            inner,
+            session_task,
+            close_emitter,
+        };
+        Ok(session)
+    }
+
+    pub async fn close(self) {
+        self.inner.close();
+        let _ = self.close_emitter.send(());
+        self.session_task.await;
+    }
+}
+
+impl SessionInner {
+    async fn room_task(
+        self: Arc<Self>,
+        mut engine_events: EngineEvents,
+        mut close_receiver: oneshot::Receiver<()>,
+    ) {
+        loop {
+            tokio::select! {
+               res = engine_events.recv() => {
+                    if let Some(event) = res {
+                        if let Err(err) = self.on_engine_event(event).await {
+                            error!("failed to handle engine event: {:?}", err);
+                        }
+                    } else {
+                        panic!("engine_events has been closed unexpectedly");
+                    }
+                },
+                 _ = &mut close_receiver => {
+                    break;
+                }
+            }
+        }
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    async fn on_engine_event(self: &Arc<Self>, event: EngineEvent) -> RoomResult<()> {
+        match event {
+            EngineEvent::ParticipantUpdate(update) => self.handle_participant_update(update).await,
+            EngineEvent::MediaTrack {
+                track,
+                stream,
+                receiver,
+            } => {
+                let stream_id = stream.id();
+                let lk_stream_id = unpack_stream_id(&stream_id);
+                if lk_stream_id.is_none() {
+                    Err(RoomError::Internal(format!(
+                        "MediaTrack event with invalid track_id: {:?}",
+                        &stream_id
+                    )))?;
+                }
+
+                let (participant_sid, track_sid) = lk_stream_id.unwrap();
+                let remote_participant = self.get_participant(&participant_sid.to_string().into());
+
+                if let Some(remote_participant) = remote_participant {
+                    tokio::spawn({
+                        let room_internal = self.clone();
+                        {
+                            let track_sid = track_sid.to_owned().into();
+                            async move {
+                                remote_participant
+                                    .add_subscribed_media_track(
+                                        RoomHandle::from(room_internal),
+                                        track_sid,
+                                        track,
+                                    )
+                                    .await;
+                            }
+                        }
+                    });
+                } else {
+                    // The server should send participant updates before sending a new offer
+                    // So this should never happen.
+                    Err(RoomError::Internal(format!(
+                        "AddTrack event with invalid participant_sid: {:?}",
+                        participant_sid
+                    )))?;
+                }
+            }
+            EngineEvent::Resuming => {}
+            EngineEvent::Resumed => {}
+            EngineEvent::Restarting => {}
+            EngineEvent::Restarted => {}
+            EngineEvent::Disconnected => {}
+        }
+
+        Ok(())
+    }
+
+    async fn close(&self) {
+        self.rtc_engine.close().await;
+    }
+
+    fn get_participant(self: &Arc<Self>, sid: &ParticipantSid) -> Option<Arc<RemoteParticipant>> {
+        self.participants.read().get(sid).cloned()
+    }
+
+    /// Update the participants inside a Room.
+    /// It'll create, update or remove a participant
+    /// It also update the participant tracks.
+    #[instrument(level = Level::DEBUG)]
+    async fn handle_participant_update(self: &Arc<Self>, update: proto::ParticipantUpdate) {
+        for pi in update.participants {
+            if pi.sid == self.local_participant.sid()
+                || pi.identity == self.local_participant.identity()
+            {
+                self.local_participant.clone().update_info(pi);
+                continue;
+            }
+
+            let remote_participant = self.get_participant(&pi.sid.clone().into());
+
+            if let Some(remote_participant) = remote_participant {
+                if pi.state == participant_info::State::Disconnected as i32 {
+                    // Participant disconnected
+                    self.handle_participant_disconnect(remote_participant)
+                } else {
+                    // Participant is already connected, update the it
+                    remote_participant.update_info(pi.clone());
+                    remote_participant
+                        .update_tracks(RoomHandle::from(self.clone()), pi.tracks)
+                        .await;
+                }
+            } else {
+                // Create a new participant
+                let remote_participant = {
+                    let pi = pi.clone();
+                    self.create_participant(pi.sid.into(), pi.identity.into(), pi.name, pi.metadata)
+                };
+                let mut handler = self.room_events.on_participant_connected.lock();
+                if let Some(cb) = handler.as_mut() {
+                    cb(ParticipantConnectedEvent {
+                        room_handle: RoomHandle::from(self.clone()),
+                        participant: remote_participant.clone(),
+                    });
+                }
+
+                remote_participant.update_info(pi.clone());
+                remote_participant
+                    .update_tracks(RoomHandle::from(self.clone()), pi.tracks)
+                    .await;
+            }
+        }
+    }
+
+    #[instrument(level = Level::DEBUG)]
+    fn handle_participant_disconnect(self: &Arc<Self>, remote_participant: Arc<RemoteParticipant>) {
+        self.participants.write().remove(&remote_participant.sid());
+
+        // TODO(theomonnom): Unpublish all tracks
+
+        let mut handler = self.room_events.on_participant_disconnected.lock();
+        if let Some(cb) = handler.as_mut() {
+            cb(ParticipantDisconnectedEvent {
+                room_handle: RoomHandle::from(self.clone()),
+                participant: remote_participant.clone(),
+            });
+        }
+    }
+
+    fn create_participant(
+        self: &Arc<Self>,
+        sid: ParticipantSid,
+        identity: ParticipantIdentity,
+        name: String,
+        metadata: String,
+    ) -> Arc<RemoteParticipant> {
+        let p = Arc::new(RemoteParticipant::new(
+            sid.clone(),
+            identity,
+            name,
+            metadata,
+        ));
+
+        macro_rules! forward_event {
+            ($type:ident, when_connected) => {
+                p.internal_events().$type({
+                    let room_internal = self.clone();
+                    move |event| {
+                        let room_internal = room_internal.clone();
+                        async move {
+                            if room_internal.state.load(Ordering::SeqCst)
+                                == ConnectionState::Connected as u8
+                            {
+                                if let Some(cb) = room_internal.room_events.$type.lock().as_mut() {
+                                    cb(event).await;
+                                }
+                            }
+                        }
+                    }
+                })
+            };
+            ($type:ident) => {
+                p.internal_events().$type({
+                    let room_internal = self.clone();
+                    move |event| {
+                        let room_internal = room_internal.clone();
+                        async move {
+                            if let Some(cb) = room_internal.room_events.$type.lock().as_mut() {
+                                cb(event).await;
+                            }
+                        }
+                    }
+                })
+            };
+        }
+
+        // Forward participantevents to room events
+        forward_event!(on_track_published, when_connected);
+        forward_event!(on_track_subscribed);
+        forward_event!(on_track_subscription_failed);
+
+        self.participants.write().insert(sid, p.clone());
+        p
+    }
+}
+
+fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> {
+    let split: Vec<&str> = stream_id.split('|').collect();
+    if split.len() == 2 {
+        let participant_sid = split.get(0).unwrap();
+        let track_sid = split.get(1).unwrap();
+        Some((participant_sid, track_sid))
+    } else {
+        None
+    }
+}

crates/livekit-core/src/room/mod.rs

@@ -4,6 +4,7 @@ use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 
 use self::id::{ParticipantIdentity, ParticipantSid};
+use self::internal::{RoomInternal, RoomSession};
 use self::participant::local_participant::LocalParticipant;
 use self::participant::remote_participant::RemoteParticipant;
 use self::participant::ParticipantInternalTrait;
@@ -17,14 +18,13 @@ use crate::proto::participant_info;
 use thiserror::Error;
 use tracing::{debug, error, instrument, trace_span, Level};
 
-use crate::rtc_engine::{
-    EngineError, EngineEvent, EngineEvents, EngineResult, RTCEngine,
-};
+use crate::rtc_engine::{EngineError, EngineEvent, EngineEvents, EngineResult, RTCEngine};
 use crate::signal_client::SignalOptions;
 
 pub use crate::rtc_engine::SimulateScenario;
 
 pub mod id;
+mod internal;
 pub mod participant;
 pub mod publication;
 pub mod track;
@@ -47,351 +47,52 @@ pub enum ConnectionState {
     Reconnecting,
 }
 
-#[derive(Debug)]
-struct RoomInner {
-    state: AtomicU8, // ConnectionState
-    sid: Mutex<String>,
-    name: Mutex<String>,
-    participants: RwLock<HashMap<ParticipantSid, Arc<RemoteParticipant>>>,
-    rtc_engine: Arc<RTCEngine>,
-    local_participant: Arc<LocalParticipant>,
+#[derive(Clone, Debug)]
+pub struct RoomHandle {
+    session: Arc<RoomSession>,
 }
 
-#[derive(Debug)]
+impl RoomHandle {
+    pub fn sid(&self) -> String {
+        self.session.sid.lock().clone()
+    }
+
+    pub fn name(&self) -> String {
+        self.internal.name.lock().clone()
+    }
+
+    pub fn local_participant(&self) -> Arc<LocalParticipant> {
+        self.internal.local_participant.clone()
+    }
+
+    pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
+        self.internal.rtc_engine.simulate_scenario(scenario).await
+    }
+}
+
+#[derive(Debug, Default)]
 pub struct Room {
-    inner: Option<Arc<RoomInner>>,
-    events: Arc<RoomEvents>,
+    session: Option<RoomHandle>,
+    events: Arc<RoomEvents>, // Keep the same RoomEvents across sessions
 }
 
 impl Room {
-    pub fn new() -> Room {
-        Self {
-            inner: None,
-            events: Default::default(),
-        }
-    }
-
     #[instrument(level = Level::DEBUG)]
-    pub async fn connect(&mut self, url: &str, token: &str) -> RoomResult<()> {
-        // Initialize the RTCEngine
-        let (rtc_engine, engine_events) = RTCEngine::new();
-        let rtc_engine = Arc::new(rtc_engine);
-        rtc_engine
-            .connect(url, token, SignalOptions::default())
-            .await?;
-
-        let join_response = rtc_engine.join_response().unwrap();
-        let pi = join_response.participant.unwrap().clone();
-        let local_participant = Arc::new(LocalParticipant::new(
-            rtc_engine.clone(),
-            pi.sid.into(),
-            pi.identity.into(),
-            pi.name,
-            pi.metadata,
-        ));
-        let room_info = join_response.room.unwrap();
-        let inner = Arc::new(RoomInner {
-            state: AtomicU8::new(ConnectionState::Connecting as u8),
-            sid: Mutex::new(room_info.sid),
-            name: Mutex::new(room_info.name),
-            participants: Default::default(),
-            rtc_engine,
-            local_participant,
-        });
-
-        for pi in join_response.other_participants {
-            let participant = {
-                let pi = pi.clone();
-                Self::create_participant(
-                    inner.clone(),
-                    self.events.clone(),
-                    pi.sid.into(),
-                    pi.identity.into(),
-                    pi.name,
-                    pi.metadata,
-                )
-            };
-            participant.update_info(pi.clone());
-            participant
-                .update_tracks(RoomHandle::from(inner.clone()), pi.tracks)
-                .await;
-        }
-
-        self.inner = Some(inner.clone());
-        tokio::spawn(Self::room_task(inner, self.events.clone(), engine_events));
-
+    pub async fn connect(&self, url: &str, token: &str) -> RoomResult<()> {
+        let room_session = Arc::new(RoomSession::connect(self.events.clone(), url, token).await?);
+        self.session = Some(room_session.clone());
         Ok(())
     }
 
+    pub async fn close(&self) {}
+
     pub fn events(&self) -> Arc<RoomEvents> {
         self.events.clone()
     }
 
     pub fn get_handle(&self) -> Option<RoomHandle> {
-        self.inner.as_ref().map(|inner| RoomHandle {
-            inner: inner.clone(),
+        self.internal.as_ref().map(|internal| RoomHandle {
+            internal: internal.clone(),
         })
     }
-
-    async fn room_task(
-        room_inner: Arc<RoomInner>,
-        room_events: Arc<RoomEvents>,
-        mut engine_events: EngineEvents,
-    ) {
-        while let Some(event) = engine_events.recv().await {
-            if let Err(err) =
-                Self::handle_event(room_inner.clone(), room_events.clone(), event).await
-            {
-                error!("failed to handle engine event: {:?}", err);
-            }
-        }
-    }
-
-    #[instrument(level = Level::DEBUG, skip(room_inner, room_events))]
-    async fn handle_event(
-        room_inner: Arc<RoomInner>,
-        room_events: Arc<RoomEvents>,
-        event: EngineEvent,
-    ) -> RoomResult<()> {
-        match event {
-            EngineEvent::ParticipantUpdate(update) => {
-                Self::handle_participant_update(room_inner.clone(), room_events.clone(), update)
-                    .await
-            }
-            EngineEvent::MediaTrack {
-                track,
-                stream,
-                receiver,
-            } => {
-                let stream_id = stream.id();
-                let lk_stream_id = unpack_stream_id(&stream_id);
-                if lk_stream_id.is_none() {
-                    Err(RoomError::Internal(format!(
-                        "MediaTrack event with invalid track_id: {:?}",
-                        &stream_id
-                    )))?;
-                }
-
-                let (participant_sid, track_sid) = lk_stream_id.unwrap();
-                let remote_participant =
-                    Self::get_participant(room_inner.clone(), &participant_sid.to_string().into());
-
-                if let Some(remote_participant) = remote_participant {
-                    tokio::spawn({
-                        let track_sid = track_sid.to_owned().into();
-                        async move {
-                            remote_participant
-                                .add_subscribed_media_track(
-                                    RoomHandle::from(room_inner),
-                                    track_sid,
-                                    track,
-                                )
-                                .await;
-                        }
-                    });
-                } else {
-                    // The server should send participant updates before sending a new offer
-                    // So this should not happen.
-                    Err(RoomError::Internal(format!(
-                        "AddTrack event with invalid participant_sid: {:?}",
-                        participant_sid
-                    )))?;
-                }
-            }
-            EngineEvent::Resuming => {}
-            EngineEvent::Resumed => {}
-            EngineEvent::Restarting => {}
-            EngineEvent::Restarted => {}
-            EngineEvent::Disconnected => {}
-        }
-
-        Ok(())
-    }
-
-    #[instrument(level = Level::DEBUG, skip(room_inner, room_events))]
-    async fn handle_participant_update(
-        room_inner: Arc<RoomInner>,
-        room_events: Arc<RoomEvents>,
-        update: proto::ParticipantUpdate,
-    ) {
-        for pi in update.participants {
-            if pi.sid == room_inner.local_participant.sid()
-                || pi.identity == room_inner.local_participant.identity()
-            {
-                room_inner.local_participant.clone().update_info(pi);
-                continue;
-            }
-
-            let remote_participant =
-                Self::get_participant(room_inner.clone(), &pi.sid.clone().into());
-
-            if let Some(remote_participant) = remote_participant {
-                if pi.state == participant_info::State::Disconnected as i32 {
-                    // Participant disconencted
-                    Self::handle_participant_disconnect(
-                        room_inner.clone(),
-                        room_events.clone(),
-                        remote_participant,
-                    )
-                } else {
-                    // Participant is already connected, update the informations
-                    remote_participant.update_info(pi.clone());
-                    remote_participant
-                        .update_tracks(RoomHandle::from(room_inner.clone()), pi.tracks)
-                        .await;
-                }
-            } else {
-                // Create a new participant and call OnConnect event
-                let remote_participant = {
-                    let pi = pi.clone();
-                    Self::create_participant(
-                        room_inner.clone(),
-                        room_events.clone(),
-                        pi.sid.into(),
-                        pi.identity.into(),
-                        pi.name,
-                        pi.metadata,
-                    )
-                };
-                let mut handler = room_events.on_participant_connected.lock();
-                if let Some(cb) = handler.as_mut() {
-                    cb(ParticipantConnectedEvent {
-                        room_handle: RoomHandle::from(room_inner.clone()),
-                        participant: remote_participant.clone(),
-                    });
-                }
-
-                remote_participant.update_info(pi.clone());
-                remote_participant
-                    .update_tracks(RoomHandle::from(room_inner.clone()), pi.tracks)
-                    .await;
-            }
-        }
-    }
-
-    #[instrument(level = Level::DEBUG, skip(room_inner, room_events))]
-    fn handle_participant_disconnect(
-        room_inner: Arc<RoomInner>,
-        room_events: Arc<RoomEvents>,
-        remote_participant: Arc<RemoteParticipant>,
-    ) {
-        room_inner
-            .participants
-            .write()
-            .remove(&remote_participant.sid());
-
-        // TODO(theomonnom): Unpublish all tracks
-
-        let mut handler = room_events.on_participant_disconnected.lock();
-        if let Some(cb) = handler.as_mut() {
-            cb(ParticipantDisconnectedEvent {
-                room_handle: RoomHandle::from(room_inner.clone()),
-                participant: remote_participant.clone(),
-            });
-        }
-    }
-
-    fn get_participant(
-        room_inner: Arc<RoomInner>,
-        sid: &ParticipantSid,
-    ) -> Option<Arc<RemoteParticipant>> {
-        room_inner.participants.read().get(sid).cloned()
-    }
-
-    fn create_participant(
-        room_inner: Arc<RoomInner>,
-        room_events: Arc<RoomEvents>,
-        sid: ParticipantSid,
-        identity: ParticipantIdentity,
-        name: String,
-        metadata: String,
-    ) -> Arc<RemoteParticipant> {
-        let p = Arc::new(RemoteParticipant::new(
-            sid.clone(),
-            identity,
-            name,
-            metadata,
-        ));
-
-        macro_rules! forward_event {
-            ($type:ident, when_connected) => {
-                p.internal_events().$type({
-                    let room_events = room_events.clone();
-                    let room_inner = room_inner.clone();
-                    move |event| {
-                        let room_events = room_events.clone();
-                        let room_inner = room_inner.clone();
-                        async move {
-                            if room_inner.state.load(Ordering::SeqCst)
-                                == ConnectionState::Connected as u8
-                            {
-                                if let Some(cb) = room_events.$type.lock().as_mut() {
-                                    cb(event).await;
-                                }
-                            }
-                        }
-                    }
-                })
-            };
-            ($type:ident) => {
-                p.internal_events().$type({
-                    let room_events = room_events.clone();
-                    move |event| {
-                        let room_events = room_events.clone();
-                        async move {
-                            if let Some(cb) = room_events.$type.lock().as_mut() {
-                                cb(event).await;
-                            }
-                        }
-                    }
-                })
-            };
-        }
-
-        // Forward participantevents to room events
-        forward_event!(on_track_published, when_connected);
-        forward_event!(on_track_subscribed);
-        forward_event!(on_track_subscription_failed);
-
-        room_inner.participants.write().insert(sid, p.clone());
-        p
-    }
-}
-
-#[derive(Clone, Debug)]
-pub struct RoomHandle {
-    inner: Arc<RoomInner>,
-}
-
-impl RoomHandle {
-    fn from(room_inner: Arc<RoomInner>) -> Self {
-        Self { inner: room_inner }
-    }
-
-    pub fn sid(&self) -> String {
-        self.inner.sid.lock().clone()
-    }
-
-    pub fn name(&self) -> String {
-        self.inner.name.lock().clone()
-    }
-
-    pub fn local_participant(&self) -> Arc<LocalParticipant> {
-        self.inner.local_participant.clone()
-    }
-
-    pub async fn simulate_scenario(&self, scenario: SimulateScenario) -> EngineResult<()> {
-        self.inner.rtc_engine.simulate_scenario(scenario).await
-    }
-}
-
-fn unpack_stream_id(stream_id: &str) -> Option<(&str, &str)> {
-    let split: Vec<&str> = stream_id.split('|').collect();
-    if split.len() == 2 {
-        let participant_sid = split.get(0).unwrap();
-        let track_sid = split.get(1).unwrap();
-        Some((participant_sid, track_sid))
-    } else {
-        None
-    }
 }

crates/livekit-core/src/room/participant/remote_participant.rs

@@ -68,7 +68,7 @@ impl RemoteParticipant {
                         return publication;
                     }
 
-                    sleep(Duration::from_millis(50)).await;
+                    tokio::task::yield_now();
                 }
             }
         };

crates/livekit-core/src/rtc_engine/mod.rs

@@ -352,7 +352,9 @@ impl EngineInner {
                 let res = inner.reconnect_task().await;
                 inner.reconnecting.store(false, Ordering::SeqCst);
 
-                if res.is_err() {
+                if res.is_ok() {
+                    warn!("RTCEngine successfully reconnected")
+                } else {
                     error!("failed to reconnect after {} attemps", RECONNECT_ATTEMPTS);
                     inner.close().await;
                 }

crates/livekit-core/src/rtc_engine/rtc_session.rs

@@ -200,9 +200,7 @@ impl RTCSession {
             join_response,
         };
 
-        // Start tasks
         let (close_emitter, close_receiver) = watch::channel(false);
-
         let inner = Arc::new(SessionInner {
             info: session_info,
             pc_state: AtomicU8::new(PCState::New as u8),
@@ -224,6 +222,10 @@ impl RTCSession {
         );
         let rtc_task = tokio::spawn(inner.clone().rtc_task(rtc_events, close_receiver.clone()));
 
+        if !inner.info.join_response.subscriber_primary {
+            inner.negotiate_publisher().await?;
+        }
+
         let session = Self {
             lk_runtime,
             inner: inner.clone(),
@@ -232,10 +234,6 @@ impl RTCSession {
             rtc_task,
         };
 
-        if !inner.info.join_response.subscriber_primary {
-            inner.negotiate_publisher().await?;
-        }
-
         Ok(session)
     }