feat: Use of params rather than header values for offsets

Author: Alan Christie

README.md

@@ -212,38 +212,42 @@ for clarity: -
         }
     }
 
+## Version 3
+Version 3 uses **params** for the specification of historical events. Version 2
+used header values to provide these values.
+
 ## Connecting to sockets (historical events)
 The streaming service keeps historical events based on a maximum age and file size.
 Consequently you can connect to your websocket and retrieve these historical events
 as long as they remain in the backend streaming queue. You identify the
-start-time of your events by using **headers** in the websocket request for your
+start-time of your events by using **params** in the websocket request for your
 stream's **LOCATION**.
 
-If you do not provide any header value your socket will only deliver new events.
+If you do not provide any value your socket will only deliver new events.
 
 You can select the start of your events buy providing either an **ordinal**,
-**timestamp**, or **datetime** string. The first event delivered will the next
+**timestamp**, or **datetime**. The first event delivered will be the next
 event after the given reference. For example, if you provide **ordinal** `100`,
 the next event you can expect to receive is an event with **ordinal** `101`.
 
 -   To stream from a specific **ordinal**, provide it as the numerical value
-    of the header property `X-StreamOffsetFromOrdinal`.
+    of the request parameter  `stream_from_ordinal`.
 
 -   To stream from a specific **timestamp**, provide it as the numerical value
-    of the header property `X-StreamOffsetFromTimestamp`.
+    of the request parameter  `stream_from_timestamp`.
 
 -   To stream from a specific **datetime**, provide the date/time string as the value
-    of the header property `X-StreamOffsetFromDatetime`. The datetime string is extremely
+    of the request parameter  `stream_from_datetime`. The datetime string is extremely
     flexible and is interpreted by the [python-dateutil] package's `parse` function.
     UTC is used as the reference for messages, and the string will be interpreted as a
     UTC value if it has no timezone specification. If you are in CEST for example, and
     it is `13:33`, and you want to retrieve times from 13:33 (local time), then you
     will need to provide a string value that has the date you are interested in,
     ans the time set to `11:33` (the UTC time for 13:33 CEST) or specify `13:33+02:00`.
 
-You can only provide one type of historical reference. If you provide a header
-value for `X-StreamOffsetFromOrdinal` for example, you cannot also provide
-a value for `X-StreamOffsetFromTimestamp`.
+You can only provide one type of historical reference. If you provide a
+value for `stream_from_ordinal` for example, you cannot also provide
+a value for `stream_from_timestamp`.
 
 ## Stream storage
 The Account Server (AS) relies on [RabbitMQ] for its event streaming service,
@@ -255,7 +259,7 @@ will just deliver the first event it has available.
 
 You will know that you have lost messages if you keep a record of the most recent
 message **ordinal**. If you received **ordinal** `100` and then, some time later,
-re-connect using an `X-StreamOffsetFromOrdinal` value of `100`, and the first message
+re-connect using an `stream_from_ordinal` value of `100`, and the first message
 your receive has the **ordinal** `150` then you can assume 49 messages have been lost.
 
 Having said all this the AS stream storage configuration is generous and, depending

app/app.py

@@ -7,14 +7,13 @@
 import time
 import uuid as python_uuid
 from logging.config import dictConfig
-from typing import Annotated, Any
+from typing import Any
 from urllib.parse import ParseResult, urlparse
 
 import shortuuid
 from dateutil.parser import parse
 from fastapi import (
     FastAPI,
-    Header,
     HTTPException,
     WebSocket,
     WebSocketDisconnect,
@@ -195,9 +194,9 @@ class EventStreamGetResponse(BaseModel):
 async def event_stream(
     websocket: WebSocket,
     uuid: str,
-    x_streamfromdatetime: Annotated[str | None, Header()] = None,
-    x_streamfromordinal: Annotated[str | None, Header()] = None,
-    x_streamfromtimestamp: Annotated[str | None, Header()] = None,
+    stream_from_datetime: str | None = None,
+    stream_from_ordinal: int | None = None,
+    stream_from_timestamp: int | None = None,
 ):
     """The websocket handler for the event-stream.
     The UUID is returned to the AS when the web-socket is created
@@ -213,14 +212,13 @@ async def event_stream(
     await websocket.accept()
     _LOGGER.info("Accepted connection (uuid=%s)", uuid)
 
-    # Custom request headers.
-    # The following are used to identify the first event in a stream: -
+    # The following parameters are used to identify the first event in a stream: -
     #
-    #   X-StreamFromDatetime - an ISO8601 date/time string
-    #   X-StreamFromTimestamp - an event timestamp (integer) from a prior message
-    #   X-StreamFromOrdinal - a message ordinal (integer 0..N)
+    #   stream_from_datetime - an ISO8601 date/time string
+    #   stream_from_ordinal - a message ordinal (integer 0..N)
+    #   stream_from_timestamp - an event timestamp (integer) from a prior message
     #
-    # Only one of the above is expected.
+    # Only one of the above is permitted.
     num_stream_from_specified: int = 0
     header_value_error: bool = False
     header_value_error_msg: str = ""
@@ -229,11 +227,11 @@ async def event_stream(
         OffsetType.NEXT
     )
     # Was a streaming offset provided?
-    if x_streamfromdatetime:
+    if stream_from_datetime:
         num_stream_from_specified += 1
         try:
-            _LOGGER.info("Found X-StreamFromDatetime=%s", x_streamfromdatetime)
-            from_datetime = parse(x_streamfromdatetime)
+            _LOGGER.info("Given stream_from_datetime=%s", stream_from_datetime)
+            from_datetime = parse(stream_from_datetime)
             # We need a RabbitMQ stream timestamp,
             # which is milliseconds since the universal time epoch (1 Jan, 1970).
             # It's easy to get 'seconds', which we multiply by 1,000
@@ -243,34 +241,34 @@ async def event_stream(
             )
         except:  # pylint: disable=bare-except
             header_value_error = True
-            header_value_error_msg = "Unable to parse X-StreamFromDatetime value"
-    if x_streamfromordinal:
-        _LOGGER.info("Found X-StreamFromOrdinal=%s", x_streamfromordinal)
+            header_value_error_msg = "Unable to parse stream_from_datetime value"
+    if stream_from_ordinal:
+        _LOGGER.info("Given stream_from_ordinal=%d", stream_from_ordinal)
         num_stream_from_specified += 1
         try:
-            from_ordinal = int(x_streamfromordinal)
+            from_ordinal = int(stream_from_ordinal)
             offset_specification = ConsumerOffsetSpecification(
                 OffsetType.OFFSET, from_ordinal
             )
         except ValueError:
             header_value_error = True
-            header_value_error_msg = "X-StreamFromOrdinal must be an integer"
-    if x_streamfromtimestamp:
-        _LOGGER.info("Found X-StreamFromTimestamp=%s", x_streamfromtimestamp)
+            header_value_error_msg = "stream_from_ordinal must be an integer"
+    if stream_from_timestamp:
+        _LOGGER.info("Found stream_from_timestamp=%s", stream_from_timestamp)
         num_stream_from_specified += 1
         try:
-            from_timestamp = int(x_streamfromtimestamp)
+            from_timestamp = int(stream_from_timestamp)
             offset_specification = ConsumerOffsetSpecification(
                 OffsetType.TIMESTAMP, from_timestamp
             )
         except ValueError:
             header_value_error = True
-            header_value_error_msg = "X-StreamFromTimestamp must be an integer"
+            header_value_error_msg = "stream_from_timestamp must be an integer"
 
     # Replace any error with a 'too many values provided' error if necessary
     if num_stream_from_specified > 1:
         header_value_error = True
-        header_value_error_msg = "Cannot provide more than one X-StreamFrom variable"
+        header_value_error_msg = "Cannot provide more than one 'stream_from_' variable"
 
     if header_value_error:
         await websocket.close(

ws_listener.py

@@ -11,19 +11,20 @@
 
 def main(c_args: argparse.Namespace):
     """Connect to the WebSocket and just read messages."""
-    headers = {}
+    params = ""
     if c_args.datetime_offset:
-        headers["X-StreamFromDatetime"] = c_args.datetime_offset
+        params = f"?stream_from_datetime={c_args.datetime_offset}"
     elif c_args.timestamp_offset:
-        headers["X-StreamFromTimestamp"] = c_args.timestamp_offset
+        params = f"?stream_from_timestamp={c_args.timestamp_offset}"
     elif c_args.ordinal_offset:
-        headers["X-StreamFromOrdinal"] = c_args.ordinal_offset
+        params = f"?stream_from_ordinal={c_args.ordinal_offset}"
 
     total_bytes = 0
     total_messages = 0
     min_bytes = 1_000_000
     max_bytes = 0
-    ws = Client.connect(c_args.location, headers=headers)
+    url = f"{c_args.location}{params}"
+    ws = Client.connect(url)
     try:
         while True: