[AIT-1038] chore(examples): migrate LiveObjects example app to the path-based API

examples/src/main/kotlin/com/ably/example/Utils.kt

@@ -1,211 +1,115 @@
+/**
+ * The LiveObjects bridge layer for the example app: Compose observers plus coroutine
+ * wrappers over the path-based API (`io.ably.lib.liveobjects`).
+ *
+ * The app works exclusively with [io.ably.lib.liveobjects.path.PathObject]s — references
+ * to a *location* in the channel's objects graph, not to a particular object. A PathObject
+ * resolves its path lazily on every call, so a stored reference stays valid even when the
+ * object at that location is replaced (e.g. "Reset all" swaps in brand-new counters), and
+ * a path subscription automatically observes whatever object currently lives there. The
+ * identity-bound alternative, [io.ably.lib.liveobjects.instance.Instance] (obtained via
+ * `pathObject.instance()`), tracks one specific object wherever it moves — not needed here,
+ * since every screen wants location semantics.
+ *
+ * ably-java's typed flavour of the API: the base PathObject exposes only type-agnostic
+ * methods; type-specific reads/writes live on sub-type views reached via `as*` casts
+ * (`asLiveCounter()`, `asLiveMap()`, `asString()`, ...). The casts never throw — a
+ * wrong-typed view degrades gracefully (reads return null/empty, writes fail with an
+ * AblyException). The root obtained from `channel.object.get()` is already a
+ * [LiveMapPathObject], so no cast is needed there.
+ */
 package com.ably.example
 
-import androidx.compose.runtime.*
-import io.ably.lib.objects.ObjectsCallback
-import io.ably.lib.objects.RealtimeObjects
-import io.ably.lib.objects.type.counter.LiveCounter
-import io.ably.lib.objects.type.counter.LiveCounterUpdate
-import io.ably.lib.objects.type.map.LiveMap
-import io.ably.lib.objects.type.map.LiveMapUpdate
-import io.ably.lib.objects.type.map.LiveMapUpdate.Change.UPDATED
-import io.ably.lib.objects.type.map.LiveMapValue
+import androidx.compose.runtime.Composable
+import androidx.compose.runtime.DisposableEffect
+import androidx.compose.runtime.LaunchedEffect
+import androidx.compose.runtime.getValue
+import androidx.compose.runtime.mutableStateOf
+import androidx.compose.runtime.remember
+import androidx.compose.runtime.setValue
+import io.ably.lib.liveobjects.ValueType
+import io.ably.lib.liveobjects.path.types.LiveCounterPathObject
+import io.ably.lib.liveobjects.path.types.LiveMapPathObject
+import io.ably.lib.liveobjects.state.ObjectStateEvent
+import io.ably.lib.liveobjects.value.LiveCounter
+import io.ably.lib.liveobjects.value.LiveMap
+import io.ably.lib.liveobjects.value.LiveMapValue
 import io.ably.lib.realtime.AblyRealtime
 import io.ably.lib.realtime.Channel
 import io.ably.lib.realtime.ChannelState
 import io.ably.lib.realtime.ChannelStateListener
-import io.ably.lib.types.AblyException
 import io.ably.lib.types.ChannelMode
 import io.ably.lib.types.ChannelOptions
-import io.ably.lib.types.ErrorInfo
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.suspendCancellableCoroutine
-import kotlin.coroutines.resume
-
-private suspend fun RealtimeObjects.getRootCoroutines(): LiveMap = suspendCancellableCoroutine { continuation ->
-  getRootAsync(object : ObjectsCallback<LiveMap> {
-    override fun onSuccess(result: LiveMap?) {
-      continuation.resume(result!!)
-    }
-
-    override fun onError(exception: AblyException?) {
-      continuation.cancel(exception)
-    }
-  })
-}
-
-private suspend fun RealtimeObjects.createCounterCoroutine(): LiveCounter =
-  suspendCancellableCoroutine { continuation ->
-    createCounterAsync(object : ObjectsCallback<LiveCounter> {
-      override fun onSuccess(result: LiveCounter?) {
-        continuation.resume(result!!)
-      }
-
-      override fun onError(exception: AblyException?) {
-        continuation.cancel(exception)
-      }
-    })
-  }
-
-private suspend fun RealtimeObjects.createMapCoroutine(): LiveMap = suspendCancellableCoroutine { continuation ->
-  createMapAsync(object : ObjectsCallback<LiveMap> {
-    override fun onSuccess(result: LiveMap?) {
-      continuation.resume(result!!)
-    }
-
-    override fun onError(exception: AblyException?) {
-      continuation.cancel(exception)
-    }
-  })
-}
-
-suspend fun LiveCounter.incrementCoroutine(amount: Int): Unit = supressCoroutineExceptions {
-  suspendCancellableCoroutine { continuation ->
-    incrementAsync(amount, object : ObjectsCallback<Void> {
-      override fun onSuccess(result: Void?) {
-        continuation.resume(Unit)
-      }
-
-      override fun onError(exception: AblyException?) {
-        continuation.cancel(exception)
-      }
-    })
-  }
-}
-
-suspend fun LiveCounter.decrementCoroutine(amount: Int): Unit = supressCoroutineExceptions {
-  suspendCancellableCoroutine { continuation ->
-    decrementAsync(amount, object : ObjectsCallback<Void> {
-      override fun onSuccess(result: Void?) {
-        continuation.resume(Unit)
-      }
-
-      override fun onError(exception: AblyException?) {
-        continuation.cancel(exception)
-      }
-    })
-  }
-}
-
-suspend fun Channel.updateOptions(options: ChannelOptions): Unit = supressCoroutineExceptions {
-  suspendCancellableCoroutine { continuation ->
-    setOptions(options, object : io.ably.lib.realtime.CompletionListener {
-      override fun onSuccess() {
-        continuation.resume(Unit)
-      }
-
-      override fun onError(reason: ErrorInfo?) {
-        continuation.cancel(AblyException.fromErrorInfo(reason))
-      }
-    })
-  }
-}
-
-suspend fun getOrCreateCounter(channel: Channel, root: LiveMap?, path: String): LiveCounter {
-  val mapValue = root?.get(path)
-  if (mapValue == null) {
-    val counter = channel.objects.createCounterCoroutine()
-    root?.setCoroutine(path, LiveMapValue.of(counter))
-    return counter
-  } else {
-    return mapValue.asLiveCounter
-  }
-}
-
-suspend fun getOrCreateMap(channel: Channel, root: LiveMap?, path: String): LiveMap {
-  val mapValue = root?.get(path)
-  if (mapValue == null) {
-    val map = channel.objects.createMapCoroutine()
-    root?.setCoroutine(path, LiveMapValue.of(map))
-    return map
-  } else {
-    return mapValue.asLiveMap
-  }
+import kotlinx.coroutines.future.await
+
+/**
+ * Returns the counter path object at [key] under [root], creating and linking a fresh
+ * zero-value counter when the path is missing or holds a non-counter value. Path objects
+ * re-resolve on every call, so the returned reference stays valid even if another client
+ * replaces the counter object at this key.
+ *
+ * The check-then-set is not atomic: another client can create the same key in between.
+ * That is fine here - the colliding operations converge deterministically via
+ * last-write-wins, and both sides end up bound to whichever counter won.
+ */
+suspend fun getOrCreateCounter(root: LiveMapPathObject, key: String): LiveCounterPathObject {
+  val path = root.get(key)
+  if (path.type != ValueType.LIVE_COUNTER) {
+    // Creates the counter and links it under root in a single published operation
+    root.set(key, LiveMapValue.of(LiveCounter.create())).await()
+  }
+  return path.asLiveCounter()
 }
 
-suspend fun LiveMap.setCoroutine(key: String, value: LiveMapValue) = supressCoroutineExceptions {
-  suspendCancellableCoroutine<Unit> { continuation ->
-    setAsync(key, value, object : ObjectsCallback<Void> {
-      override fun onSuccess(result: Void?) {
-        continuation.resume(Unit)
-      }
-
-      override fun onError(exception: AblyException?) {
-        continuation.cancel(exception)
-      }
-    })
-  }
-}
-
-suspend fun LiveMap.removeCoroutine(key: String) = supressCoroutineExceptions {
-  suspendCancellableCoroutine<Unit> { continuation ->
-    removeAsync(key, object : ObjectsCallback<Void> {
-      override fun onSuccess(result: Void?) {
-        continuation.resume(Unit)
-      }
-
-      override fun onError(exception: AblyException?) {
-        continuation.cancel(exception)
-      }
-    })
-  }
+/**
+ * Returns the map path object at [key] under [root], creating and linking a fresh empty
+ * map when the path is missing or holds a non-map value. Same last-write-wins caveat as
+ * [getOrCreateCounter].
+ */
+suspend fun getOrCreateMap(root: LiveMapPathObject, key: String): LiveMapPathObject {
+  val path = root.get(key)
+  if (path.type != ValueType.LIVE_MAP) {
+    root.set(key, LiveMapValue.of(LiveMap.create())).await()
+  }
+  return path.asLiveMap()
 }
 
 @Composable
-fun observeCounter(channel: Channel, root: LiveMap?, path: String): CounterState {
-  var counter by remember { mutableStateOf<LiveCounter?>(null) }
+fun observeCounter(root: LiveMapPathObject?, key: String): CounterState {
+  var counter by remember { mutableStateOf<LiveCounterPathObject?>(null) }
   var counterValue by remember { mutableStateOf<Int?>(null) }
 
-  LaunchedEffect(root) {
-    supressCoroutineExceptions {
-      counter = getOrCreateCounter(channel, root, path)
+  LaunchedEffect(root, key) {
+    root?.let {
+      supressCoroutineExceptions {
+        counter = getOrCreateCounter(it, key)
+      }
     }
   }
 
   DisposableEffect(counter) {
     counterValue = counter?.value()?.toInt()
 
-    val listener: (LiveCounterUpdate) -> Unit = {
-      counter?.value()?.let {
-        counterValue = it.toInt()
-      }
+    // The path subscription fires both for increments on the counter and for the key
+    // being replaced with a new counter (e.g. "Reset all" on another device); the path
+    // re-resolves on read, so no explicit rebinding is needed.
+    val subscription = counter?.subscribe {
+      counterValue = counter?.value()?.toInt()
     }
 
-    counter?.subscribe(listener)
-
     onDispose {
-      counter?.unsubscribe(listener)
-    }
-  }
-
-  DisposableEffect(root) {
-    val listener: (LiveMapUpdate) -> Unit = { rootUpdate ->
-      val counterHasBeenRemoved = rootUpdate.update
-        .filter { (_, change) -> change == UPDATED }
-        .any { (keyName) -> keyName == path }
-
-      if (counterHasBeenRemoved) root?.get(path)?.asLiveCounter?.let { counter = it }
-    }
-
-    root?.subscribe(listener)
-
-    onDispose {
-      root?.unsubscribe(listener)
+      subscription?.unsubscribe()
     }
   }
 
   return CounterState(counterValue, counter) {
-    coroutineScope {
-      launch {
-        counter = channel.objects.createCounterCoroutine().also {
-          root?.setCoroutine(path, LiveMapValue.of(it))
-        }
-      }
-    }
+    // Reset by replacing the object at this key with a brand-new zero-value counter;
+    // fire-and-forget - the path subscription refreshes the displayed value on ack
+    root?.set(key, LiveMapValue.of(LiveCounter.create()))
   }
 }
 
-data class CounterState(val value: Int?, val counter: LiveCounter?, val reset: suspend () -> Unit)
+data class CounterState(val value: Int?, val counter: LiveCounterPathObject?, val reset: () -> Unit)
 
 @Composable
 fun observeChannelState(channel: Channel): ChannelState {
@@ -227,46 +131,74 @@ fun observeChannelState(channel: Channel): ChannelState {
 }
 
 @Composable
-fun observeMap(channel: Channel, root: LiveMap?, path: String): Pair<Map<String, String>, LiveMap?> {
-  var map by remember { mutableStateOf<LiveMap?>(null) }
+fun observeMap(root: LiveMapPathObject?, key: String): Pair<Map<String, String>, LiveMapPathObject?> {
+  var map by remember { mutableStateOf<LiveMapPathObject?>(null) }
   var mapValue by remember { mutableStateOf<Map<String, String>>(mapOf()) }
 
-  LaunchedEffect(root) {
-    supressCoroutineExceptions {
-      map = getOrCreateMap(channel, root, path)
+  fun readEntries(liveMap: LiveMapPathObject?): Map<String, String> =
+    liveMap?.entries()
+      ?.mapNotNull { (entryKey, valuePath) -> valuePath.asString().value()?.let { entryKey to it } }
+      ?.toMap()
+      ?: mapOf()
+
+  LaunchedEffect(root, key) {
+    root?.let {
+      supressCoroutineExceptions {
+        map = getOrCreateMap(it, key)
+      }
     }
   }
 
   DisposableEffect(map) {
-    map?.entries()?.associate { (key, value) -> key to value.asString }?.let {
-      mapValue = it
+    mapValue = readEntries(map)
+
+    // Fires for every entry change on the map at this path (default subscription depth
+    // covers nested updates), after which entries are re-read from the resolved map.
+    val subscription = map?.subscribe {
+      mapValue = readEntries(map)
     }
 
-    val listener: (LiveMapUpdate) -> Unit = {
-      map?.entries()?.associate { (key, value) -> key to value.asString }?.let {
-        mapValue = it
-      }
+    onDispose {
+      subscription?.unsubscribe()
     }
+  }
+
+  return mapValue to map
+}
 
-    map?.subscribe(listener)
+/**
+ * Observes the channel's objects synchronization state via `channel.object.on(event)`.
+ * Returns the most recent [ObjectStateEvent], or null before any event has been received
+ * (events fire on transitions only, so an already-completed sync emits nothing).
+ */
+@Composable
+fun observeObjectsSyncState(channel: Channel): ObjectStateEvent? {
+  var syncState by remember { mutableStateOf<ObjectStateEvent?>(null) }
+
+  DisposableEffect(channel) {
+    // There is no wildcard subscription - register one listener per event
+    val subscriptions = listOf(ObjectStateEvent.SYNCING, ObjectStateEvent.SYNCED).map { event ->
+      channel.`object`.on(event) { stateEvent -> syncState = stateEvent }
+    }
 
     onDispose {
-      map?.unsubscribe(listener)
+      subscriptions.forEach { it.unsubscribe() }
     }
   }
 
-  return mapValue to map
+  return syncState
 }
 
 @Composable
-fun observeRootObject(channel: Channel): LiveMap? {
+fun observeRootObject(channel: Channel): LiveMapPathObject? {
   val channelState = observeChannelState(channel)
-  var root: LiveMap? by remember { mutableStateOf(null) }
+  var root: LiveMapPathObject? by remember { mutableStateOf(null) }
 
   LaunchedEffect(channelState) {
     if (channelState == ChannelState.attached) {
       supressCoroutineExceptions {
-        root = channel.objects.getRootCoroutines()
+        // Completes once the objects synchronization state has reached SYNCED
+        root = channel.`object`.get().await()
       }
     }
   }