[WIP]Make call reusable after leave - 3

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/CallStats.kt

@@ -19,12 +19,12 @@ package io.getstream.video.android.core
 import android.os.Build
 import io.getstream.log.taggedLogger
 import io.getstream.video.android.core.call.stats.model.RtcStatsReport
+import io.getstream.video.android.core.coroutines.flows.RestartableStateFlow
+import io.getstream.video.android.core.coroutines.scopes.RestartableProducerScope
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.stateIn
 import org.webrtc.CameraEnumerationAndroid
 import org.webrtc.RTCStats
 import stream.video.sfu.models.TrackType
@@ -81,18 +81,37 @@ public data class LocalStats(
     val deviceModel: String,
 )
 
-public class CallStats(val call: Call, val callScope: CoroutineScope) {
+// TODO Rahul, need to pass RestartableScope
+public class CallStats internal constructor(
+    val call: Call,
+    private val restartableProducerScope: RestartableProducerScope,
+) {
+
+    @Deprecated(
+        "Kept for binary compatibility.",
+        level = DeprecationLevel.ERROR,
+    )
+    public constructor(
+        call: Call,
+        callScope: CoroutineScope,
+    ) : this(call, RestartableProducerScope())
     private val logger by taggedLogger("CallStats")
 
     private val supervisorJob = SupervisorJob()
-    private val scope = CoroutineScope(callScope.coroutineContext + supervisorJob)
+
+    private val scope = CoroutineScope(restartableProducerScope.coroutineContext + supervisorJob)
     // TODO: cleanup the scope
 
     val publisher = PeerConnectionStats(scope)
     val subscriber = PeerConnectionStats(scope)
     val _local = MutableStateFlow<LocalStats?>(null)
-    val local: StateFlow<LocalStats?> =
-        _local.stateIn(scope, SharingStarted.WhileSubscribed(), null)
+    val local: StateFlow<LocalStats?> = RestartableStateFlow(_local, restartableProducerScope, null)
+
+    @Deprecated(
+        "Kept for binary compatibility.",
+        level = DeprecationLevel.ERROR,
+    )
+    fun getCallScope(): CoroutineScope = restartableProducerScope
 
     fun updateFromRTCStats(stats: RtcStatsReport?, isPublisher: Boolean = true) {
         if (stats == null) return

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/MediaManager.kt

@@ -80,6 +80,8 @@ import stream.video.sfu.models.AudioBitrateProfile
 import stream.video.sfu.models.VideoDimension
 import java.nio.ByteBuffer
 import java.util.UUID
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.resumeWithException
 
 sealed class DeviceStatus {
@@ -284,7 +286,7 @@ class ScreenShareManager(
 
     private val logger by taggedLogger("Media:ScreenShareManager")
 
-    private val _status = MutableStateFlow<DeviceStatus>(DeviceStatus.NotSelected)
+    internal val _status = MutableStateFlow<DeviceStatus>(DeviceStatus.NotSelected)
     val status: StateFlow<DeviceStatus> = _status
 
     public val isEnabled: StateFlow<Boolean> = _status.mapState { it is DeviceStatus.Enabled }
@@ -549,8 +551,9 @@ class MicrophoneManager(
     // Internal data
     private val logger by taggedLogger("Media:MicrophoneManager")
 
-    private lateinit var audioHandler: AudioHandler
-    private var setupCompleted: Boolean = false
+    private var audioHandler: AudioHandler? = null
+    private var setupCompleted: AtomicBoolean = AtomicBoolean(false)
+    private var mediaManagerSetupState = AtomicReference(MediaManagerSetupState.NONE)
     internal var audioManager: AudioManager? = null
     internal var priorStatus: DeviceStatus? = null
 
@@ -577,7 +580,7 @@ class MicrophoneManager(
     val selectedUsbDevice: StateFlow<UsbAudioInputDevice?> = _selectedUsbDevice
 
     // Exposed state
-    private val _status = MutableStateFlow<DeviceStatus>(DeviceStatus.NotSelected)
+    internal val _status = MutableStateFlow<DeviceStatus>(DeviceStatus.NotSelected)
 
     /** The status of the audio */
     val status: StateFlow<DeviceStatus> = _status
@@ -882,20 +885,43 @@ class MicrophoneManager(
         if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {
             cleanupUsbDeviceDetection()
         }
-        setupCompleted = false
+        setupCompleted.set(false)
+        audioHandler = null
+    }
+
+    /**
+     * Resets the microphone status to NotSelected to allow re-initialization on next join.
+     */
+    internal fun reset() {
+        _status.value = DeviceStatus.NotSelected
+        mediaManagerSetupState.set(MediaManagerSetupState.NONE)
     }
 
     fun canHandleDeviceSwitch() = audioUsageProvider.invoke() != AudioAttributes.USAGE_MEDIA
 
     // Internal logic
     internal fun setup(preferSpeaker: Boolean = false, onAudioDevicesUpdate: (() -> Unit)? = null) {
         synchronized(this) {
+            logger.d {
+                "[setup] setupCompleted = ${setupCompleted.get()}, mediaManagerSetupState = ${mediaManagerSetupState.get()}"
+            }
+            val localMediaManagerSetupState = mediaManagerSetupState.get()
+            when (localMediaManagerSetupState) {
+                MediaManagerSetupState.FINISHED -> {
+                    onAudioDevicesUpdate?.invoke()
+                    return@synchronized
+                }
+                MediaManagerSetupState.STARTED -> return@synchronized // TODO Rahul, ideally the method call should be queued. Test this before merge
+                else -> {}
+            }
+
+            mediaManagerSetupState.set(MediaManagerSetupState.STARTED)
+
             var capturedOnAudioDevicesUpdate = onAudioDevicesUpdate
 
-            if (setupCompleted) {
+            if (setupCompleted.get()) {
                 capturedOnAudioDevicesUpdate?.invoke()
                 capturedOnAudioDevicesUpdate = null
-
                 return
             }
 
@@ -907,41 +933,50 @@ class MicrophoneManager(
                 setupUsbDeviceDetection()
             }
 
-            if (canHandleDeviceSwitch() && !::audioHandler.isInitialized) {
-                audioHandler = AudioSwitchHandler(
-                    context = mediaManager.context,
-                    preferredDeviceList = listOf(
-                        AudioDevice.BluetoothHeadset::class.java,
-                        AudioDevice.WiredHeadset::class.java,
-                    ) + if (preferSpeaker) {
-                        listOf(
-                            AudioDevice.Speakerphone::class.java,
-                            AudioDevice.Earpiece::class.java,
-                        )
-                    } else {
-                        listOf(
-                            AudioDevice.Earpiece::class.java,
-                            AudioDevice.Speakerphone::class.java,
-                        )
-                    },
-                    audioDeviceChangeListener = { devices, selected ->
-                        logger.i { "[audioSwitch] audio devices. selected $selected, available devices are $devices" }
-
-                        _devices.value = devices.map { it.fromAudio() }
-                        _selectedDevice.value = selected?.fromAudio()
-
-                        setupCompleted = true
-
-                        capturedOnAudioDevicesUpdate?.invoke()
-                        capturedOnAudioDevicesUpdate = null
-                    },
-                )
+            if (canHandleDeviceSwitch()) {
+                if (audioHandler == null) {
+                    // First time initialization
+                    audioHandler = AudioSwitchHandler(
+                        context = mediaManager.context,
+                        preferredDeviceList = listOf(
+                            AudioDevice.BluetoothHeadset::class.java,
+                            AudioDevice.WiredHeadset::class.java,
+                        ) + if (preferSpeaker) {
+                            listOf(
+                                AudioDevice.Speakerphone::class.java,
+                                AudioDevice.Earpiece::class.java,
+                            )
+                        } else {
+                            listOf(
+                                AudioDevice.Earpiece::class.java,
+                                AudioDevice.Speakerphone::class.java,
+                            )
+                        },
+                        audioDeviceChangeListener = { devices, selected ->
+                            logger.i { "[audioSwitch] audio devices. selected $selected, available devices are $devices" }
+
+                            _devices.value = devices.map { it.fromAudio() }
+                            _selectedDevice.value = selected?.fromAudio()
+
+                            setupCompleted.set(true)
+                            mediaManagerSetupState.set(MediaManagerSetupState.FINISHED)
+                            capturedOnAudioDevicesUpdate?.invoke()
+                            capturedOnAudioDevicesUpdate = null
+                        },
+                    )
 
-                logger.d { "[setup] Calling start on instance $audioHandler" }
-                audioHandler.start()
+                    logger.d { "[setup] Calling start on instance $audioHandler" }
+                    audioHandler?.start()
+                } else {
+                    // audioHandler exists but was stopped (cleanup was called), restart it
+                    logger.d { "[setup] Restarting audioHandler after cleanup" }
+                    audioHandler?.start()
+                    mediaManagerSetupState.set(MediaManagerSetupState.FINISHED)
+                }
             } else {
-                logger.d { "[MediaManager#setup] Usage is MEDIA or audioHandle is already initialized" }
+                logger.d { "[MediaManager#setup] Usage is MEDIA" }
                 capturedOnAudioDevicesUpdate?.invoke()
+                mediaManagerSetupState.set(MediaManagerSetupState.FINISHED)
             }
         }
     }
@@ -952,7 +987,7 @@ class MicrophoneManager(
     )
 
     private fun ifAudioHandlerInitialized(then: (audioHandler: AudioSwitchHandler) -> Unit) {
-        if (this::audioHandler.isInitialized) {
+        if (audioHandler != null) {
             then(this.audioHandler as AudioSwitchHandler)
         } else {
             logger.e { "Audio handler not initialized. Ensure calling setup(), before using the handler." }
@@ -998,7 +1033,7 @@ class CameraManager(
     private val logger by taggedLogger("Media:CameraManager")
 
     /** The status of the camera. enabled or disabled */
-    private val _status = MutableStateFlow<DeviceStatus>(DeviceStatus.NotSelected)
+    internal val _status = MutableStateFlow<DeviceStatus>(DeviceStatus.NotSelected)
     public val status: StateFlow<DeviceStatus> = _status
 
     /** Represents whether the camera is enabled */
@@ -1343,6 +1378,13 @@ class CameraManager(
         setupCompleted = false
     }
 
+    /**
+     * Resets the camera status to NotSelected to allow re-initialization on next join.
+     */
+    internal fun reset() {
+        _status.value = DeviceStatus.NotSelected
+    }
+
     private fun createCameraDeviceWrapper(
         id: String,
         cameraManager: CameraManager?,
@@ -1424,6 +1466,7 @@ class MediaManagerImpl(
     val audioUsage: Int = defaultAudioUsage,
     val audioUsageProvider: (() -> Int) = { audioUsage },
 ) {
+    private val logger by taggedLogger("MediaManagerImpl")
     internal val camera =
         CameraManager(this, eglBaseContext, DefaultCameraCharacteristicsValidator())
     internal val microphone = MicrophoneManager(this, audioUsage, audioUsageProvider)
@@ -1545,6 +1588,20 @@ class MediaManagerImpl(
         // Cleanup camera and microphone infrastructure
         camera.cleanup()
         microphone.cleanup()
+        reset()
+    }
+
+    /**
+     * Resets device statuses to NotSelected to allow re-initialization on next join.
+     * Should be called after cleanup when preparing for rejoin.
+     */
+    internal fun reset() {
+        logger.d { "[reset]" }
+        camera.reset()
+        microphone.reset()
+
+        speaker._status.value = DeviceStatus.NotSelected
+        screenShare._status.value = DeviceStatus.NotSelected
     }
 }
 

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/MediaManagerSetupState.kt

@@ -0,0 +1,21 @@
+/*
+ * Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-video-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.video.android.core
+
+internal enum class MediaManagerSetupState {
+    NONE, STARTED, FINISHED
+}

stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/ParticipantState.kt

@@ -20,6 +20,8 @@ import androidx.compose.runtime.Stable
 import io.getstream.android.video.generated.models.MuteUsersResponse
 import io.getstream.log.taggedLogger
 import io.getstream.result.Result
+import io.getstream.video.android.core.coroutines.flows.RestartableStateFlow
+import io.getstream.video.android.core.coroutines.scopes.RestartableProducerScope
 import io.getstream.video.android.core.internal.InternalStreamVideoApi
 import io.getstream.video.android.core.model.AudioTrack
 import io.getstream.video.android.core.model.MediaTrack
@@ -31,10 +33,8 @@ import io.getstream.video.android.core.utils.combineStates
 import io.getstream.video.android.core.utils.mapState
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.combine
-import kotlinx.coroutines.flow.stateIn
 import org.threeten.bp.Instant
 import org.threeten.bp.OffsetDateTime
 import org.threeten.bp.ZoneOffset
@@ -46,24 +46,23 @@ import stream.video.sfu.models.TrackType
  * Represents the state of a participant in a call.
  *
  * * A list of participants is shared when you join a call the SFU send you the participant joined event.
- *
+ * @param sessionId The SFU returns a session id for each participant. This session id is unique
+ * @param scope The coroutine scope for this participant
+ * @param callActions The call actions interface for performing operations on this participant
+ * @param initialUserId The current version of the user, this is the start for participant.user stateflow
+ * @param source A prefix to identify tracks, internal
+ * @param trackLookupPrefix
  */
-@Stable
+@Stable // TODO Rahul, need to fix its breaking change before merge
 public data class ParticipantState(
-    /** The SFU returns a session id for each participant. This session id is unique */
     var sessionId: String = "",
-    /** The coroutine scope for this participant */
     private val scope: CoroutineScope,
-    /** The call actions interface for performing operations on this participant */
     private val callActions: CallActions,
-    /** The current version of the user, this is the start for participant.user stateflow */
     private val initialUserId: String,
     val source: ParticipantSource = ParticipantSource.PARTICIPANT_SOURCE_WEBRTC_UNSPECIFIED,
-    /** A prefix to identify tracks, internal */
     @InternalStreamVideoApi
     var trackLookupPrefix: String = "",
 ) {
-
     private val logger by taggedLogger("ParticipantState")
 
     val isLocal by lazy {
@@ -156,18 +155,22 @@ public data class ParticipantState(
     internal val _reactions = MutableStateFlow<List<Reaction>>(emptyList())
     val reactions: StateFlow<List<Reaction>> = _reactions
 
-    val video: StateFlow<Video?> = combine(
-        _videoTrack,
-        _videoEnabled,
-        _videoPaused,
-    ) { track, enabled, paused ->
-        Video(
-            sessionId = sessionId,
-            track = track,
-            enabled = enabled,
-            paused = paused,
-        )
-    }.stateIn(scope, SharingStarted.Lazily, null)
+    val video: StateFlow<Video?> = RestartableStateFlow(
+        combine(
+            _videoTrack,
+            _videoEnabled,
+            _videoPaused,
+        ) { track, enabled, paused ->
+            Video(
+                sessionId = sessionId,
+                track = track,
+                enabled = enabled,
+                paused = paused,
+            )
+        },
+        scope as RestartableProducerScope,
+        null,
+    )
 
     val audio: StateFlow<Audio?> = combineStates(_audioTrack, _audioEnabled) { track, enabled ->
         Audio(