fixed #15

CHANGELOG.md

 Previous release notes are hosted on [GitHub](https://github.com/smok-serwis/coolamqp/releases).
 Since v1.3.2 they'll be put here and in release description.
 
-v2.0.1
+v2.1.0
 ======
 
 * removed extra logging from argumentify
 * user will be notified upon declaring an auto-delete durable exchange
 * deprecated Consumer(fail_on_first_time_resource_locked)
+* we now support [streams](https://www.rabbitmq.com/docs/streams)
 
 v2.0.0
 ======

README.md

@@ -17,6 +17,7 @@ Why CoolAMQP?
     * [Negative Acknowledgements](https://www.rabbitmq.com/docs/nack)
 * traceable using [opentracing](https://opentracing.io/)
 * code coverage is 81% at the moment
+* full support for [streams](https://www.rabbitmq.com/docs/streams)
 * 120 second stress tests are part of each release
 
 

coolamqp/__init__.py

-__version__ = '2.0.1a2'
+__version__ = '2.1.0a1'

docker-compose.yml

 version: '3.2'
 services:
   amqp:
-    image: rabbitmq:3-management
+    image: rabbitmq:4.0-management
   unittest:
+    command: nose2 -vv tests.test_clustering.test_streams
     build:
       context: .
       dockerfile: tests/Dockerfile

docs/index.rst

@@ -9,6 +9,7 @@ Welcome to CoolAMQP's documentation!
     tutorials
     how-to-guide
     caveats
+    streams
     tracing
     reference
     frames

docs/streams.rst

+Streams
+=======
+
+RabbitMQ 4 added a new feature called `streams <https://www.rabbitmq.com/docs/streams>`_ , perhaps to look more like Kafka.
+Anyway, we fully support this feature, but there are a few caveats you must watch out for:
+
+If you specify argument :code:`x-stream-offset` and aim to provide a number, please don't provide it as a string.
+This will cause your connection to RabbitMQ to crash.
+You can naturally provide "first" or "next" or "last".
+
+Basically everything works `as documented <https://www.rabbitmq.com/docs/streams#consuming>`_. Happy usage!

tests/test_clustering/test_streams.py

@@ -12,23 +12,24 @@ from coolamqp.exceptions import ConnectionDead
 from coolamqp.objects import NodeDefinition, Queue, Exchange, Message
 
 NODE = NodeDefinition(os.environ.get('AMQP_HOST', '127.0.0.1'), 'guest', 'guest', heartbeat=20)
+logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.DEBUG)
 logging.getLogger('coolamqp').setLevel(logging.DEBUG)
-logger = logging.getLogger(__name__)
 
 
-class TestConnecting(unittest.TestCase):
+class TestStreams(unittest.TestCase):
 
-    def test_argumented_exchange(self):
+    def test_streams(self):
         c = Cluster(NODE)
         c.start(wait=True, timeout=None)
-        if c.properties.version[0] < 4:
+        if not c.properties.properties['version'].startswith('4'):
             c.shutdown(wait=True)
             return
 
         stream = Queue('my-stream', durable=True, auto_delete=False, exclusive=False,
                        arguments={'x-queue-type': 'stream'})
-        c.declare(stream)
+        c.declare(stream).result()
+
         for i in range(10):
             c.publish(Message(('dupa%s' % (i, )).encode('utf-8')), routing_key='my-stream', confirm=True).result()
 
@@ -36,15 +37,15 @@ class TestConnecting(unittest.TestCase):
 
         def handle_msg(msg):
             test['a'] += 1
+            msg.ack()
 
-        cons, fut = c.consume(stream, qos=10, on_message=handle_msg, arguments={'x-stream-offset': '3'}, no_ack=True)
+        cons, fut = c.consume(stream, qos=10, on_message=handle_msg, arguments={'x-stream-offset': 4}, no_ack=False)
         fut.result()
-        logger.warning(repr(test))
         time.sleep(3)
-        cons.cancel()
-
-        cons, fut = c.consume(stream, qos=10, on_message=handle_msg, arguments={'x-stream-offset': 'first'}, no_ack=True)
+        cons.cancel().result()
+        self.assertEqual(test['a'], 6)
+        cons, fut = c.consume(stream, qos=10, on_message=handle_msg, arguments={'x-stream-offset': 'first'}, no_ack=False)
         fut.result()
         time.sleep(3)
         cons.cancel()
-        logger.warning(repr(test))
+        self.assertEqual(test['a'], 16)