Merge pull request #1216 from ably/chore/liveobjects-basic-implementation-for-path-based-interfaces

[AIT-928] feat(liveobjects): add path-based RealtimeObject and channel.object accessor

Author: sacOO7

lib/src/main/java/io/ably/lib/object/RealtimeObject.java

@@ -0,0 +1,93 @@
+package io.ably.lib.object;
+
+import io.ably.lib.object.path.types.LiveMapPathObject;
+import io.ably.lib.object.state.ObjectStateChange;
+import io.ably.lib.object.state.ObjectStateEvent;
+import io.ably.lib.types.AblyException;
+import io.ably.lib.types.ErrorInfo;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * The RealtimeObject interface is the entry point to the strongly-typed, path-based
+ * LiveObjects API on a channel. It exposes the root of the objects graph as a
+ * {@link LiveMapPathObject} and, via {@link ObjectStateChange}, lets callers observe
+ * synchronization state transitions for the channel's objects.
+ *
+ * <p>Implementations of this interface must be thread-safe as they may be accessed
+ * from multiple threads concurrently.
+ *
+ * <p>Spec: RTO23
+ */
+public interface RealtimeObject extends ObjectStateChange {
+
+    /**
+     * Retrieves a {@link LiveMapPathObject} rooted at the channel's root {@code LiveMap}.
+     * The returned object has an empty path and resolves to the root {@code LiveMap}; use
+     * its navigation methods to address nested values within the objects graph.
+     *
+     * <p>When called without a type variable, we return a default root type which is based
+     * on the globally defined interface for the Objects feature. A user can provide an
+     * explicit type to set the type structure on this particular channel. This is useful
+     * when working with multiple channels with different underlying data structures.
+     *
+     * <p>This operation requires the {@code OBJECT_SUBSCRIBE} channel mode. It implicitly
+     * attaches the channel if it is not already attached; the returned future completes once
+     * the objects synchronization state has transitioned to {@code SYNCED}, and completes
+     * exceptionally with an {@code AblyException} if synchronization fails.
+     *
+     * <p>Spec: RTO23, RTO23f (typed SDKs return a {@link LiveMapPathObject})
+     *
+     * @return a future that completes with the root {@link LiveMapPathObject} for this
+     *         channel's objects graph.
+     */
+    @NotNull
+    CompletableFuture<LiveMapPathObject> get();
+
+    /**
+     * Null-Object guard for {@link RealtimeObject}, used as the value of {@code channel.object}
+     * when the LiveObjects plugin is not installed.
+     *
+     * <p>Because {@code channel.object} is a field, dereferencing it can never throw; instead
+     * every method here fails fast with the plugin-missing error, so {@code get()}, {@code on()},
+     * {@code off()} and {@code offAll()} surface a clear, consistent error rather than a
+     * {@link NullPointerException}.
+     *
+     * <p>A stateless singleton ({@link #INSTANCE}) shared across all channels that lack the
+     * plugin. Adding a method to {@link RealtimeObject} will fail compilation here until it is
+     * guarded, which is the intended safety net.
+     */
+    final class Unavailable implements RealtimeObject {
+
+        public static final Unavailable INSTANCE = new Unavailable();
+
+        private Unavailable() {}
+
+        @Override
+        public @NotNull CompletableFuture<LiveMapPathObject> get() {
+            throw missing();
+        }
+
+        @Override
+        public Subscription on(@NotNull ObjectStateEvent event, ObjectStateChange.@NotNull Listener listener) {
+            throw missing();
+        }
+
+        @Override
+        public void off(ObjectStateChange.@NotNull Listener listener) {
+            throw missing();
+        }
+
+        @Override
+        public void offAll() {
+            throw missing();
+        }
+
+        private static RuntimeException missing() {
+            return new IllegalStateException("LiveObjects plugin hasn't been installed", AblyException.fromErrorInfo(
+                new ErrorInfo("add runtimeOnly('io.ably:liveobjects:<ably-version>') to your dependency tree", 400, 40019)
+            ));
+        }
+    }
+}

lib/src/main/java/io/ably/lib/object/state/ObjectStateChange.java

@@ -0,0 +1,56 @@
+package io.ably.lib.object.state;
+
+import io.ably.lib.object.Subscription;
+import org.jetbrains.annotations.NonBlocking;
+import org.jetbrains.annotations.NotNull;
+
+public interface ObjectStateChange {
+    /**
+     * Subscribes to a specific Objects synchronization state event.
+     *
+     * <p>This method registers the provided listener to be notified when the specified
+     * synchronization state event occurs. The returned subscription can be used to
+     * unsubscribe later when the notifications are no longer needed.
+     *
+     * @param event the synchronization state event to subscribe to (SYNCING or SYNCED)
+     * @param listener the listener that will be called when the event occurs
+     * @return a subscription object that can be used to unsubscribe from the event
+     */
+    @NonBlocking
+    Subscription on(@NotNull ObjectStateEvent event, @NotNull ObjectStateChange.Listener listener);
+
+    /**
+     * Unsubscribes the specified listener from all synchronization state events.
+     *
+     * <p>After calling this method, the provided listener will no longer receive
+     * any synchronization state event notifications.
+     *
+     * @param listener the listener to unregister from all events
+     */
+    @NonBlocking
+    void off(@NotNull ObjectStateChange.Listener listener);
+
+    /**
+     * Unsubscribes all listeners from all synchronization state events.
+     *
+     * <p>After calling this method, no listeners will receive any synchronization
+     * state event notifications until new listeners are registered.
+     */
+    @NonBlocking
+    void offAll();
+
+    /**
+     * Interface for receiving notifications about Objects synchronization state changes.
+     * <p>
+     * Implement this interface and register it with an {@code ObjectStateEmitter} to be notified
+     * when synchronization state transitions occur.
+     */
+    interface Listener {
+        /**
+         * Called when the synchronization state changes.
+         *
+         * @param objectStateEvent The new state event (SYNCING or SYNCED)
+         */
+        void onStateChanged(ObjectStateEvent objectStateEvent);
+    }
+}

lib/src/main/java/io/ably/lib/object/state/ObjectStateEvent.java

@@ -0,0 +1,19 @@
+package io.ably.lib.object.state;
+
+/**
+ * Represents the synchronization state of Ably Objects.
+ * <p>
+ * This enum is used to notify listeners about state changes in the synchronization process.
+ * Clients can register an {@link ObjectStateChange.Listener} to receive these events.
+ */
+public enum ObjectStateEvent {
+    /**
+     * Indicates that synchronization between local and remote objects is in progress.
+     */
+    SYNCING,
+
+    /**
+     * Indicates that synchronization has completed successfully and objects are in sync.
+     */
+    SYNCED
+}

lib/src/main/java/io/ably/lib/realtime/ChannelBase.java

@@ -13,6 +13,7 @@
 import io.ably.lib.http.Http;
 import io.ably.lib.http.HttpCore;
 import io.ably.lib.http.HttpUtils;
+import io.ably.lib.object.RealtimeObject;
 import io.ably.lib.objects.RealtimeObjects;
 import io.ably.lib.objects.LiveObjectsPlugin;
 import io.ably.lib.rest.MessageEditsMixin;
@@ -112,6 +113,8 @@ public abstract class ChannelBase extends EventEmitter<ChannelEvent, ChannelStat
 
     private volatile MessageEditsMixin messageEditsMixin;
 
+    public RealtimeObject object;
+
     public RealtimeObjects getObjects() throws AblyException {
         if (liveObjectsPlugin == null) {
             throw AblyException.fromErrorInfo(
@@ -1695,7 +1698,12 @@ else if(stateChange.current.equals(failureState)) {
         this.decodingContext = new DecodingContext();
         this.liveObjectsPlugin = liveObjectsPlugin;
         if (liveObjectsPlugin != null) {
-            liveObjectsPlugin.getInstance(name); // Make objects instance ready to process sync messages
+            liveObjectsPlugin.getInstance(name);
+            // TODO(objects-migration): assign `this.object` to the real RealtimeObject once the
+            // LiveObjects plugin exposes io.ably.lib.object.RealtimeObject (getInstance currently
+            // returns the legacy io.ably.lib.objects.RealtimeObjects type).
+        } else {
+            this.object = RealtimeObject.Unavailable.INSTANCE;
         }
         this.annotations = new RealtimeAnnotations(
             this,