[AIT-316] feat: introduce support for message annotations

- Added `RealtimeAnnotations` class to manage annotation creation, deletion, and subscription on realtime channels.
- Introduced `Annotation` and `AnnotationAction` types to encapsulate annotation details and actions.
- Extended flags to include `ANNOTATION_PUBLISH` and `ANNOTATION_SUBSCRIBE`.
- Refactored data encoding logic into `ably.util.encoding`.
- Integrated annotation handling into `RealtimeChannel` and `RestChannel`.

Author: ttypic

ably/realtime/annotations.py

@@ -5,7 +5,7 @@
 
 from ably.rest.annotations import RestAnnotations, construct_validate_annotation
 from ably.transport.websockettransport import ProtocolMessageAction
-from ably.types.annotation import Annotation, AnnotationAction
+from ably.types.annotation import AnnotationAction
 from ably.types.channelstate import ChannelState
 from ably.types.flags import Flag
 from ably.util.eventemitter import EventEmitter
@@ -40,13 +40,13 @@ def __init__(self, channel: RealtimeChannel, connection_manager: ConnectionManag
         self.__subscriptions = EventEmitter()
         self.__rest_annotations = RestAnnotations(channel)
 
-    async def publish(self, msg_or_serial, annotation: dict | Annotation, params: dict=None):
+    async def publish(self, msg_or_serial, annotation: dict, params: dict | None = None):
         """
         Publish an annotation on a message via the realtime connection.
 
         Args:
             msg_or_serial: Either a message serial (string) or a Message object
-            annotation: Dict containing annotation properties (type, name, data, etc.) or Annotation object
+            annotation: Dict containing annotation properties (type, name, data, etc.)
             params: Optional dict of query parameters
 
         Returns:
@@ -84,7 +84,12 @@ async def publish(self, msg_or_serial, annotation: dict | Annotation, params: di
         # Send via WebSocket
         await self.__connection_manager.send_protocol_message(protocol_message)
 
-    async def delete(self, msg_or_serial, annotation: dict | Annotation, params=None, timeout=None):
+    async def delete(
+        self,
+        msg_or_serial,
+        annotation: dict,
+        params: dict | None = None,
+    ):
         """
         Delete an annotation on a message.
 
@@ -93,20 +98,16 @@ async def delete(self, msg_or_serial, annotation: dict | Annotation, params=None
 
         Args:
             msg_or_serial: Either a message serial (string) or a Message object
-            annotation: Dict containing annotation properties or Annotation object
+            annotation: Dict containing annotation properties
             params: Optional dict of query parameters
-            timeout: Optional timeout (not used for realtime, kept for compatibility)
 
         Returns:
             None
 
         Raises:
             AblyException: If the request fails or inputs are invalid
         """
-        if isinstance(annotation, Annotation):
-            annotation_values = annotation.as_dict()
-        else:
-            annotation_values = annotation.copy()
+        annotation_values = annotation.copy()
         annotation_values['action'] = AnnotationAction.ANNOTATION_DELETE
         return await self.publish(msg_or_serial, annotation_values, params)
 
@@ -161,13 +162,13 @@ async def subscribe(self, *args):
 
         # Check if ANNOTATION_SUBSCRIBE mode is enabled
         if self.__channel.state == ChannelState.ATTACHED:
-            if not Flag.ANNOTATION_SUBSCRIBE in self.__channel.modes:
+            if Flag.ANNOTATION_SUBSCRIBE not in self.__channel.modes:
                 raise AblyException(
-                    "You are trying to add an annotation listener, but you haven't requested the "
+                    message="You are trying to add an annotation listener, but you haven't requested the "
                     "annotation_subscribe channel mode in ChannelOptions, so this won't do anything "
                     "(we only deliver annotations to clients who have explicitly requested them)",
-                    93001,
-                    400
+                    code=93001,
+                    status_code=400,
                 )
 
     def unsubscribe(self, *args):
@@ -219,7 +220,7 @@ def _process_incoming(self, incoming_annotations):
             annotation_type = annotation.type or ''
             self.__subscriptions._emit(annotation_type, annotation)
 
-    async def get(self, msg_or_serial, params=None):
+    async def get(self, msg_or_serial, params: dict | None = None):
         """
         Retrieve annotations for a message with pagination support.
 

ably/realtime/channel.py

@@ -11,6 +11,7 @@
 from ably.rest.channel import Channels as RestChannels
 from ably.transport.websockettransport import ProtocolMessageAction
 from ably.types.annotation import Annotation
+from ably.types.channelmode import ChannelMode, decode_channel_mode, encode_channel_mode
 from ably.types.channeloptions import ChannelOptions
 from ably.types.channelstate import ChannelState, ChannelStateChange
 from ably.types.flags import Flag, has_flag
@@ -21,7 +22,6 @@
 from ably.util.eventemitter import EventEmitter
 from ably.util.exceptions import AblyException, IncompatibleClientIdException
 from ably.util.helper import Timer, is_callable_or_coroutine, validate_message_size
-from ably.types.channelmode import ChannelMode, decode_channel_mode, encode_channel_mode
 
 if TYPE_CHECKING:
     from ably.realtime.realtime import AblyRealtime
@@ -68,7 +68,7 @@ def __init__(self, realtime: AblyRealtime, name: str, channel_options: ChannelOp
         self.__error_reason: AblyException | None = None
         self.__channel_options = channel_options or ChannelOptions()
         self.__params: dict[str, str] | None = None
-        self.__modes: list[ChannelMode] = list()  # Channel mode flags from ATTACHED message
+        self.__modes: list[ChannelMode] = []  # Channel mode flags from ATTACHED message
 
         # Delta-specific fields for RTL19/RTL20 compliance
         vcdiff_decoder = self.__realtime.options.vcdiff_decoder if self.__realtime.options.vcdiff_decoder else None
@@ -911,6 +911,10 @@ def presence(self):
         """Get the RealtimePresence object for this channel"""
         return self.__presence
 
+    @property
+    def annotations(self) -> RealtimeAnnotations:
+        return self._Channel__annotations
+
     @property
     def modes(self):
         """Get the list of channel modes"""

ably/rest/annotations.py

@@ -9,6 +9,7 @@
 from ably.http.paginatedresult import PaginatedResult, format_params
 from ably.types.annotation import (
     Annotation,
+    AnnotationAction,
     make_annotation_response_handler,
 )
 from ably.types.message import Message
@@ -48,7 +49,7 @@ def serial_from_msg_or_serial(msg_or_serial):
     return message_serial
 
 
-def construct_validate_annotation(msg_or_serial, annotation: dict | Annotation):
+def construct_validate_annotation(msg_or_serial, annotation: dict):
     """
     Construct and validate an Annotation from input values.
 
@@ -71,11 +72,8 @@ def construct_validate_annotation(msg_or_serial, annotation: dict | Annotation):
             status_code=400,
             code=40003,
         )
-    elif isinstance(annotation, Annotation):
-        annotation_values = annotation.as_dict()
-    else:
-        annotation_values = annotation
 
+    annotation_values = annotation.copy()
     annotation_values['message_serial'] = message_serial
 
     return Annotation.from_values(annotation_values)
@@ -108,23 +106,27 @@ def __base_path_for_serial(self, serial):
         channel_path = '/channels/{}/'.format(parse.quote_plus(self.__channel.name, safe=':'))
         return channel_path + 'messages/' + parse.quote_plus(serial, safe=':') + '/annotations'
 
-    async def publish(self, msg_or_serial, annotation_values, params=None, timeout=None):
+    async def publish(
+        self,
+        msg_or_serial,
+        annotation: dict | Annotation,
+        params: dict | None = None,
+    ):
         """
         Publish an annotation on a message.
 
         Args:
             msg_or_serial: Either a message serial (string) or a Message object
-            annotation_values: Dict containing annotation properties (type, name, data, etc.)
+            annotation: Dict containing annotation properties (type, name, data, etc.) or Annotation object
             params: Optional dict of query parameters
-            timeout: Optional timeout for the HTTP request
 
         Returns:
             None
 
         Raises:
             AblyException: If the request fails or inputs are invalid
         """
-        annotation = construct_validate_annotation(msg_or_serial, annotation_values)
+        annotation = construct_validate_annotation(msg_or_serial, annotation)
 
         # Convert to wire format
         request_body = annotation.as_dict(binary=self.__channel.ably.options.use_binary_protocol)
@@ -145,9 +147,14 @@ async def publish(self, msg_or_serial, annotation_values, params=None, timeout=N
             path += '?' + parse.urlencode(params)
 
         # Send request
-        await self.__channel.ably.http.post(path, body=request_body, timeout=timeout)
-
-    async def delete(self, msg_or_serial, annotation_values, params=None, timeout=None):
+        await self.__channel.ably.http.post(path, body=request_body)
+
+    async def delete(
+        self,
+        msg_or_serial,
+        annotation: dict | Annotation,
+        params: dict | None = None,
+    ):
         """
         Delete an annotation on a message.
 
@@ -156,9 +163,8 @@ async def delete(self, msg_or_serial, annotation_values, params=None, timeout=No
 
         Args:
             msg_or_serial: Either a message serial (string) or a Message object
-            annotation_values: Dict containing annotation properties
+            annotation: Dict containing annotation properties or Annotation object
             params: Optional dict of query parameters
-            timeout: Optional timeout for the HTTP request
 
         Returns:
             None
@@ -167,11 +173,14 @@ async def delete(self, msg_or_serial, annotation_values, params=None, timeout=No
             AblyException: If the request fails or inputs are invalid
         """
         # Set action to delete
-        annotation_values = annotation_values.copy()
-        annotation_values['action'] = 'annotation.delete'
-        return await self.publish(msg_or_serial, annotation_values, params, timeout)
+        if isinstance(annotation, Annotation):
+            annotation_values = annotation.as_dict()
+        else:
+            annotation_values = annotation.copy()
+        annotation_values['action'] = AnnotationAction.ANNOTATION_DELETE
+        return await self.publish(msg_or_serial, annotation_values, params)
 
-    async def get(self, msg_or_serial, params=None):
+    async def get(self, msg_or_serial, params: dict | None = None):
         """
         Retrieve annotations for a message with pagination support.
 

ably/rest/auth.py

@@ -90,7 +90,7 @@ def __init__(self, ably: AblyRest | AblyRealtime, options: Options):
     async def get_auth_transport_param(self):
         auth_credentials = {}
         if self.auth_options.client_id:
-            auth_credentials["client_id"] = self.auth_options.client_id
+            auth_credentials["clientId"] = self.auth_options.client_id
         if self.__auth_mechanism == Auth.Method.BASIC:
             key_name = self.__auth_options.key_name
             key_secret = self.__auth_options.key_secret

ably/rest/channel.py

@@ -31,6 +31,8 @@
 
 
 class Channel:
+    __annotations: RestAnnotations
+
     def __init__(self, ably, name, options):
         self.__ably = ably
         self.__name = name
@@ -366,7 +368,7 @@ def presence(self):
         return self.__presence
 
     @property
-    def annotations(self):
+    def annotations(self) -> RestAnnotations:
         return self.__annotations
 
     @options.setter

ably/transport/websockettransport.py

@@ -189,6 +189,7 @@ async def on_protocol_message(self, msg):
             ProtocolMessageAction.DETACHED,
             ProtocolMessageAction.MESSAGE,
             ProtocolMessageAction.PRESENCE,
+            ProtocolMessageAction.ANNOTATION,
             ProtocolMessageAction.SYNC
         ):
             self.connection_manager.on_channel_message(msg)

ably/types/annotation.py

@@ -122,22 +122,17 @@ def as_dict(self, binary=False):
 
         Note: Annotations are not encrypted as they need to be parsed by the server.
         """
-        # Encode data
-        encoded = encode_data(self.data, self._encoding_array, binary)
-
         request_body = {
             'action': int(self.action) if self.action is not None else None,
             'serial': self.serial,
             'messageSerial': self.message_serial,
             'type': self.type,  # Annotation type (not data type)
             'name': self.name,
             'count': self.count,
-            'data': encoded.get('data'),
-            'encoding': encoded.get('encoding', ''),
-            'dataType': encoded.get('type'),  # Data type (not annotation type)
             'clientId': self.client_id or None,
             'timestamp': self.timestamp or None,
             'extras': self.extras,
+            **encode_data(self.data, self._encoding_array, binary)
         }
 
         # None values aren't included

ably/types/channeloptions.py

@@ -2,9 +2,9 @@
 
 from typing import Any
 
+from ably.types.channelmode import ChannelMode
 from ably.util.crypto import CipherParams
 from ably.util.exceptions import AblyException
-from ably.types.channelmode import ChannelMode
 
 
 class ChannelOptions:
@@ -18,7 +18,12 @@ class ChannelOptions:
         Channel parameters that configure the behavior of the channel.
     """
 
-    def __init__(self, cipher: CipherParams | None = None, params: dict | None = None, modes: list[ChannelMode] | None = None):
+    def __init__(
+        self,
+        cipher: CipherParams | None = None,
+        params: dict | None = None,
+        modes: list[ChannelMode] | None = None
+    ):
         self.__cipher = cipher
         self.__params = params
         self.__modes = modes