feat: Rest and Realtime clients adapters

Adapters for the Kotlin API for `ably-java` and `ably-android`

pubsub-adapter/src/main/kotlin/io/ably/lib/Utils.kt

@@ -0,0 +1,38 @@
+package io.ably.lib
+
+import com.ably.query.OrderBy
+import com.ably.query.TimeUnit
+import io.ably.lib.types.Param
+
+fun buildStatsParams(
+  start: Long?,
+  end: Long?,
+  limit: Int,
+  orderBy: OrderBy,
+  unit: TimeUnit,
+) = buildList {
+  addAll(buildHistoryParams(start, end, limit, orderBy))
+  add(Param("unit", unit.toString()))
+}
+
+fun buildHistoryParams(
+  start: Long?,
+  end: Long?,
+  limit: Int,
+  orderBy: OrderBy,
+) = buildList {
+  start?.let { add(Param("start", it)) }
+  end?.let { add(Param("end", it)) }
+  add(Param("limit", limit))
+  add(Param("direction", orderBy.direction))
+}
+
+fun buildRestPresenceParams(
+  limit: Int,
+  clientId: String?,
+  connectionId: String?,
+) = buildList {
+  add(Param("limit", limit))
+  clientId?.let { add(Param("clientId", it)) }
+  connectionId?.let { add(Param("connectionId", it)) }
+}

pubsub-adapter/src/main/kotlin/io/ably/lib/realtime/RealtimeChannelAdapter.kt

@@ -0,0 +1,68 @@
+package io.ably.lib.realtime
+
+import com.ably.Subscription
+import com.ably.pubsub.RealtimeChannel
+import com.ably.pubsub.RealtimePresence
+import com.ably.query.OrderBy
+import io.ably.lib.buildHistoryParams
+import io.ably.lib.types.*
+
+internal class RealtimeChannelAdapter(private val javaChannel: Channel) : RealtimeChannel {
+  override val name: String
+    get() = javaChannel.name
+  override val presence: RealtimePresence
+    get() = RealtimePresenceAdapter(javaChannel.presence)
+  override val state: ChannelState
+    get() = javaChannel.state
+  override val reason: ErrorInfo
+    get() = javaChannel.reason
+  override val properties: ChannelProperties
+    get() = javaChannel.properties
+
+  override fun attach(listener: CompletionListener?) = javaChannel.attach(listener)
+
+  override fun detach(listener: CompletionListener?) = javaChannel.detach(listener)
+
+  override fun subscribe(listener: ChannelBase.MessageListener): Subscription {
+    javaChannel.subscribe(listener)
+    return Subscription {
+      javaChannel.unsubscribe(listener)
+    }
+  }
+
+  override fun subscribe(eventName: String, listener: ChannelBase.MessageListener): Subscription {
+    javaChannel.subscribe(eventName, listener)
+    return Subscription {
+      javaChannel.unsubscribe(eventName, listener)
+    }
+  }
+
+  override fun subscribe(eventNames: List<String>, listener: ChannelBase.MessageListener): Subscription {
+    javaChannel.subscribe(eventNames.toTypedArray(), listener)
+    return Subscription {
+      javaChannel.unsubscribe(eventNames.toTypedArray(), listener)
+    }
+  }
+
+  override fun publish(name: String?, data: Any?, listener: CompletionListener?) =
+    javaChannel.publish(name, data, listener)
+
+  override fun publish(message: Message, listener: CompletionListener?) = javaChannel.publish(message, listener)
+
+  override fun publish(messages: List<Message>, listener: CompletionListener?) =
+    javaChannel.publish(messages.toTypedArray(), listener)
+
+  override fun setOptions(options: ChannelOptions) = javaChannel.setOptions(options)
+
+  override fun history(start: Long?, end: Long?, limit: Int, orderBy: OrderBy): PaginatedResult<Message> =
+    javaChannel.history(buildHistoryParams(start, end, limit, orderBy).toTypedArray())
+
+  override fun historyAsync(
+    callback: Callback<AsyncPaginatedResult<Message>>,
+    start: Long?,
+    end: Long?,
+    limit: Int,
+    orderBy: OrderBy,
+  ) =
+    javaChannel.historyAsync(buildHistoryParams(start, end, limit, orderBy).toTypedArray(), callback)
+}

pubsub-adapter/src/main/kotlin/io/ably/lib/realtime/RealtimeChannelsAdapter.kt

@@ -0,0 +1,20 @@
+package io.ably.lib.realtime
+
+import com.ably.pubsub.Channels
+import com.ably.pubsub.RealtimeChannel
+import io.ably.lib.types.ChannelOptions
+
+internal class RealtimeChannelsAdapter(private val javaChannels: AblyRealtime.Channels) : Channels<RealtimeChannel> {
+  override fun contains(name: String): Boolean = javaChannels.containsKey(name)
+
+  override fun get(name: String): RealtimeChannel = RealtimeChannelAdapter(javaChannels.get(name))
+
+  override fun get(name: String, options: ChannelOptions): RealtimeChannel =
+    RealtimeChannelAdapter(javaChannels.get(name, options))
+
+  override fun release(name: String) = javaChannels.release(name)
+
+  override fun iterator(): Iterator<RealtimeChannel> = iterator {
+    javaChannels.entrySet().forEach { yield(RealtimeChannelAdapter(it.value)) }
+  }
+}

pubsub-adapter/src/main/kotlin/io/ably/lib/realtime/RealtimeClientAdapter.kt

@@ -0,0 +1,71 @@
+package io.ably.lib.realtime
+
+import com.ably.http.HttpMethod
+import com.ably.pubsub.Channels
+import com.ably.pubsub.RealtimeChannel
+import com.ably.pubsub.RealtimeClient
+import com.ably.query.OrderBy
+import com.ably.query.TimeUnit
+import io.ably.lib.buildStatsParams
+import io.ably.lib.http.HttpCore
+import io.ably.lib.push.Push
+import io.ably.lib.rest.Auth
+import io.ably.lib.types.*
+
+/**
+ * Wrapper for Realtime client
+ */
+fun RealtimeClient(javaClient: AblyRealtime): RealtimeClient = RealtimeClientAdapter(javaClient)
+
+internal class RealtimeClientAdapter(private val javaClient: AblyRealtime) : RealtimeClient {
+  override val channels: Channels<out RealtimeChannel>
+    get() = RealtimeChannelsAdapter(javaClient.channels)
+  override val connection: Connection
+    get() = javaClient.connection
+  override val auth: Auth
+    get() = javaClient.auth
+  override val options: ClientOptions
+    get() = javaClient.options
+  override val push: Push
+    get() = javaClient.push
+
+  override fun time(): Long = javaClient.time()
+
+  override fun timeAsync(callback: Callback<Long>) = javaClient.timeAsync(callback)
+
+  override fun stats(
+    start: Long?,
+    end: Long?,
+    limit: Int,
+    orderBy: OrderBy,
+    unit: TimeUnit
+  ): PaginatedResult<Stats> = javaClient.stats(buildStatsParams(start, end, limit, orderBy, unit).toTypedArray())
+
+  override fun statsAsync(
+    callback: Callback<AsyncPaginatedResult<Stats>>,
+    start: Long?,
+    end: Long?,
+    limit: Int,
+    orderBy: OrderBy,
+    unit: TimeUnit
+  ) = javaClient.statsAsync(buildStatsParams(start, end, limit, orderBy, unit).toTypedArray(), callback)
+
+  override fun request(
+    path: String,
+    method: HttpMethod,
+    params: List<Param>,
+    body: HttpCore.RequestBody?,
+    headers: List<Param>,
+  ) = javaClient.request(method.toString(), path, params.toTypedArray(), body, headers.toTypedArray())!!
+
+  override fun requestAsync(
+    path: String,
+    callback: AsyncHttpPaginatedResponse.Callback,
+    method: HttpMethod,
+    params: List<Param>,
+    body: HttpCore.RequestBody?,
+    headers: List<Param>,
+  ) = javaClient.requestAsync(method.toString(), path, params.toTypedArray(), body, headers.toTypedArray(), callback)
+
+  override fun close() = javaClient.close()
+}

pubsub-adapter/src/main/kotlin/io/ably/lib/realtime/RealtimePresenceAdapter.kt

@@ -0,0 +1,73 @@
+package io.ably.lib.realtime
+
+import com.ably.Subscription
+import com.ably.pubsub.RealtimePresence
+import com.ably.query.OrderBy
+import io.ably.lib.buildHistoryParams
+import io.ably.lib.types.*
+import java.util.*
+
+internal class RealtimePresenceAdapter(private val javaPresence: Presence) : RealtimePresence {
+  override fun get(clientId: String?, connectionId: String?, waitForSync: Boolean): List<PresenceMessage> {
+    val params = buildList {
+      clientId?.let { add(Param(Presence.GET_CLIENTID, it)) }
+      connectionId?.let { add(Param(Presence.GET_CONNECTIONID, it)) }
+      add(Param(Presence.GET_WAITFORSYNC, waitForSync))
+    }
+    return javaPresence.get(*params.toTypedArray()).toList()
+  }
+
+  override fun subscribe(listener: Presence.PresenceListener): Subscription {
+    javaPresence.subscribe(listener)
+    return Subscription {
+      javaPresence.unsubscribe(listener)
+    }
+  }
+
+  override fun subscribe(
+    action: PresenceMessage.Action,
+    listener: Presence.PresenceListener,
+  ): Subscription {
+    javaPresence.subscribe(action, listener)
+    return Subscription {
+      javaPresence.unsubscribe(action, listener)
+    }
+  }
+
+  override fun subscribe(
+    actions: EnumSet<PresenceMessage.Action>,
+    listener: Presence.PresenceListener,
+  ): Subscription {
+    javaPresence.subscribe(actions, listener)
+    return Subscription {
+      javaPresence.unsubscribe(actions, listener)
+    }
+  }
+
+  override fun enter(data: Any?, listener: CompletionListener?) = javaPresence.enter(data, listener)
+
+  override fun update(data: Any?, listener: CompletionListener?) = javaPresence.update(data, listener)
+
+  override fun leave(data: Any?, listener: CompletionListener?) = javaPresence.leave(data, listener)
+
+  override fun enterClient(clientId: String, data: Any?, listener: CompletionListener?) =
+    javaPresence.enterClient(clientId, data, listener)
+
+  override fun updateClient(clientId: String, data: Any?, listener: CompletionListener?) =
+    javaPresence.updateClient(clientId, data, listener)
+
+  override fun leaveClient(clientId: String?, data: Any?, listener: CompletionListener?) =
+    javaPresence.leaveClient(clientId, data, listener)
+
+  override fun history(start: Long?, end: Long?, limit: Int, orderBy: OrderBy): PaginatedResult<PresenceMessage> =
+    javaPresence.history(buildHistoryParams(start, end, limit, orderBy).toTypedArray())
+
+  override fun historyAsync(
+    callback: Callback<AsyncPaginatedResult<PresenceMessage>>,
+    start: Long?,
+    end: Long?,
+    limit: Int,
+    orderBy: OrderBy,
+  ) =
+    javaPresence.historyAsync(buildHistoryParams(start, end, limit, orderBy).toTypedArray(), callback)
+}

pubsub-adapter/src/main/kotlin/io/ably/lib/rest/RestChannelAdapter.kt

@@ -0,0 +1,38 @@
+package io.ably.lib.rest
+
+import com.ably.pubsub.RestChannel
+import com.ably.pubsub.RestPresence
+import com.ably.query.OrderBy
+import io.ably.lib.buildHistoryParams
+import io.ably.lib.realtime.CompletionListener
+import io.ably.lib.types.*
+
+internal class RestChannelAdapter(private val javaChannel: Channel) : RestChannel {
+  override val name: String
+    get() = javaChannel.name
+
+  override val presence: RestPresence
+    get() = RestPresenceAdapter(javaChannel.presence)
+
+  override fun publish(name: String?, data: Any?) = javaChannel.publish(name, data)
+
+  override fun publish(messages: List<Message>) = javaChannel.publish(messages.toTypedArray())
+
+  override fun publishAsync(name: String?, data: Any?, listener: CompletionListener) =
+    javaChannel.publishAsync(name, data, listener)
+
+  override fun publishAsync(messages: List<Message>, listener: CompletionListener) =
+    javaChannel.publishAsync(messages.toTypedArray(), listener)
+
+  override fun history(start: Long?, end: Long?, limit: Int, orderBy: OrderBy): PaginatedResult<Message> =
+    javaChannel.history(buildHistoryParams(start, end, limit, orderBy).toTypedArray())
+
+  override fun historyAsync(
+    callback: Callback<AsyncPaginatedResult<Message>>,
+    start: Long?,
+    end: Long?,
+    limit: Int,
+    orderBy: OrderBy,
+  ) =
+    javaChannel.historyAsync(buildHistoryParams(start, end, limit, orderBy).toTypedArray(), callback)
+}

pubsub-adapter/src/main/kotlin/io/ably/lib/rest/RestChannelsAdapter.kt

@@ -0,0 +1,20 @@
+package io.ably.lib.rest
+
+import com.ably.pubsub.Channels
+import com.ably.pubsub.RestChannel
+import io.ably.lib.types.ChannelOptions
+
+internal class RestChannelsAdapter(private val javaChannels: AblyBase.Channels) : Channels<RestChannel> {
+  override fun contains(name: String): Boolean = javaChannels.containsKey(name)
+
+  override fun get(name: String): RestChannel = RestChannelAdapter(javaChannels.get(name))
+
+  override fun get(name: String, options: ChannelOptions): RestChannel =
+    RestChannelAdapter(javaChannels.get(name, options))
+
+  override fun release(name: String) = javaChannels.release(name)
+
+  override fun iterator(): Iterator<RestChannel> = iterator {
+    javaChannels.entrySet().forEach { yield(RestChannelAdapter(it.value)) }
+  }
+}