Merge pull request #1048 from ably/ECO-5139/parse-action-and-serial

[ECO-5139] feat: add `action` and `serial` fields

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

@@ -25,6 +25,7 @@
 import io.ably.lib.types.DeltaExtras;
 import io.ably.lib.types.ErrorInfo;
 import io.ably.lib.types.Message;
+import io.ably.lib.types.MessageAction;
 import io.ably.lib.types.MessageDecodeException;
 import io.ably.lib.types.MessageSerializer;
 import io.ably.lib.types.PaginatedResult;
@@ -843,6 +844,12 @@ private void onMessage(final ProtocolMessage protocolMessage) {
             if(msg.connectionId == null) msg.connectionId = protocolMessage.connectionId;
             if(msg.timestamp == 0) msg.timestamp = protocolMessage.timestamp;
             if(msg.id == null) msg.id = protocolMessage.id + ':' + i;
+            // (TM2p)
+            if(msg.version == null) msg.version = String.format("%s:%03d", protocolMessage.channelSerial, i);
+            // (TM2k)
+            if(msg.serial == null && msg.action == MessageAction.MESSAGE_CREATE) msg.serial = msg.version;
+            // (TM2o)
+            if(msg.createdAt == null && msg.action == MessageAction.MESSAGE_CREATE) msg.createdAt = msg.timestamp;
 
             try {
                 msg.decode(options, decodingContext);

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

@@ -278,6 +278,20 @@ protected Long readLong(final JsonObject map, final String key) {
         return element.getAsLong();
     }
 
+    /**
+     * Read an optional numerical value.
+     * @return The value, or null if the key was not present in the map.
+     * @throws ClassCastException if an element exists for that key and that element is not a {@link JsonPrimitive}
+     * or is not a valid int value.
+     */
+    protected Integer readInt(final JsonObject map, final String key) {
+        final JsonElement element = map.get(key);
+        if (null == element || element instanceof JsonNull) {
+            return null;
+        }
+        return element.getAsInt();
+    }
+
     /* Msgpack processing */
     boolean readField(MessageUnpacker unpacker, String fieldName, MessageFormat fieldType) throws IOException {
         boolean result = true;

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

@@ -46,9 +46,41 @@ public class Message extends BaseMessage {
      */
     public String connectionKey;
 
+    /**
+     * (TM2k) serial string – an opaque string that uniquely identifies the message. If a message received from Ably
+     * (whether over realtime or REST, eg history) with an action of MESSAGE_CREATE does not contain a serial,
+     * the SDK must set it equal to its version.
+     */
+    public String serial;
+
+    /**
+     * (TM2p) version string – an opaque string that uniquely identifies the message, and is different for different versions.
+     * If a message received from Ably over a realtime transport does not contain a version,
+     * the SDK must set it to <channelSerial>:<padded_index> from the channelSerial field of the enclosing ProtocolMessage,
+     * and padded_index is the index of the message inside the messages array of the ProtocolMessage,
+     * left-padded with 0s to three digits (for example, the second entry might be foo:001)
+     */
+    public String version;
+
+    /**
+     * (TM2j) action enum
+     */
+    public MessageAction action;
+
+    /**
+     * (TM2o) createdAt time in milliseconds since epoch. If a message received from Ably
+     * (whether over realtime or REST, eg history) with an action of MESSAGE_CREATE does not contain a createdAt,
+     * the SDK must set it equal to the TM2f timestamp.
+     */
+    public Long createdAt;
+
     private static final String NAME = "name";
     private static final String EXTRAS = "extras";
     private static final String CONNECTION_KEY = "connectionKey";
+    private static final String SERIAL = "serial";
+    private static final String VERSION = "version";
+    private static final String ACTION = "action";
+    private static final String CREATED_AT = "createdAt";
 
     /**
      * Default constructor
@@ -128,6 +160,10 @@ void writeMsgpack(MessagePacker packer) throws IOException {
         int fieldCount = super.countFields();
         if(name != null) ++fieldCount;
         if(extras != null) ++fieldCount;
+        if(serial != null) ++fieldCount;
+        if(version != null) ++fieldCount;
+        if(action != null) ++fieldCount;
+        if(createdAt != null) ++fieldCount;
         packer.packMapHeader(fieldCount);
         super.writeFields(packer);
         if(name != null) {
@@ -138,6 +174,22 @@ void writeMsgpack(MessagePacker packer) throws IOException {
             packer.packString(EXTRAS);
             extras.write(packer);
         }
+        if(serial != null) {
+            packer.packString(SERIAL);
+            packer.packString(serial);
+        }
+        if(version != null) {
+            packer.packString(VERSION);
+            packer.packString(version);
+        }
+        if(action != null) {
+            packer.packString(ACTION);
+            packer.packInt(action.ordinal());
+        }
+        if(createdAt != null) {
+            packer.packString(CREATED_AT);
+            packer.packLong(createdAt);
+        }
     }
 
     Message readMsgpack(MessageUnpacker unpacker) throws IOException {
@@ -157,6 +209,14 @@ Message readMsgpack(MessageUnpacker unpacker) throws IOException {
                 name = unpacker.unpackString();
             } else if (fieldName.equals(EXTRAS)) {
                 extras = MessageExtras.read(unpacker);
+            } else if (fieldName.equals(SERIAL)) {
+                serial = unpacker.unpackString();
+            } else if (fieldName.equals(VERSION)) {
+                version = unpacker.unpackString();
+            } else if (fieldName.equals(ACTION)) {
+                action = MessageAction.tryFindByOrdinal(unpacker.unpackInt());
+            } else if (fieldName.equals(CREATED_AT)) {
+                createdAt = unpacker.unpackLong();
             } else {
                 Log.v(TAG, "Unexpected field: " + fieldName);
                 unpacker.skipValue();
@@ -313,6 +373,12 @@ protected void read(final JsonObject map) throws MessageDecodeException {
             }
             extras = MessageExtras.read((JsonObject) extrasElement);
         }
+
+        serial = readString(map, SERIAL);
+        version = readString(map, VERSION);
+        Integer actionOrdinal = readInt(map, ACTION);
+        action = actionOrdinal == null ? null : MessageAction.tryFindByOrdinal(actionOrdinal);
+        createdAt = readLong(map, CREATED_AT);
     }
 
     public static class Serializer implements JsonSerializer<Message>, JsonDeserializer<Message> {
@@ -328,6 +394,18 @@ public JsonElement serialize(Message message, Type typeOfMessage, JsonSerializat
             if (message.connectionKey != null) {
                 json.addProperty(CONNECTION_KEY, message.connectionKey);
             }
+            if (message.serial != null) {
+                json.addProperty(SERIAL, message.serial);
+            }
+            if (message.version != null) {
+                json.addProperty(VERSION, message.version);
+            }
+            if (message.action != null) {
+                json.addProperty(ACTION, message.action.ordinal());
+            }
+            if (message.createdAt != null) {
+                json.addProperty(CREATED_AT, message.createdAt);
+            }
             return json;
         }
 

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

@@ -0,0 +1,15 @@
+package io.ably.lib.types;
+
+public enum MessageAction {
+    MESSAGE_UNSET, // 0
+    MESSAGE_CREATE, // 1
+    MESSAGE_UPDATE, // 2
+    MESSAGE_DELETE, // 3
+    ANNOTATION_CREATE, // 4
+    ANNOTATION_DELETE, // 5
+    META_OCCUPANCY; // 6
+
+    static MessageAction tryFindByOrdinal(int ordinal) {
+        return values().length <= ordinal ? null: values()[ordinal];
+    }
+}

lib/src/test/java/io/ably/lib/test/realtime/RealtimeMessageTest.java

@@ -3,6 +3,8 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -17,7 +19,9 @@
 import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonPrimitive;
+import io.ably.lib.types.MessageAction;
 import io.ably.lib.types.MessageExtras;
+import io.ably.lib.types.Param;
 import io.ably.lib.util.Serialisation;
 import org.junit.Ignore;
 import org.junit.Rule;
@@ -970,4 +974,40 @@ public void opaque_message_extras() throws AblyException {
             }
         }
     }
+
+    /**
+     * Check that important chat SDK fields are populated (serial, action, createdAt)
+     */
+    @Test
+    public void should_have_serial_action_createdAt() throws AblyException {
+        ClientOptions opts = createOptions(testVars.keys[7].keyStr);
+        opts.clientId = "chat";
+        try (AblyRealtime realtime = new AblyRealtime(opts)) {
+            final Channel channel = realtime.channels.get("foo::$chat::$chatMessages");
+            CompletionWaiter msgComplete = new CompletionWaiter();
+            channel.subscribe(message -> {
+                assertNotNull(message.serial);
+                assertNotNull(message.version);
+                assertNotNull(message.createdAt);
+                assertEquals(MessageAction.MESSAGE_CREATE, message.action);
+                assertEquals("chat.message", message.name);
+                assertEquals("hello world!", ((JsonObject)message.data).get("text").getAsString());
+                msgComplete.onSuccess();
+            });
+
+            /* publish to the channel */
+            JsonObject chatMessage = new JsonObject();
+            chatMessage.addProperty("text", "hello world!");
+            realtime.request(
+                "POST",
+                "/chat/v2/rooms/foo/messages",
+                new Param[] { new Param("v", 3) },
+                HttpUtils.requestBodyFromGson(chatMessage, opts.useBinaryProtocol),
+                null
+            );
+
+            // wait until we get message on the channel
+            assertNull(msgComplete.waitFor(1, 10_000));
+        }
+    }
 }

lib/src/test/java/io/ably/lib/types/MessageTest.java

@@ -46,4 +46,69 @@ public void serialize_message_with_name_and_data() {
         assertEquals("test-data", serializedObject.get("data").getAsString());
         assertEquals("test-name", serializedObject.get("name").getAsString());
     }
+
+    @Test
+    public void serialize_message_with_serial() {
+        // Given
+        Message message = new Message("test-name", "test-data");
+        message.clientId = "test-client-id";
+        message.connectionKey = "test-key";
+        message.action = MessageAction.MESSAGE_CREATE;
+        message.serial = "01826232498871-001@abcdefghij:001";
+
+        // When
+        JsonElement serializedElement = serializer.serialize(message, null, null);
+
+        // Then
+        JsonObject serializedObject = serializedElement.getAsJsonObject();
+        assertEquals("test-client-id", serializedObject.get("clientId").getAsString());
+        assertEquals("test-key", serializedObject.get("connectionKey").getAsString());
+        assertEquals("test-data", serializedObject.get("data").getAsString());
+        assertEquals("test-name", serializedObject.get("name").getAsString());
+        assertEquals(1, serializedObject.get("action").getAsInt());
+        assertEquals("01826232498871-001@abcdefghij:001", serializedObject.get("serial").getAsString());
+    }
+
+    @Test
+    public void deserialize_message_with_serial() throws Exception {
+        // Given
+       JsonObject jsonObject = new JsonObject();
+       jsonObject.addProperty("clientId", "test-client-id");
+       jsonObject.addProperty("data", "test-data");
+       jsonObject.addProperty("name", "test-name");
+       jsonObject.addProperty("action", 1);
+       jsonObject.addProperty("serial", "01826232498871-001@abcdefghij:001");
+
+        // When
+        Message message = Message.fromEncoded(jsonObject, new ChannelOptions());
+
+        // Then
+        assertEquals("test-client-id", message.clientId);
+        assertEquals("test-data", message.data);
+        assertEquals("test-name", message.name);
+        assertEquals(MessageAction.MESSAGE_CREATE, message.action);
+        assertEquals("01826232498871-001@abcdefghij:001", message.serial);
+    }
+
+
+    @Test
+    public void deserialize_message_with_unknown_action() throws Exception {
+        // Given
+        JsonObject jsonObject = new JsonObject();
+        jsonObject.addProperty("clientId", "test-client-id");
+        jsonObject.addProperty("data", "test-data");
+        jsonObject.addProperty("name", "test-name");
+        jsonObject.addProperty("action", 10);
+        jsonObject.addProperty("serial", "01826232498871-001@abcdefghij:001");
+
+        // When
+        Message message = Message.fromEncoded(jsonObject, new ChannelOptions());
+
+        // Then
+        assertEquals("test-client-id", message.clientId);
+        assertEquals("test-data", message.data);
+        assertEquals("test-name", message.name);
+        assertNull(message.action);
+        assertEquals("01826232498871-001@abcdefghij:001", message.serial);
+    }
 }

lib/src/test/resources/local/testAppSpec.json

@@ -19,8 +19,11 @@
 		},
 		{
 			"capability": "{\"persisted:text_protocol:channel0\":[\"publish\",\"subscribe\",\"history\"],\"persisted:text_protocol:channel1\":[\"publish\",\"subscribe\",\"history\"],\"persisted:binary_protocol:channel0\":[\"publish\",\"subscribe\",\"history\"],\"persisted:binary_protocol:channel1\":[\"publish\",\"subscribe\",\"history\"],\"persisted:*\":[\"subscribe\",\"history\"]}"
-		}
-	],
+		},
+    {
+      "capability": "{ \"[*]*\":[\"*\"] }"
+    }
+  ],
 	"namespaces": [
 		{
 			"id": "persisted",
@@ -78,4 +81,4 @@
 			]
 		}
 	]
-}
+}