[ECO-5063] feat: add SDK tracking for Rest and Realtime calls

chat-android/build.gradle.kts

@@ -51,13 +51,16 @@ buildConfig {
 
 dependencies {
     api(libs.ably.android)
+    implementation(libs.ably.pubsub.adapter)
     implementation(libs.gson)
     implementation(libs.coroutine.core)
 
     testImplementation(libs.junit)
     testImplementation(libs.mockk)
     testImplementation(libs.coroutine.test)
     testImplementation(libs.bundles.ktor.client)
+    testImplementation(libs.nanohttpd)
+    testImplementation(libs.turbine)
     androidTestImplementation(libs.androidx.test.core)
     androidTestImplementation(libs.androidx.test.runner)
     androidTestImplementation(libs.androidx.junit)

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

@@ -1,5 +1,7 @@
 package com.ably.chat
 
+import com.ably.http.HttpMethod
+import com.ably.pubsub.RealtimeClient
 import com.google.gson.JsonElement
 import io.ably.lib.types.AblyException
 import io.ably.lib.types.AsyncHttpPaginatedResponse
@@ -30,7 +32,7 @@ internal class ChatApi(
         val params = fromSerial?.let { baseParams + Param("fromSerial", it) } ?: baseParams
         return makeAuthorizedPaginatedRequest(
             url = "/chat/v2/rooms/$roomId/messages",
-            method = "GET",
+            method = HttpMethod.Get,
             params = params,
         ) {
             val messageJsonObject = it.requireJsonObject()
@@ -62,7 +64,7 @@ internal class ChatApi(
 
         return makeAuthorizedRequest(
             "/chat/v2/rooms/$roomId/messages",
-            "POST",
+            HttpMethod.Post,
             body,
         )?.let {
             val serial = it.requireString(MessageProperty.Serial)
@@ -92,7 +94,7 @@ internal class ChatApi(
         // CHA-M8c
         return makeAuthorizedRequest(
             "/chat/v2/rooms/${message.roomId}/messages/${message.serial}",
-            "PUT",
+            HttpMethod.Put,
             body,
         )?.let {
             val version = it.requireString(MessageProperty.Version)
@@ -122,7 +124,7 @@ internal class ChatApi(
 
         return makeAuthorizedRequest(
             "/chat/v2/rooms/${message.roomId}/messages/${message.serial}/delete",
-            "POST",
+            HttpMethod.Post,
             body,
         )?.let {
             val version = it.requireString(MessageProperty.Version)
@@ -148,7 +150,7 @@ internal class ChatApi(
      * return occupancy for specified room
      */
     suspend fun getOccupancy(roomId: String): OccupancyEvent {
-        return this.makeAuthorizedRequest("/chat/v2/rooms/$roomId/occupancy", "GET")?.let {
+        return this.makeAuthorizedRequest("/chat/v2/rooms/$roomId/occupancy", HttpMethod.Get)?.let {
             OccupancyEvent(
                 connections = it.requireInt("connections"),
                 presenceMembers = it.requireInt("presenceMembers"),
@@ -158,17 +160,17 @@ internal class ChatApi(
 
     private suspend fun makeAuthorizedRequest(
         url: String,
-        method: String,
+        method: HttpMethod,
         body: JsonElement? = null,
     ): JsonElement? = suspendCancellableCoroutine { continuation ->
         val requestBody = body.toRequestBody()
         realtimeClient.requestAsync(
-            method,
-            url,
-            arrayOf(apiProtocolParam),
-            requestBody,
-            arrayOf(),
-            object : AsyncHttpPaginatedResponse.Callback {
+            path = url,
+            method = method,
+            params = listOf(apiProtocolParam),
+            body = requestBody,
+            headers = listOf(),
+            callback = object : AsyncHttpPaginatedResponse.Callback {
                 override fun onResponse(response: AsyncHttpPaginatedResponse?) {
                     continuation.resume(response?.items()?.firstOrNull())
                 }
@@ -192,17 +194,17 @@ internal class ChatApi(
 
     private suspend fun <T> makeAuthorizedPaginatedRequest(
         url: String,
-        method: String,
+        method: HttpMethod,
         params: List<Param> = listOf(),
         transform: (JsonElement) -> T?,
     ): PaginatedResult<T> = suspendCancellableCoroutine { continuation ->
         realtimeClient.requestAsync(
-            method,
-            url,
-            (params + apiProtocolParam).toTypedArray(),
-            null,
-            arrayOf(),
-            object : AsyncHttpPaginatedResponse.Callback {
+            method = method,
+            path = url,
+            params = params + apiProtocolParam,
+            body = null,
+            headers = listOf(),
+            callback = object : AsyncHttpPaginatedResponse.Callback {
                 override fun onResponse(response: AsyncHttpPaginatedResponse?) {
                     continuation.resume(response.toPaginatedResult(transform))
                 }

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

@@ -1,8 +1,9 @@
 package com.ably.chat
 
+import com.ably.pubsub.WrapperSdkProxyOptions
+import com.ably.pubsub.createWrapperSdkProxy
 import io.ably.lib.realtime.AblyRealtime
-
-typealias RealtimeClient = AblyRealtime
+import io.ably.lib.realtime.RealtimeClient
 
 /**
  * This is the core client for Ably chat. It provides access to chat rooms.
@@ -27,22 +28,28 @@ interface ChatClient {
     /**
      * The underlying Ably Realtime client.
      */
-    val realtime: RealtimeClient
+    val realtime: AblyRealtime
 
     /**
      * The resolved client options for the client, including any defaults that have been set.
      */
     val clientOptions: ClientOptions
 }
 
-fun ChatClient(realtimeClient: RealtimeClient, clientOptions: ClientOptions = ClientOptions()): ChatClient =
+fun ChatClient(realtimeClient: AblyRealtime, clientOptions: ClientOptions = ClientOptions()): ChatClient =
     DefaultChatClient(realtimeClient, clientOptions)
 
 internal class DefaultChatClient(
-    override val realtime: RealtimeClient,
+    override val realtime: AblyRealtime,
     override val clientOptions: ClientOptions,
 ) : ChatClient {
 
+    private val realtimeClientWrapper = RealtimeClient(realtime).createWrapperSdkProxy(
+        WrapperSdkProxyOptions(
+            agents = mapOf("chat-kotlin" to BuildConfig.APP_VERSION),
+        ),
+    )
+
     private val logger: Logger = if (clientOptions.logHandler != null) {
         CustomLogger(
             clientOptions.logHandler,
@@ -53,23 +60,23 @@ internal class DefaultChatClient(
         AndroidLogger(clientOptions.logLevel, buildLogContext())
     }
 
-    private val chatApi = ChatApi(realtime, clientId, logger.withContext(tag = "AblyChatAPI"))
+    private val chatApi = ChatApi(realtimeClientWrapper, clientId, logger.withContext(tag = "AblyChatAPI"))
 
     override val rooms: Rooms = DefaultRooms(
-        realtimeClient = realtime,
+        realtimeClient = realtimeClientWrapper,
         chatApi = chatApi,
         clientOptions = clientOptions,
         clientId = clientId,
         logger = logger,
     )
 
     override val connection: Connection = DefaultConnection(
-        pubSubConnection = realtime.connection,
+        pubSubConnection = realtimeClientWrapper.connection,
         logger = logger.withContext(tag = "RealtimeConnection"),
     )
 
     override val clientId: String
-        get() = realtime.auth.clientId
+        get() = realtimeClientWrapper.auth.clientId
 
     private fun buildLogContext() = LogContext(
         tag = "ChatClient",
@@ -78,8 +85,8 @@ internal class DefaultChatClient(
             "instanceId" to generateUUID(),
         ),
         dynamicContext = mapOf(
-            "connectionId" to { realtime.connection.id },
-            "connectionState" to { realtime.connection.state.name },
+            "connectionId" to { realtimeClientWrapper.connection.id },
+            "connectionState" to { realtimeClientWrapper.connection.state.name },
         ),
     )
 }

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

@@ -1,17 +1,17 @@
 package com.ably.chat
 
+import com.ably.pubsub.RealtimeChannel
 import io.ably.lib.types.ErrorInfo
 import io.ably.lib.util.EventEmitter
-import io.ably.lib.realtime.ChannelBase as AblyRealtimeChannel
 
 /**
  * Represents an object that has a channel and therefore may care about discontinuities.
  */
-interface HandlesDiscontinuity {
+internal interface HandlesDiscontinuity {
     /**
      * The channel that this object is associated with.
      */
-    val channel: AblyRealtimeChannel
+    val channelWrapper: RealtimeChannel
 
     /**
      * Called when a discontinuity is detected on the channel.

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

@@ -1,6 +1,8 @@
 package com.ably.chat
 
+import com.ably.annotations.InternalAPI
 import com.ably.chat.OrderBy.NewestFirst
+import com.ably.pubsub.RealtimeChannel
 import com.google.gson.JsonObject
 import io.ably.lib.realtime.Channel
 import io.ably.lib.realtime.ChannelState
@@ -314,7 +316,10 @@ internal class DefaultMessages(
      */
     private val messagesChannelName = "${room.roomId}::\$chat::\$chatMessages"
 
-    override val channel: Channel = realtimeChannels.get(messagesChannelName, room.options.messagesChannelOptions()) // CHA-RC2f
+    override val channelWrapper: RealtimeChannel = realtimeChannels.get(messagesChannelName, room.options.messagesChannelOptions())
+
+    @OptIn(InternalAPI::class)
+    override val channel: Channel = channelWrapper.javaChannel // CHA-RC2f
 
     override val attachmentErrorCode: ErrorCode = ErrorCode.MessagesAttachmentFailed
 
@@ -334,7 +339,8 @@ internal class DefaultMessages(
                 updateChannelSerialsAfterDiscontinuity(requireAttachSerial())
             }
         }
-        channel.on(channelStateListener)
+        @OptIn(InternalAPI::class)
+        channelWrapper.javaChannel.on(channelStateListener)
     }
 
     // CHA-M5c, CHA-M5d - Updated channel serial after discontinuity
@@ -372,9 +378,9 @@ internal class DefaultMessages(
         }
         channelSerialMap[messageListener] = deferredChannelSerial
         // (CHA-M4d)
-        channel.subscribe(PubSubEventName.ChatMessage, messageListener)
+        val subscription = channelWrapper.subscribe(PubSubEventName.ChatMessage, messageListener)
         // (CHA-M5) setting subscription point
-        if (channel.state == ChannelState.attached) {
+        if (channelWrapper.state == ChannelState.attached) {
             channelSerialMap[messageListener] = CompletableDeferred(requireChannelSerial())
         }
 
@@ -383,7 +389,7 @@ internal class DefaultMessages(
             roomId = roomId,
             subscription = {
                 channelSerialMap.remove(messageListener)
-                channel.unsubscribe(PubSubEventName.ChatMessage, messageListener)
+                subscription.unsubscribe()
             },
             fromSerialProvider = {
                 channelSerialMap[messageListener]
@@ -431,19 +437,20 @@ internal class DefaultMessages(
         )
 
     private fun requireChannelSerial(): String {
-        return channel.properties.channelSerial
+        return channelWrapper.properties.channelSerial
             ?: throw clientError("Channel has been attached, but channelSerial is not defined")
     }
 
     private fun requireAttachSerial(): String {
-        return channel.properties.attachSerial
+        return channelWrapper.properties.attachSerial
             ?: throw clientError("Channel has been attached, but attachSerial is not defined")
     }
 
     override fun release() {
-        channel.off(channelStateListener)
+        @OptIn(InternalAPI::class)
+        channelWrapper.javaChannel.off(channelStateListener)
         channelSerialMap.clear()
-        realtimeChannels.release(channel.name)
+        realtimeChannels.release(channelWrapper.name)
     }
 }
 

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

@@ -2,6 +2,7 @@
 
 package com.ably.chat
 
+import com.ably.pubsub.RealtimeChannel
 import com.google.gson.JsonObject
 import com.google.gson.JsonPrimitive
 import io.ably.lib.realtime.Channel
@@ -85,6 +86,8 @@ internal class DefaultOccupancy(
 
     override val channel: Channel = room.messages.channel
 
+    override val channelWrapper: RealtimeChannel = room.messages.channelWrapper
+
     private val listeners: MutableList<Occupancy.Listener> = CopyOnWriteArrayList()
 
     private val eventBus = MutableSharedFlow<OccupancyEvent>(
@@ -108,11 +111,7 @@ internal class DefaultOccupancy(
             internalChannelListener(it)
         }
 
-        channel.subscribe(occupancyListener)
-
-        occupancySubscription = Subscription {
-            channel.unsubscribe(occupancyListener)
-        }
+        occupancySubscription = channelWrapper.subscribe(occupancyListener).asChatSubscription()
     }
 
     // (CHA-O4)

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

@@ -2,6 +2,8 @@
 
 package com.ably.chat
 
+import com.ably.pubsub.RealtimeChannel
+import com.ably.pubsub.RealtimePresence
 import com.google.gson.JsonElement
 import com.google.gson.JsonObject
 import io.ably.lib.realtime.Channel
@@ -146,9 +148,11 @@ internal class DefaultPresence(
 
     override val channel: Channel = room.messages.channel
 
+    override val channelWrapper: RealtimeChannel = room.messages.channelWrapper
+
     private val logger = room.logger.withContext(tag = "Presence")
 
-    private val presence = channel.presence
+    private val presence: RealtimePresence = channelWrapper.presence
 
     override suspend fun get(waitForSync: Boolean, clientId: String?, connectionId: String?): List<PresenceMember> {
         room.ensureAttached(logger) // CHA-PR6d, CHA-PR6c, CHA-PR6h
@@ -190,11 +194,7 @@ internal class DefaultPresence(
             listener.onEvent(presenceEvent)
         }
 
-        presence.subscribe(presenceListener)
-
-        return Subscription {
-            presence.unsubscribe(presenceListener)
-        }
+        return presence.subscribe(presenceListener).asChatSubscription()
     }
 
     private fun wrapInUserCustomData(data: PresenceData?) = data?.let {

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

@@ -1,5 +1,6 @@
 package com.ably.chat
 
+import com.ably.pubsub.RealtimeClient
 import io.ably.lib.types.ErrorInfo
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineName
@@ -264,6 +265,7 @@ internal class DefaultRoom(
                             featureLogger.debug("ensureAttached(); waiting complete, room is now ATTACHED")
                             attachDeferred.complete(Unit)
                         }
+
                         RoomStatus.Attaching -> statusLifecycle.onChangeOnce {
                             if (it.current == RoomStatus.Attached) {
                                 featureLogger.debug("ensureAttached(); waiting complete, room is now ATTACHED")
@@ -275,6 +277,7 @@ internal class DefaultRoom(
                                 attachDeferred.completeExceptionally(exception)
                             }
                         }
+
                         else -> {
                             featureLogger.error(
                                 "ensureAttached(); waiting complete, room ATTACHING failed with error: ${statusLifecycle.error}",