fix subscribers list (rabbitmq-community#246)

Fixes: rabbitmq-community#241


This PR refactors the consumer/subscriber system to use numeric IDs instead of string names for internal tracking, addressing issues with subscriber list management. The changes enhance performance and eliminate bugs associated with string-based subscriber identification.

- Converts subscriber tracking from string-based names to integer-based IDs for better performance and reliability
- Updates MessageContext and EventContext to include stream information directly instead of requiring lookups
- Adds validation for max_subscribers_by_connection limits and introduces new exception types


Breaking changes
======

1. Subscriber returns not the `subscriber_id`, but instead the`reference`

The bug was here [Tag 0.3.1]:
https://github.com/rabbitmq-community/rstream/blob/654a9ef23118d96098fe00861b3d661da7886030/rstream/consumer.py#L197-L203 

Given two references with the same name the  `subscriber = self._subscribers[reference]`  is not consistent.

With this PR the `_subscribers` is `[int, _Subscriber]`
```python
self._subscribers: dict[int, _Subscriber] = {}
```

Where the int is the subscriber id that _must_ be unique for connection by protocol.

2. remove the `get_stream` function `message_context.consumer.get_stream(message_context.subscriber_name)` 

The `get_stream` is not needed anymore since the `stream` is now passed on the `message_context` and also `event_context` 

3. `subscribe_name` is now optional
```python
class MessageContext:
    consumer: Consumer
    stream: str
    subscriber_id: int
    subscriber_name: Optional[str]
    offset: int
    timestamp: int
```
---------

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

Makefile

@@ -8,7 +8,7 @@ install:
 format:
 	which poetry
 	poetry --version
-	poetry run isort --skip venv .
+	poetry run isort .
 	poetry run black --exclude=venv .
 	poetry run flake8 --exclude=venv,local_tests,docs/examples --max-line-length=120 --ignore=E203,W503,E701,E704
 	poetry run mypy .

docs/examples/basic_consumers/basic_consumer.py

@@ -24,9 +24,11 @@ async def consume():
     loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))
 
     async def on_message(msg: AMQPMessage, message_context: MessageContext):
-        stream = message_context.consumer.get_stream(message_context.subscriber_name)
-        offset = message_context.offset
-        print("Got message: {} from stream {}, offset {}".format(msg, stream, offset))
+        print(
+            "Got message: {} from stream {}, offset {}".format(
+                msg, message_context.stream, message_context.offset
+            )
+        )
 
     await consumer.start()
     await consumer.subscribe(stream=STREAM, callback=on_message, decoder=amqp_decoder)

docs/examples/basic_consumers/basic_consumer_binary.py

@@ -24,9 +24,11 @@ async def consume():
     loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))
 
     async def on_message(msg: AMQPMessage, message_context: MessageContext):
-        stream = message_context.consumer.get_stream(message_context.subscriber_name)
-        offset = message_context.offset
-        print("Got message: {} from stream {}, offset {}".format(msg, stream, offset))
+        print(
+            "Got message: {} from stream {}, offset {}".format(
+                msg, message_context.stream, message_context.offset
+            )
+        )
 
     await consumer.start()
     await consumer.subscribe(stream=STREAM, callback=on_message)

docs/examples/basic_consumers/basic_consumer_offset_next.py

@@ -26,10 +26,13 @@ async def consume():
     loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))
 
     async def on_message(msg: AMQPMessage, message_context: MessageContext):
-        stream = message_context.consumer.get_stream(message_context.subscriber_name)
-        offset = message_context.offset
-
-        print("Got message: {}".format(msg) + " from stream " + stream + " offset: " + str(offset))
+        print(
+            "Got message: {}".format(msg)
+            + " from stream "
+            + message_context.stream
+            + " offset: "
+            + str(message_context.offset)
+        )
 
     await consumer.start()
     await consumer.subscribe(

docs/examples/basic_consumers/basic_consumer_offset_offset.py

@@ -26,9 +26,11 @@ async def consume():
     loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))
 
     async def on_message(msg: AMQPMessage, message_context: MessageContext):
-        stream = message_context.consumer.get_stream(message_context.subscriber_name)
-        offset = message_context.offset
-        print("Got message: {} from stream {}, offset {}".format(msg, stream, offset))
+        print(
+            "Got message: {} from stream {}, offset {}".format(
+                msg, message_context.stream, message_context.offset
+            )
+        )
 
     await consumer.start()
     # Possible values of OffsetType are: FIRST (default), NEXT, LAST, TIMESTAMP and OFFSET

docs/examples/filtering/consumer_filtering.py

@@ -28,9 +28,11 @@ async def consume():
     loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))
 
     async def on_message(msg: AMQPMessage, message_context: MessageContext):
-        stream = message_context.consumer.get_stream(message_context.subscriber_name)
-        offset = message_context.offset
-        print("Got message: {} from stream {}, offset {}".format(msg, stream, offset))
+        print(
+            "Got message: {} from stream {}, offset {}".format(
+                msg, message_context.stream, message_context.offset
+            )
+        )
         print("Application property: " + str(msg.application_properties[b"region"]))
         global cont
         cont = cont + 1

docs/examples/filtering/super_stream_consumer_filtering.py

@@ -24,9 +24,11 @@ async def consume():
     loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))
 
     async def on_message(msg: AMQPMessage, message_context: MessageContext):
-        stream = message_context.consumer.get_stream(message_context.subscriber_name)
-        offset = message_context.offset
-        print("Got message: {} from stream {}, offset {}".format(msg, stream, offset))
+        print(
+            "Got message: {} from stream {}, offset {}".format(
+                msg, message_context.stream, message_context.offset
+            )
+        )
         print("Application property: " + str(msg.application_properties[b"region"]))
         global cont
         cont = cont + 1

docs/examples/manual_server_offset_tracking/consumer.py

@@ -23,19 +23,24 @@ async def on_message(msg: AMQPMessage, message_context: MessageContext):
     global lock
 
     consumer = message_context.consumer
-    stream = await message_context.consumer.stream(message_context.subscriber_name)
-    offset = message_context.offset
 
-    print("Got message: {} from stream {}, offset {}".format(msg, stream, offset))
+    print(
+        "Got message: {} from stream {}, offset {}".format(
+            msg, message_context.stream, message_context.offset
+        )
+    )
 
     # store the offset every 1000 messages received
     async with lock:
         cont = cont + 1
         # store the offset every 1000 messages received
         if cont % 1000 == 0:
-            await consumer.store_offset(
-                stream=stream, offset=offset, subscriber_name=message_context.subscriber_name
-            )
+            if message_context.subscriber_name is not None:
+                await consumer.store_offset(
+                    stream=message_context.stream,
+                    offset=message_context.offset,
+                    subscriber_name=message_context.subscriber_name,
+                )
 
 
 async def consume():

docs/examples/reliable_client/BestPracticesClient.py

@@ -183,9 +183,11 @@ async def on_message(msg: AMQPMessage, message_context: MessageContext):
     messages_consumed = messages_consumed + 1
     # some printf after some messages consumed in order to check that we are working...
     if (messages_consumed % 100000) == 0:
-        stream = await message_context.consumer.stream(message_context.subscriber_name)
-        offset = message_context.offset
-        print("Received message: {} from stream: {} - message offset: {}".format(msg, stream, offset))
+        print(
+            "Received message: {} from stream: {} - message offset: {}".format(
+                msg, message_context.stream, message_context.offset
+            )
+        )
 
 
 async def publish(rabbitmq_configuration: dict):

docs/examples/reliable_client/appsettings.json

@@ -7,8 +7,8 @@
     "Virtualhost": "/",
     "LoadBalancer": true,
     "SuperStream": true,
-    "MaxPublishersByConnection": 256,
-    "MaxSubscribersByConnection": 256,
+    "MaxPublishersByConnection": 250,
+    "MaxSubscribersByConnection": 250,
     "Producers": 3,
     "Consumers": 3,
     "DelayDuringSendMs":0,