Added abstract liveobjects serialization impl. using msgpack and interfaces

Author: sacOO7

lib/src/main/java/io/ably/lib/objects/Helpers.java

@@ -0,0 +1,21 @@
+package io.ably.lib.objects;
+
+import io.ably.lib.util.Log;
+
+public class Helpers {
+
+    private static final String TAG = Helpers.class.getName();
+    public static final LiveObjectsSerializer liveObjectsSerializer = getLiveObjectsSerializer();
+
+    private static LiveObjectsSerializer getLiveObjectsSerializer() {
+        try {
+            // Replace with the fully qualified name of the implementing class
+            Class<?> clazz = Class.forName("io.ably.lib.objects.DefaultLiveObjectsSerializer");
+            return (LiveObjectsSerializer) clazz.getDeclaredConstructor().newInstance();
+        } catch (Exception e) {
+            // log the error using Log.e
+            Log.e(TAG, ": Failed to create LiveObjectsSerializer instance", e);
+            return null;
+        }
+    }
+}

lib/src/main/java/io/ably/lib/objects/LiveObjectsSerializer.java

@@ -0,0 +1,27 @@
+package io.ably.lib.objects;
+
+import io.ably.lib.plugins.PluginSerializer;
+import org.jetbrains.annotations.NotNull;
+import org.msgpack.core.MessagePacker;
+import org.msgpack.core.MessageUnpacker;
+
+import java.io.IOException;
+
+public abstract class LiveObjectsSerializer implements PluginSerializer {
+    @NotNull
+    public Object[] readMsgpackArray(@NotNull MessageUnpacker unpacker) throws IOException {
+        int count = unpacker.unpackArrayHeader();
+        Object[] result = new Object[count];
+        for(int i = 0; i < count; i++)
+            result[i] = readMsgpack(unpacker);
+        return result;
+    }
+
+    public void writeMsgpackArray(@NotNull Object[] objects, @NotNull MessagePacker packer) throws IOException {
+        int count = objects.length;
+        packer.packArrayHeader(count);
+        for(Object object : objects) {
+            writeMsgpack(object, packer);
+        }
+    }
+}

lib/src/main/java/io/ably/lib/plugins/PluginSerializer.java

@@ -0,0 +1,34 @@
+package io.ably.lib.plugins;
+
+import org.jetbrains.annotations.NotNull;
+import org.msgpack.core.MessagePacker;
+import org.msgpack.core.MessageUnpacker;
+
+import java.io.IOException;
+
+/**
+ * The `PluginSerializer` interface defines methods for serializing and deserializing objects
+ * using the MessagePack format. Implementations of this interface are responsible for
+ * converting objects to and from MessagePack binary format.
+ */
+public interface PluginSerializer {
+
+    /**
+     * Reads and deserializes an object from a `MessageUnpacker` instance.
+     *
+     * @param unpacker The `MessageUnpacker` used to read the serialized data.
+     * @return The deserialized object.
+     * @throws IOException If an I/O error occurs during deserialization.
+     */
+    @NotNull
+    Object readMsgpack(@NotNull MessageUnpacker unpacker) throws IOException;
+
+    /**
+     * Serializes an object and writes it to a `MessagePacker` instance.
+     *
+     * @param obj The object to be serialized.
+     * @param packer The `MessagePacker` used to write the serialized data.
+     * @throws IOException If an I/O error occurs during serialization.
+     */
+    void writeMsgpack(@NotNull Object obj, @NotNull MessagePacker packer) throws IOException;
+}

lib/src/main/java/io/ably/lib/types/ProtocolMessage.java

@@ -4,6 +4,7 @@
 import java.lang.reflect.Type;
 import java.util.Map;
 
+import org.jetbrains.annotations.Nullable;
 import org.msgpack.core.MessageFormat;
 import org.msgpack.core.MessagePacker;
 import org.msgpack.core.MessageUnpacker;
@@ -18,6 +19,8 @@
 
 import io.ably.lib.util.Log;
 
+import static io.ably.lib.objects.Helpers.liveObjectsSerializer;
+
 /**
  * A message sent and received over the Realtime protocol.
  * A ProtocolMessage always relates to a single channel only, but
@@ -116,6 +119,11 @@ public ProtocolMessage(Action action, String channel) {
     public ConnectionDetails connectionDetails;
     public AuthDetails auth;
     public Map<String, String> params;
+    /**
+     * This will be null if we skipped decoding this property due to user not requesting Objects functionality
+     */
+    @Nullable
+    public Object[] state;
 
     public boolean hasFlag(final Flag flag) {
         return (flags & flag.getMask()) == flag.getMask();
@@ -139,6 +147,7 @@ void writeMsgpack(MessagePacker packer) throws IOException {
         if(flags != 0) ++fieldCount;
         if(params != null) ++fieldCount;
         if(channelSerial != null) ++fieldCount;
+        if(state != null) ++fieldCount;
         packer.packMapHeader(fieldCount);
         packer.packString("action");
         packer.packInt(action.getValue());
@@ -174,6 +183,14 @@ void writeMsgpack(MessagePacker packer) throws IOException {
             packer.packString("channelSerial");
             packer.packString(channelSerial);
         }
+        if(state != null) {
+            if (liveObjectsSerializer != null) {
+                packer.packString("state");
+                liveObjectsSerializer.writeMsgpackArray(state, packer);
+            } else {
+                Log.w(TAG, "Skipping 'state' field serialization because LiveObjectsSerializer is not set");
+            }
+        }
     }
 
     ProtocolMessage readMsgpack(MessageUnpacker unpacker) throws IOException {
@@ -233,6 +250,14 @@ ProtocolMessage readMsgpack(MessageUnpacker unpacker) throws IOException {
                 case "params":
                     params = MessageSerializer.readStringMap(unpacker);
                     break;
+                case "state":
+                    if (liveObjectsSerializer != null) {
+                        state = liveObjectsSerializer.readMsgpackArray(unpacker);
+                    } else {
+                        Log.w(TAG, "Skipping 'state' field deserialization because LiveObjectsSerializer is not set");
+                        unpacker.skipValue();
+                    }
+                    break;
                 default:
                     Log.v(TAG, "Unexpected field: " + fieldName);
                     unpacker.skipValue();

live-objects/build.gradle.kts

@@ -9,6 +9,7 @@ repositories {
 
 dependencies {
     implementation(project(":java"))
+    implementation(libs.bundles.common)
     testImplementation(kotlin("test"))
     implementation(libs.coroutine.core)
 

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

@@ -23,7 +23,7 @@ internal suspend fun LiveObjectsAdapter.sendAsync(message: ProtocolMessage) {
   deferred.await()
 }
 
-internal enum class MessageFormat(private val value: String) {
+internal enum class ProtocolMessageFormat(private val value: String) {
   MSGPACK("msgpack"),
   JSON("json");
 

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

@@ -1,5 +1,6 @@
 package io.ably.lib.objects
 
+import org.msgpack.core.MessageUnpacker
 import java.nio.ByteBuffer
 
 /**
@@ -195,7 +196,7 @@ internal data class ObjectOperation(
   /** The initial value encoding defines how the initialValue should be interpreted.
    * Spec: OOP3i
    */
-  val initialValueEncoding: MessageFormat? = null
+  val initialValueEncoding: ProtocolMessageFormat? = null
 )
 
 /**

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

@@ -0,0 +1,35 @@
+package io.ably.lib.objects
+
+import org.msgpack.core.MessagePacker
+import org.msgpack.core.MessageUnpacker
+
+/**
+ * Base class for serializing and deserializing live objects.
+ * Initializes with a default serializer that uses MessagePack format.
+ * Initialized by the LiveObjectsAdapter.
+ */
+internal class DefaultLiveObjectsSerializer: LiveObjectsSerializer() {
+  override fun readMsgpack(unpacker: MessageUnpacker): Any {
+    return getObjectMessageFromMsgpack(unpacker)
+  }
+
+  override fun writeMsgpack(obj: Any, packer: MessagePacker) {
+    if (obj is ObjectMessage) {
+      obj.writeMsgpack(packer)
+    }
+  }
+}
+
+/**
+ * Extension function to deserialize an ObjectMessage from a MessageUnpacker.
+ */
+internal fun getObjectMessageFromMsgpack(unpacker: MessageUnpacker): ObjectMessage {
+  TODO("Not yet implemented")
+}
+
+/**
+ * Extension function to serialize an ObjectMessage to a MessagePacker.
+ */
+private fun ObjectMessage.writeMsgpack(packer: MessagePacker) {
+  TODO("Not yet implemented")
+}