[ECO-5457] Refactored ObjectsStateSubscription to ObjectsSubscription

- Updated usage doc for the same

Author: sacOO7

…ects/state/ObjectsStateSubscription.java …bly/lib/objects/ObjectsSubscription.javalib/src/main/java/io/ably/lib/objects/state/ObjectsStateSubscription.java renamed to lib/src/main/java/io/ably/lib/objects/ObjectsSubscription.java lib/src/main/java/io/ably/lib/objects/state/ObjectsStateSubscription.java renamed to lib/src/main/java/io/ably/lib/objects/ObjectsSubscription.java

@@ -1,16 +1,18 @@
-package io.ably.lib.objects.state;
+package io.ably.lib.objects;
 
 /**
- * Represents a subscription that can be unsubscribed from.
- * This interface provides a way to clean up and remove subscriptions when they
- * are no longer needed.
+ * Represents a objects subscription that can be unsubscribed from.
+ * This interface provides a way to clean up and remove subscriptions when they are no longer needed.
  * Example usage:
- * ```java
- * ObjectsStateSubscription s = objects.subscribe(ObjectsStateEvent.SYNCING, new ObjectsStateListener() {});
+ * <pre>
+ * {@code
+ * ObjectsSubscription s = objects.subscribe(ObjectsStateEvent.SYNCING, new ObjectsStateListener() {});
  * // Later when done with the subscription
  * s.unsubscribe();
+ * }
+ * </pre>
  */
-public interface ObjectsStateSubscription {
+public interface ObjectsSubscription {
     /**
      * This method should be called when the subscription is no longer needed,
      * it will make sure no further events will be sent to the subscriber and

lib/src/main/java/io/ably/lib/objects/state/ObjectsStateChange.java

@@ -1,5 +1,6 @@
 package io.ably.lib.objects.state;
 
+import io.ably.lib.objects.ObjectsSubscription;
 import org.jetbrains.annotations.NonBlocking;
 import org.jetbrains.annotations.NotNull;
 
@@ -16,7 +17,7 @@ public interface ObjectsStateChange {
      * @return a subscription object that can be used to unsubscribe from the event
      */
     @NonBlocking
-    ObjectsStateSubscription on(@NotNull ObjectsStateEvent event, @NotNull ObjectsStateChange.Listener listener);
+    ObjectsSubscription on(@NotNull ObjectsStateEvent event, @NotNull ObjectsStateChange.Listener listener);
 
     /**
      * Unsubscribes the specified listener from all synchronization state events.

live-objects/src/main/kotlin/io/ably/lib/objects/DefaultLiveObjects.kt

@@ -2,7 +2,6 @@ package io.ably.lib.objects
 
 import io.ably.lib.objects.state.ObjectsStateChange
 import io.ably.lib.objects.state.ObjectsStateEvent
-import io.ably.lib.objects.state.ObjectsStateSubscription
 import io.ably.lib.realtime.ChannelState
 import io.ably.lib.types.Callback
 import io.ably.lib.types.ProtocolMessage
@@ -38,7 +37,8 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
   /**
    * Coroutine scope for handling callbacks asynchronously.
    */
-  private val callbackScope = CoroutineScope(Dispatchers.Default + CoroutineName("LiveObjectsCallback-$channelName"))
+  private val callbackScope =
+    CoroutineScope(Dispatchers.Default + CoroutineName("LiveObjectsCallback-$channelName") + SupervisorJob())
 
   /**
    * Event bus for handling incoming object messages sequentially.
@@ -90,7 +90,7 @@ internal class DefaultLiveObjects(private val channelName: String, internal val
     TODO("Not yet implemented")
   }
 
-  override fun on(event: ObjectsStateEvent, listener: ObjectsStateChange.Listener): ObjectsStateSubscription =
+  override fun on(event: ObjectsStateEvent, listener: ObjectsStateChange.Listener): ObjectsSubscription =
     objectsManager.on(event, listener)
 
   override fun off(listener: ObjectsStateChange.Listener) = objectsManager.off(listener)

live-objects/src/main/kotlin/io/ably/lib/objects/ObjectsState.kt

@@ -2,7 +2,6 @@ package io.ably.lib.objects
 
 import io.ably.lib.objects.state.ObjectsStateChange
 import io.ably.lib.objects.state.ObjectsStateEvent
-import io.ably.lib.objects.state.ObjectsStateSubscription
 import io.ably.lib.util.EventEmitter
 import io.ably.lib.util.Log
 import kotlinx.coroutines.*
@@ -66,9 +65,9 @@ internal abstract class ObjectsStateCoordinator : ObjectsStateChange, HandlesObj
 
   private val emitterScope = CoroutineScope(Dispatchers.Default.limitedParallelism(1) + SupervisorJob())
 
-  override fun on(event: ObjectsStateEvent, listener: ObjectsStateChange.Listener): ObjectsStateSubscription {
+  override fun on(event: ObjectsStateEvent, listener: ObjectsStateChange.Listener): ObjectsSubscription {
     externalObjectStateEmitter.on(event, listener)
-    return ObjectsStateSubscription {
+    return ObjectsSubscription {
       externalObjectStateEmitter.off(event, listener)
     }
   }
@@ -105,7 +104,12 @@ internal abstract class ObjectsStateCoordinator : ObjectsStateChange, HandlesObj
 }
 
 private class ObjectsStateEmitter : EventEmitter<ObjectsStateEvent, ObjectsStateChange.Listener>() {
+  private val tag = "ObjectsStateEmitter"
   override fun apply(listener: ObjectsStateChange.Listener?, event: ObjectsStateEvent?, vararg args: Any?) {
-    listener?.onStateChanged(event!!)
+    try {
+      listener?.onStateChanged(event!!)
+    } catch (t: Throwable) {
+      Log.e(tag, "Error occurred while executing listener callback for event: $event", t)
+    }
   }
 }

live-objects/src/main/kotlin/io/ably/lib/objects/Utils.kt

@@ -3,6 +3,7 @@ package io.ably.lib.objects
 import io.ably.lib.types.AblyException
 import io.ably.lib.types.Callback
 import io.ably.lib.types.ErrorInfo
+import io.ably.lib.util.Log
 import kotlinx.coroutines.*
 
 internal fun ablyException(
@@ -62,7 +63,9 @@ internal fun <T> CoroutineScope.launchWithCallback(callback: Callback<T>, block:
   launch {
     try {
       val result = block()
-      callback.onSuccess(result)
+      try { callback.onSuccess(result) } catch (t: Throwable) {
+        Log.e("asyncCallback", "Error occurred while executing callback's onSuccess handler", t)
+      } // catch and don't rethrow error from callback
     } catch (throwable: Throwable) {
       val exception = throwable as? AblyException
       callback.onError(exception?.errorInfo)