[CHA-RL3][ECO-5011] Room RELEASE with retry

chat-android/src/main/java/com/ably/chat/Messages.kt

@@ -223,10 +223,12 @@ internal class DefaultMessagesSubscription(
 
 internal class DefaultMessages(
     private val roomId: String,
-    realtimeChannels: AblyRealtime.Channels,
+    private val realtimeChannels: AblyRealtime.Channels,
     private val chatApi: ChatApi,
 ) : Messages, ContributesToRoomLifecycleImpl(), ResolvedContributor {
 
+    override val featureName: String = "messages"
+
     private var listeners: Map<Messages.Listener, DeferredValue<String>> = emptyMap()
 
     private var channelStateListener: ChannelStateListener
@@ -239,8 +241,6 @@ internal class DefaultMessages(
      */
     private val messagesChannelName = "$roomId::\$chat::\$chatMessages"
 
-    override val featureName: String = "messages"
-
     override val channel = realtimeChannels.get(messagesChannelName, ChatChannelOptions())
 
     override val contributor: ContributesToRoomLifecycle = this
@@ -303,10 +303,6 @@ internal class DefaultMessages(
 
     override suspend fun send(params: SendMessageParams): Message = chatApi.sendMessage(roomId, params)
 
-    fun release() {
-        channel.off(channelStateListener)
-    }
-
     /**
      * Associate deferred channel serial value with the current channel's serial
      *
@@ -371,6 +367,10 @@ internal class DefaultMessages(
             }
         }
     }
+
+    override fun release() {
+        realtimeChannels.release(channel.name)
+    }
 }
 
 /**

chat-android/src/main/java/com/ably/chat/Occupancy.kt

@@ -80,4 +80,8 @@ internal class DefaultOccupancy(
     override suspend fun get(): OccupancyEvent {
         TODO("Not yet implemented")
     }
+
+    override fun release() {
+        // No need to do anything, since it uses same channel as messages
+    }
 }

chat-android/src/main/java/com/ably/chat/Presence.kt

@@ -167,4 +167,8 @@ internal class DefaultPresence(
     override fun subscribe(listener: Presence.Listener): Subscription {
         TODO("Not yet implemented")
     }
+
+    override fun release() {
+        // No need to do anything, since it uses same channel as messages
+    }
 }

chat-android/src/main/java/com/ably/chat/Room.kt

@@ -105,6 +105,11 @@ interface Room {
      * Detaches from the room to stop receiving events in realtime.
      */
     suspend fun detach()
+
+    /**
+     * Releases the room, underlying channels are removed from the core SDK to prevent leakage.
+     */
+    suspend fun release()
 }
 
 internal class DefaultRoom(
@@ -199,7 +204,7 @@ internal class DefaultRoom(
         reactions.channel.detachCoroutine()
     }
 
-    fun release() {
-        messages.release()
+    override suspend fun release() {
+        _lifecycleManager?.release()
     }
 }

chat-android/src/main/java/com/ably/chat/RoomLifecycleManager.kt

@@ -40,6 +40,11 @@ interface ContributesToRoomLifecycle : EmitsDiscontinuities, HandlesDiscontinuit
      * @returns The error that should be used when the feature fails to detach.
      */
     val detachmentErrorCode: ErrorCodes
+
+    /**
+     * Underlying Realtime feature channel is removed from the core SDK to prevent leakage.
+     */
+    fun release()
 }
 
 abstract class ContributesToRoomLifecycleImpl : ContributesToRoomLifecycle {
@@ -175,6 +180,11 @@ class RoomLifecycleManager(
      */
     private val _firstAttachesCompleted = mutableMapOf<ResolvedContributor, Boolean>()
 
+    /**
+     * Are we in the process of releasing the room?
+     */
+    private var _releaseInProgress = false
+
     /**
      * Retry duration in milliseconds, used by internal doRetry and runDownChannelsOnFailedAttach methods
      */
@@ -589,6 +599,95 @@ class RoomLifecycleManager(
      * state (e.g. attached), release will throw exception.
      */
     internal suspend fun release() {
-        // TODO("Need to impl. room release")
+        val deferredRelease = atomicCoroutineScope.async(LifecycleOperationPrecedence.Release.priority) { // CHA-RL2i
+            // If we're already released, this is a no-op
+            if (_statusLifecycle.status === RoomStatus.Released) {
+                return@async
+            }
+
+            // If we're already detached, then we can transition to released immediately
+            if (_statusLifecycle.status === RoomStatus.Detached ||
+                _statusLifecycle.status === RoomStatus.Initialized
+            ) {
+                _statusLifecycle.setStatus(RoomStatus.Released)
+                return@async
+            }
+
+            // If we're in the process of releasing, we should wait for it to complete
+            if (_releaseInProgress) {
+                return@async listenToRoomRelease()
+            }
+
+            // We force the room status to be releasing
+            clearAllTransientDetachTimeouts()
+            _operationInProgress = true
+            _releaseInProgress = true
+            _statusLifecycle.setStatus(RoomStatus.Releasing)
+
+            // Do the release until it completes
+            return@async releaseChannels()
+        }
+        deferredRelease.await()
+    }
+
+    private suspend fun listenToRoomRelease() = suspendCancellableCoroutine { continuation ->
+        _statusLifecycle.onChangeOnce {
+            if (it.current == RoomStatus.Released) {
+                continuation.resume(Unit)
+            } else {
+                val err = AblyException.fromErrorInfo(
+                    ErrorInfo(
+                        "failed to release room; existing attempt failed${it.errorMessage}",
+                        HttpStatusCodes.InternalServerError,
+                        ErrorCodes.PreviousOperationFailed.errorCode,
+                    ),
+                )
+                continuation.resumeWithException(err)
+            }
+        }
+    }
+
+    /**
+     *  Releases the room by detaching all channels. If the release operation fails, we wait
+     *  a short period and then try again.
+     */
+    private suspend fun releaseChannels() {
+        var contributorsReleased = kotlin.runCatching { doRelease() }
+        while (contributorsReleased.isFailure) {
+            // Wait a short period and then try again
+            delay(_retryDurationInMs)
+            contributorsReleased = kotlin.runCatching { doRelease() }
+        }
+    }
+
+    /**
+     * Performs the release operation. This will detach all channels in the room that aren't
+     * already detached or in the failed state.
+     */
+    @Suppress("RethrowCaughtException")
+    private suspend fun doRelease() = coroutineScope {
+        _contributors.map { contributor: ResolvedContributor ->
+            async {
+                // Failed channels, we can ignore
+                if (contributor.channel.state == ChannelState.failed) {
+                    return@async
+                }
+                // Detached channels, we can ignore
+                if (contributor.channel.state == ChannelState.detached) {
+                    return@async
+                }
+                try {
+                    contributor.channel.detachCoroutine()
+                } catch (ex: Throwable) {
+                    // TODO - log error here before rethrowing
+                    throw ex
+                }
+            }
+        }.awaitAll()
+        _contributors.forEach {
+            it.contributor.release()
+        }
+        _releaseInProgress = false
+        _statusLifecycle.setStatus(RoomStatus.Released)
     }
 }

chat-android/src/main/java/com/ably/chat/RoomReactions.kt

@@ -106,13 +106,14 @@ data class SendReactionParams(
 internal class DefaultRoomReactions(
     roomId: String,
     private val clientId: String,
-    realtimeChannels: AblyRealtime.Channels,
+    private val realtimeChannels: AblyRealtime.Channels,
 ) : RoomReactions, ContributesToRoomLifecycleImpl(), ResolvedContributor {
 
+    override val featureName = "reactions"
+
     private val roomReactionsChannelName = "$roomId::\$chat::\$reactions"
 
     override val channel: AblyRealtimeChannel = realtimeChannels.get(roomReactionsChannelName, ChatChannelOptions())
-    override val featureName = "reactions"
 
     override val contributor: ContributesToRoomLifecycle = this
 
@@ -161,4 +162,8 @@ internal class DefaultRoomReactions(
         channel.subscribe(RoomReactionEventType.Reaction.eventName, messageListener)
         return Subscription { channel.unsubscribe(RoomReactionEventType.Reaction.eventName, messageListener) }
     }
+
+    override fun release() {
+        realtimeChannels.release(channel.name)
+    }
 }

chat-android/src/main/java/com/ably/chat/Rooms.kt

@@ -72,9 +72,7 @@ internal class DefaultRooms(
     }
 
     override suspend fun release(roomId: String) {
-        synchronized(this) {
-            val room = roomIdToRoom.remove(roomId)
-            room?.release()
-        }
+        val room = roomIdToRoom.remove(roomId)
+        room?.release()
     }
 }

chat-android/src/main/java/com/ably/chat/Typing.kt

@@ -107,4 +107,8 @@ internal class DefaultTyping(
     override suspend fun stop() {
         TODO("Not yet implemented")
     }
+
+    override fun release() {
+        realtimeClient.channels.release(channel.name)
+    }
 }

chat-android/src/main/java/com/ably/chat/Utils.kt

@@ -57,6 +57,13 @@ val Channel.errorMessage: String
         ", ${reason.message}"
     }
 
+val RoomStatusChange.errorMessage: String
+    get() = if (error == null) {
+        ""
+    } else {
+        ", ${error.message}"
+    }
+
 @Suppress("FunctionName")
 fun ChatChannelOptions(init: (ChannelOptions.() -> Unit)? = null): ChannelOptions {
     val options = ChannelOptions()