Repository: rabbitmq-community/rstream (language: python)

Issue:
Issue when unsubscribing/subscribing again a subscriber_name
Hi,

we would like to seek to a different timestamp while a subscription is already running.
I tried to simplify our current approach to this script:

```python
import asyncio
import datetime

import rstream as rs

STREAM = "hello-stream"
APP_ID = "abcdef123"


async def seek_somewhere(consumer: rs.Consumer):
    await asyncio.sleep(1)
    print("SEEKING NOW")
    await consumer.unsubscribe(APP_ID)

    # await self._setup()
    timestamp = int(datetime.datetime(2023, 7, 18,
                                      tzinfo=datetime.timezone.utc).timestamp())
    offset = rs.ConsumerOffsetSpecification(rs.OffsetType.TIMESTAMP, timestamp)
    await consumer.subscribe(
        stream=STREAM,
        subscriber_name=APP_ID,
        callback=print,
        decoder=rs.amqp_decoder,
        offset_specification=offset
    )
    await consumer.start()
    print("finished seeking")


async def main():
    consumer = rs.Consumer(
        host="localhost",
        username="guest",
        password="guest",
        connection_name=APP_ID
    )
    await consumer.subscribe(
        stream=STREAM,
        subscriber_name=APP_ID,
        callback=print,
        decoder=rs.amqp_decoder,
    )
    task = asyncio.create_task(seek_somewhere(consumer))
    await consumer.run()
    await task


if __name__ == '__main__':
    asyncio.run(main())

```
However, this doesn't work. Messages are received, until the seeking is attempted, after which no more messages are received (but the program keeps running).

What is the best way to do this? Do we have to create a new consumer?

Thanks for any help in advance.


Discussion / hints:
I tried to additionally clear the subscribers first, but that didn't help. Then, I got it working by simply closing the consumer instead of just unsubscribing. However, I don't think this is a good solution, as it has to initiate a new connection.

I had a look at what close() does, and it seems it's just these three steps:
1. unsubscribe
2. remove subscribers
3. close connection

but since manually doing just 1 and 2 didn't work, it seems to require closing the connection, which doesn't make sense to me. Surely I am missing something?
Hey Hi @ptoews thanks to have reported this. I will have a look even if it may take a while!
@ptoews it seems like the issue is when we subscribe again with the same subscriber_name, I need to investigate why this is happening. Can you try to subscribe in seek_somewhere with a different APP_ID?
@DanielePalaia Yes, that did work, thank you!
@ptoews Great! I'll leave this issue open (changing the title) as there is probably a bug in the code 

Fix patch:
diff --git a/pyproject.toml b/pyproject.toml
index d65f62e..8d98774 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "rstream"
-version = "0.20.1"
+version = "0.20.2"
 description = "A python client for RabbitMQ Streams"
 authors = ["George Fortunatov <qweeeze@gmail.com>", "Daniele Palaia <dpalaia@vmware.com>"]
 readme = "README.md"
diff --git a/rstream/consumer.py b/rstream/consumer.py
index 3786601..32ade3f 100644
--- a/rstream/consumer.py
+++ b/rstream/consumer.py
@@ -319,6 +319,12 @@ async def unsubscribe(self, subscriber_name: str) -> None:
         except BaseException as exc:
             logger.warning("exception in unsubscribe of Consumer:" + str(exc))
 
+        stream = subscriber.stream
+
+        if stream in self._clients:
+            await self._clients[stream].remove_stream(stream)
+            await self._clients[stream].free_available_id(subscriber.subscription_id)
+
         del self._subscribers[subscriber_name]
 
     async def query_offset(self, stream: str, subscriber_name: str) -> int:


Test patch:
diff --git a/tests/test_consumer.py b/tests/test_consumer.py
index d31fe3d..3d81b33 100644
--- a/tests/test_consumer.py
+++ b/tests/test_consumer.py
@@ -261,6 +261,28 @@ async def test_consume_with_resubscribe(stream: str, consumer: Consumer, produce
     assert captured == [b"one", b"two"]
 
 
+async def test_consume_with_resubscribe_msg(stream: str, consumer: Consumer, producer: Producer) -> None:
+    captured: list[bytes] = []
+    subscriber_name = await consumer.subscribe(
+        stream, callback=lambda message, message_context: captured.append(bytes(message))
+    )
+    for i in range(100):
+        await producer.send_wait(stream, b"one")
+    await wait_for(lambda: len(captured) >= 100)
+
+    await consumer.unsubscribe(subscriber_name)
+    await consumer.subscribe(
+        stream,
+        subscriber_name=subscriber_name,
+        callback=lambda message, message_context: captured.append(bytes(message)),
+        offset_specification=ConsumerOffsetSpecification(OffsetType.NEXT, None),
+    )
+
+    for i in range(100):
+        await producer.send_wait(stream, b"two")
+    await wait_for(lambda: len(captured) >= 200)
+
+
 async def test_consume_superstream_with_resubscribe(
     super_stream: str, super_stream_consumer: SuperStreamConsumer, super_stream_producer: SuperStreamProducer
 ) -> None: