Merge pull request #1096 from ably/ECO-5065/annotation-support

[ECO-5065] feat: add support for managing message annotations

Author: ttypic

.gitmodules

@@ -1,3 +0,0 @@
-[submodule "lib/src/test/resources/ably-common"]
-	path = lib/src/test/resources/ably-common
-	url = https://github.com/ably/ably-common.git

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

@@ -15,6 +15,7 @@
 import io.ably.lib.http.HttpUtils;
 import io.ably.lib.objects.LiveObjects;
 import io.ably.lib.objects.LiveObjectsPlugin;
+import io.ably.lib.rest.RestAnnotations;
 import io.ably.lib.transport.ConnectionManager;
 import io.ably.lib.transport.ConnectionManager.QueuedMessage;
 import io.ably.lib.transport.Defaults;
@@ -105,6 +106,8 @@ public LiveObjects getObjects() throws AblyException {
         return liveObjectsPlugin.getInstance(name);
     }
 
+    public final RealtimeAnnotations annotations;
+
     /***
      * internal
      *
@@ -887,7 +890,7 @@ private void onMessage(final ProtocolMessage protocolMessage) {
             if(msg.createdAt == null && msg.action == MessageAction.MESSAGE_CREATE) msg.createdAt = msg.timestamp;
 
             try {
-                msg.decode(options, decodingContext);
+                if (msg.data != null) msg.decode(options, decodingContext);
             } catch (MessageDecodeException e) {
                 if (e.errorInfo.code == 40018) {
                     Log.e(TAG, String.format(Locale.ROOT, "Delta message decode failure - %s. Message id = %s, channel = %s", e.errorInfo.message, msg.id, name));
@@ -1310,6 +1313,10 @@ else if(stateChange.current.equals(failureState)) {
         state = ChannelState.initialized;
         this.decodingContext = new DecodingContext();
         this.liveObjectsPlugin = liveObjectsPlugin;
+        this.annotations = new RealtimeAnnotations(
+            this,
+            new RestAnnotations(name, ably.http, ably.options, options)
+        );
     }
 
     void onChannelMessage(ProtocolMessage msg) {
@@ -1376,6 +1383,9 @@ void onChannelMessage(ProtocolMessage msg) {
         case error:
             setFailed(msg.error);
             break;
+        case annotation:
+            annotations.onAnnotation(msg);
+            break;
         default:
             Log.e(TAG, "onChannelMessage(): Unexpected message action (" + msg.action + ")");
         }
@@ -1402,6 +1412,17 @@ public void once(ChannelState state, ChannelStateListener listener) {
         super.once(state.getChannelEvent(), listener);
     }
 
+    /**
+     * (Internal) Sends a protocol message and provides a callback for completion.
+     *
+     * @param protocolMessage the protocol message to be sent
+     * @param listener the listener to be notified upon completion of the message delivery
+     */
+    public void sendProtocolMessage(ProtocolMessage protocolMessage, CompletionListener listener) throws AblyException {
+        ConnectionManager connectionManager = ably.connection.connectionManager;
+        connectionManager.send(protocolMessage, ably.options.queueMessages, listener);
+    }
+
     private static final String TAG = Channel.class.getName();
     final AblyRealtime ably;
     final String basePath;