[ECO-5009][CHA-RL1] Room ATTACH with retry

chat-android/build.gradle.kts

@@ -45,6 +45,7 @@ buildConfig {
 dependencies {
     api(libs.ably.android)
     implementation(libs.gson)
+    implementation(libs.coroutine.core)
 
     testImplementation(libs.junit)
     testImplementation(libs.mockk)

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

@@ -0,0 +1,93 @@
+package com.ably.chat
+
+import java.util.concurrent.PriorityBlockingQueue
+import kotlin.coroutines.cancellation.CancellationException
+import kotlin.coroutines.coroutineContext
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancelChildren
+import kotlinx.coroutines.launch
+
+/**
+ * AtomicCoroutineScope is a thread safe wrapper to run multiple operations mutually exclusive.
+ * All operations are atomic and run with given priority.
+ * Accepts scope as a constructor parameter to run operations under the given scope.
+ * See [Kotlin Dispatchers](https://kt.academy/article/cc-dispatchers) for more information.
+ */
+class AtomicCoroutineScope(private val scope: CoroutineScope = CoroutineScope(Dispatchers.Default)) {
+
+    private val sequentialScope: CoroutineScope = CoroutineScope(Dispatchers.Default.limitedParallelism(1))
+
+    private class Job<T : Any>(
+        private val priority: Int,
+        val coroutineBlock: suspend CoroutineScope.() -> T,
+        val deferredResult: CompletableDeferred<T>,
+        val queuedPriority: Int,
+    ) : Comparable<Job<*>> {
+        override fun compareTo(other: Job<*>) = when {
+            this.priority == other.priority -> this.queuedPriority.compareTo(other.queuedPriority)
+            else -> this.priority.compareTo(other.priority)
+        }
+    }
+
+    // Handles jobs of any type
+    private val jobs: PriorityBlockingQueue<Job<*>> = PriorityBlockingQueue() // Accessed from both sequentialScope and async method
+    private var isRunning = false // Only accessed from sequentialScope
+    private var queueCounter = 0 // Only accessed from synchronized method
+
+    val finishedProcessing: Boolean
+        get() = jobs.isEmpty() && !isRunning
+
+    val pendingJobCount: Int
+        get() = jobs.size
+
+    /**
+     * Defines priority for the operation execution and
+     * executes given coroutineBlock mutually exclusive under given scope.
+     */
+    @Synchronized
+    fun <T : Any> async(priority: Int = 0, coroutineBlock: suspend CoroutineScope.() -> T): CompletableDeferred<T> {
+        val deferredResult = CompletableDeferred<T>()
+        jobs.add(Job(priority, coroutineBlock, deferredResult, queueCounter++))
+        sequentialScope.launch {
+            if (!isRunning) {
+                isRunning = true
+                while (jobs.isNotEmpty()) {
+                    val job = jobs.poll()
+                    job?.let {
+                        safeExecute(it)
+                    }
+                }
+                isRunning = false
+            }
+        }
+        return deferredResult
+    }
+
+    private suspend fun <T : Any> safeExecute(job: Job<T>) {
+        try {
+            // Appends coroutineContext to cancel current/pending jobs when AtomicCoroutineScope is cancelled
+            scope.launch(coroutineContext) {
+                try {
+                    val result = job.coroutineBlock(this)
+                    job.deferredResult.complete(result)
+                } catch (t: Throwable) {
+                    job.deferredResult.completeExceptionally(t)
+                }
+            }.join()
+        } catch (t: Throwable) {
+            job.deferredResult.completeExceptionally(t)
+        }
+    }
+
+    /**
+     * Cancels ongoing and pending operations with given error.
+     * See [Coroutine cancellation](https://kt.academy/article/cc-cancellation#cancellation-in-a-coroutine-scope) for more information.
+     */
+    @Synchronized
+    fun cancel(message: String?, cause: Throwable? = null) {
+        queueCounter = 0
+        sequentialScope.coroutineContext.cancelChildren(CancellationException(message, cause))
+    }
+}

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

@@ -87,7 +87,7 @@ internal class ChatApi(private val realtimeClient: RealtimeClient, private val c
                 ErrorInfo(
                     "Metadata contains reserved 'ably-chat' key",
                     HttpStatusCodes.BadRequest,
-                    ErrorCodes.InvalidRequestBody,
+                    ErrorCodes.InvalidRequestBody.errorCode,
                 ),
             )
         }
@@ -98,7 +98,7 @@ internal class ChatApi(private val realtimeClient: RealtimeClient, private val c
                 ErrorInfo(
                     "Headers contains reserved key with reserved 'ably-chat' prefix",
                     HttpStatusCodes.BadRequest,
-                    ErrorCodes.InvalidRequestBody,
+                    ErrorCodes.InvalidRequestBody.errorCode,
                 ),
             )
         }

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

@@ -0,0 +1,55 @@
+package com.ably.chat
+
+import io.ably.lib.types.ErrorInfo
+import io.ably.lib.util.EventEmitter
+import io.ably.lib.util.Log
+import io.ably.lib.realtime.ChannelBase as AblyRealtimeChannel
+
+/**
+ * Represents an object that has a channel and therefore may care about discontinuities.
+ */
+interface HandlesDiscontinuity {
+    /**
+     * A promise of the channel that this object is associated with. The promise
+     * is resolved when the feature has finished initializing.
+     */
+    val channel: AblyRealtimeChannel
+
+    /**
+     * Called when a discontinuity is detected on the channel.
+     * @param reason The error that caused the discontinuity.
+     */
+    fun discontinuityDetected(reason: ErrorInfo?)
+}
+
+/**
+ * An interface to be implemented by objects that can emit discontinuities to listeners.
+ */
+interface EmitsDiscontinuities {
+    /**
+     * Register a listener to be called when a discontinuity is detected.
+     * @param listener The listener to be called when a discontinuity is detected.
+     */
+    fun onDiscontinuity(listener: Listener): Subscription
+
+    /**
+     * An interface for listening when discontinuity happens
+     */
+    fun interface Listener {
+        /**
+         * A function that can be called when discontinuity happens.
+         * @param reason reason for discontinuity
+         */
+        fun discontinuityEmitted(reason: ErrorInfo?)
+    }
+}
+
+class DiscontinuityEmitter : EventEmitter<String, EmitsDiscontinuities.Listener>() {
+    override fun apply(listener: EmitsDiscontinuities.Listener?, event: String?, vararg args: Any?) {
+        try {
+            listener?.discontinuityEmitted(args[0] as? ErrorInfo?)
+        } catch (t: Throwable) {
+            Log.e("DiscontinuityEmitter", "Unexpected exception calling Discontinuity Listener", t)
+        }
+    }
+}

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

@@ -3,105 +3,106 @@ package com.ably.chat
 /**
  * Error codes for the Chat SDK.
  */
-object ErrorCodes {
+enum class ErrorCodes(val errorCode: Int) {
+
     /**
      * The messages feature failed to attach.
      */
-    const val MessagesAttachmentFailed = 102_001
+    MessagesAttachmentFailed(102_001),
 
     /**
      * The presence feature failed to attach.
      */
-    const val PresenceAttachmentFailed = 102_002
+    PresenceAttachmentFailed(102_002),
 
     /**
      * The reactions feature failed to attach.
      */
-    const val ReactionsAttachmentFailed = 102_003
+    ReactionsAttachmentFailed(102_003),
 
     /**
      * The occupancy feature failed to attach.
      */
-    const val OccupancyAttachmentFailed = 102_004
+    OccupancyAttachmentFailed(102_004),
 
     /**
      * The typing feature failed to attach.
      */
-    const val TypingAttachmentFailed = 102_005
-    // 102006 - 102049 reserved for future use for attachment errors
+    TypingAttachmentFailed(102_005),
+    // 102_006 - 102_049 reserved for future use for attachment errors
 
     /**
      * The messages feature failed to detach.
      */
-    const val MessagesDetachmentFailed = 102_050
+    MessagesDetachmentFailed(102_050),
 
     /**
      * The presence feature failed to detach.
      */
-    const val PresenceDetachmentFailed = 102_051
+    PresenceDetachmentFailed(102_051),
 
     /**
      * The reactions feature failed to detach.
      */
-    const val ReactionsDetachmentFailed = 102_052
+    ReactionsDetachmentFailed(102_052),
 
     /**
      * The occupancy feature failed to detach.
      */
-    const val OccupancyDetachmentFailed = 102_053
+    OccupancyDetachmentFailed(102_053),
 
     /**
      * The typing feature failed to detach.
      */
-    const val TypingDetachmentFailed = 102_054
-    // 102055 - 102099 reserved for future use for detachment errors
+    TypingDetachmentFailed(102_054),
+    // 102_055 - 102_099 reserved for future use for detachment errors
 
     /**
      * The room has experienced a discontinuity.
      */
-    const val RoomDiscontinuity = 102_100
+    RoomDiscontinuity(102_100),
 
     // Unable to perform operation;
 
     /**
      * Cannot perform operation because the room is in a failed state.
      */
-    const val RoomInFailedState = 102_101
+    RoomInFailedState(102_101),
 
     /**
      * Cannot perform operation because the room is in a releasing state.
      */
-    const val RoomIsReleasing = 102_102
+    RoomIsReleasing(102_102),
 
     /**
      * Cannot perform operation because the room is in a released state.
      */
-    const val RoomIsReleased = 102_103
+    RoomIsReleased(102_103),
 
     /**
      * Cannot perform operation because the previous operation failed.
      */
-    const val PreviousOperationFailed = 102_104
+    PreviousOperationFailed(102_104),
 
     /**
      * An unknown error has happened in the room lifecycle.
      */
-    const val RoomLifecycleError = 102_105
+    RoomLifecycleError(102_105),
 
     /**
      * The request cannot be understood
      */
-    const val BadRequest = 40_000
+    BadRequest(40_000),
 
     /**
      * Invalid request body
      */
-    const val InvalidRequestBody = 40_001
+    InvalidRequestBody(40_001),
 
     /**
      * Internal error
      */
-    const val InternalError = 50_000
+    InternalError(50_000),
 }
 
 /**