From 0c65aff13c8e61da502f4e21762a8985897deccf Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 27 Nov 2024 00:48:27 +0000 Subject: [PATCH] [ECO-5139] feat: add `action` and `serial` fields Add 256 bit AES CBC encrypted variable length data generated by Java client library SDK (#49) --- .../io/ably/lib/realtime/ChannelBase.java | 7 ++ .../java/io/ably/lib/types/BaseMessage.java | 14 ++++ .../main/java/io/ably/lib/types/Message.java | 78 +++++++++++++++++++ .../java/io/ably/lib/types/MessageAction.java | 15 ++++ .../test/realtime/RealtimeMessageTest.java | 40 ++++++++++ .../java/io/ably/lib/types/MessageTest.java | 65 ++++++++++++++++ lib/src/test/resources/local/testAppSpec.json | 9 ++- 7 files changed, 225 insertions(+), 3 deletions(-) create mode 100644 lib/src/main/java/io/ably/lib/types/MessageAction.java diff --git a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java b/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java index 2ef6a3718..b84ba7dc0 100644 --- a/lib/src/main/java/io/ably/lib/realtime/ChannelBase.java +++ b/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); diff --git a/lib/src/main/java/io/ably/lib/types/BaseMessage.java b/lib/src/main/java/io/ably/lib/types/BaseMessage.java index a46b73b20..44b91d7e2 100644 --- a/lib/src/main/java/io/ably/lib/types/BaseMessage.java +++ b/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; diff --git a/lib/src/main/java/io/ably/lib/types/Message.java b/lib/src/main/java/io/ably/lib/types/Message.java index 9551c0c26..99eed55f5 100644 --- a/lib/src/main/java/io/ably/lib/types/Message.java +++ b/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 : 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, JsonDeserializer { @@ -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; } diff --git a/lib/src/main/java/io/ably/lib/types/MessageAction.java b/lib/src/main/java/io/ably/lib/types/MessageAction.java new file mode 100644 index 000000000..8c80e914c --- /dev/null +++ b/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]; + } +} diff --git a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeMessageTest.java b/lib/src/test/java/io/ably/lib/test/realtime/RealtimeMessageTest.java index c161ce93e..2d00524f1 100644 --- a/lib/src/test/java/io/ably/lib/test/realtime/RealtimeMessageTest.java +++ b/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)); + } + } } diff --git a/lib/src/test/java/io/ably/lib/types/MessageTest.java b/lib/src/test/java/io/ably/lib/types/MessageTest.java index 3abeb9fe6..9e58d9c3b 100644 --- a/lib/src/test/java/io/ably/lib/types/MessageTest.java +++ b/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); + } } diff --git a/lib/src/test/resources/local/testAppSpec.json b/lib/src/test/resources/local/testAppSpec.json index 979c2cca9..a0721dbcb 100644 --- a/lib/src/test/resources/local/testAppSpec.json +++ b/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 @@ ] } ] -} \ No newline at end of file +}