Merge remote-tracking branch 'origin/main' into certificate-fixes-for-windows

Author: Gsantomaggio

.ci/ubuntu/gha-setup.sh

@@ -7,7 +7,7 @@ set -o xtrace
 script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 readonly script_dir
 echo "[INFO] script_dir: '$script_dir'"
-readonly rabbitmq_image=rabbitmq:4.1.0-beta.4-management-alpine
+readonly rabbitmq_image=rabbitmq:4.1.0-management
 
 
 readonly docker_name_prefix='rabbitmq-amqp-python-client'

.gitignore

@@ -15,3 +15,4 @@ __pycache__/
 local*
 .githooks/
 .venv/
+.ci/ubuntu/log/*

Makefile

@@ -7,5 +7,13 @@ rabbitmq-server:
 rabbitmq-server-stop:
 	 ./.ci/ubuntu/gha-setup.sh stop
 
+format:
+	poetry run isort --skip rabbitmq_amqp_python_client/qpid  .
+	poetry run black rabbitmq_amqp_python_client/
+	poetry run black tests/
+	poetry run flake8 --exclude=venv,local_tests,docs/examples,rabbitmq_amqp_python_client/qpid --max-line-length=120 --ignore=E203,W503
+
+test: format
+	poetry run pytest .
 help:
 	cat Makefile

README.md

@@ -1,161 +1,31 @@
-# RabbitMQ AMQP 1.0 Python Client
-
-This library is in early stages of development. It is meant to be used with RabbitMQ 4.0.
-
-# Table of Contents
-
-- [How to Build the project and run the tests](#How-to-Build-the-project-and-run-the-tests)
-- [Installation](#Installation)
-- [Getting started](#Getting-Started)
-    * [Creating a connection](#Creating-a-connection)
-    * [Managing resources](#Managing-resources)
-    * [Publishing messages](#Publishing-messages)
-    * [Consuming messages](#Consuming-messages)
-    * [Support for streams](#support-for-streams)
-    * [SSL connection](#ssl-connections)
-    * [Oauth authentication](#oauth-authentication)
-    * [Managing disconnections](#Managing-disconnections)
-
-
-## How to Build the project and run the tests
-
-- Start a RabbitMQ 4.x broker
-- poetry build: build the source project
-- poetry install: resolves and install dependencies
-- poetry run pytest: run the tests
-
-## Installation
+## RabbitMQ AMQP 1.0 Python Client
+This library is meant to be used with RabbitMQ 4.0. Suitable for testing in pre-production environments.
 
 The client is distributed via [`PIP`](https://pypi.org/project/rabbitmq-amqp-python-client/):
 ```bash
  pip install rabbitmq-amqp-python-client
 ```
 
-## Getting Started
-
-An example is provided [`here`](./examples/getting_started/getting_started.py) you can run it after starting a RabbitMQ 4.0 broker with:
-
-poetry run python ./examples/getting_started/getting_started.py
-
-Also consider to have a look to the examples documented in the RabbitMQ website: 
-
-https://www.rabbitmq.com/client-libraries/amqp-client-libraries
-
-### Creating a connection
-
-A connection to the RabbitMQ AMQP 1.0 server can be established using the Environment object.
-
-For example:
-
-```python
-    environment = Environment("amqp://guest:guest@localhost:5672/")
-    connection = environment.connection()
-    connection.dial()
-```
-
-### Managing resources
-
-Once we have a Connection object we can get a Management object in order to submit to the server management operations
-(es: declare/delete queues and exchanges, purging queues, binding/unbinding objects ecc...)
-
-For example (this code is declaring an exchange and a queue:
-
-```python
-    management = connection.management()
-
-    print("declaring exchange and queue")
-    management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
-
-    management.declare_queue(
-        QuorumQueueSpecification(name=queue_name)
-    )
-```
-
-### Publishing messages
-
-Once we have a Connection object we can get a Publisher object in order to send messages to the server (to an exchange or queue)
-
-For example:
-
-```python
-    addr_queue = AddressHelper.queue_address(queue_name)
-    publisher = connection.publisher(addr)
-
-    # publish messages
-    for i in range(messages_to_publish):
-        publisher.publish(Message(body="test"))
-
-    publisher.close()
-```
-
-### Consuming messages
-
-Once we have a Connection object we can get a Consumer object in order to consumer messages from the server (queue).
-
-Messages are received through a callback
-
-For example:
-
-Create a class which extends AMQPMessagingHandler which defines at minimum the on_consumer method, that will receive the 
-messages consumed:
-
-```python
-class MyMessageHandler(AMQPMessagingHandler):
-
-    def __init__(self):
-        super().__init__()
-        self._count = 0
-
-    def on_message(self, event: Event):
-        print("received message: " + str(event.message.body))
-
-        # accepting
-        self.delivery_context.accept(event)
-```
-
-Then from connection get a consumer object:
-
-```python
-    addr_queue = AddressHelper.queue_address(queue_name)
-    consumer = connection.consumer(addr_queue, handler=MyMessageHandler())
-
-    try:
-        consumer.run()
-    except KeyboardInterrupt:
-        pass
-
-    consumer.close()
-```
-
-The consumer will run indefinitively waiting for messages to arrive.
-
-### Support for streams
-
-The client natively supports streams: https://www.rabbitmq.com/blog/2021/07/13/rabbitmq-streams-overview
-
-You can consume from a given offset or specify a default starting point (FIRST, NEXT, LAST).
-
-Streams filtering is also supported: https://www.rabbitmq.com/blog/2023/10/16/stream-filtering
-
-You can check the [`stream example`](./examples/streams/example_with_streams.py) to see how to work with RabbitMQ streams.
+### Getting Started    
 
-### SSL connections
+Inside the [examples](./examples) folder you can find a set of examples that show how to use the client.
 
-The client supports TLS/SSL connections.
 
-You can check the [`ssl example`](./examples/tls/tls_example.py) to see how to establish a secured connection
+### Documentation
 
-### Oauth authentication
+[Client Guide](https://www.rabbitmq.com/client-libraries/amqp-client-libraries) select the python section.
 
-The client supports oauth2 authentication.
 
-You can check the [`oauth2 example`](./examples/oauth/oaut.py) to see how to establish and refresh a connection using an oauth2 token
+### Build
 
-### Managing disconnections
+- `make rabbitmq-server`: run the RabbitMQ server in a docker container
+- `poetry build`: build the source project
+- `poetry install`: resolves and install dependencies
+- `make test`: run the tests
 
-The client supports automatic reconnection with the ability to reconnect Managements, Producers and Consumers
+Note for MAC users:
+- TLS does not work, see: https://github.com/rabbitmq/rabbitmq-amqp-python-client/issues/64
 
-You can check the [`reconnection example`](./examples/reconnection/reconnection_example.py) to see how to manage disconnections
 
 
 

create_tag.sh

@@ -0,0 +1,12 @@
+#/bin/bash
+version=$1
+# Regex pattern for the valid strings
+regex="^v([0-9]+)\.([0-9]+)\.([0-9]+)(-(alpha|beta|rc)\.([0-9]+))?$"
+
+if [[ $version =~ $regex ]]; then
+    echo "Creating tag with: " $1
+    git tag -a -s -u $2 -m "rabbitmq-amqp-python-client $1" $1 && git push && git push --tags
+    echo "Tag created: " $1
+else
+    echo "Invalid version" $1
+fi

examples/README.md

@@ -4,4 +4,4 @@ Client examples
  - [Reconnection](./reconnection/reconnection_example.py) - Producer and Consumer example with reconnection
  - [TLS](./tls/tls_example.py) - Producer and Consumer using a TLS connection
  - [Streams](./streams/example_with_streams.py) - Example supporting stream capabilities
- - [Oauth](./oauth/oauth.py) - Connection through Oauth token
+ - [Oauth](./oauth/oAuth2.py) - Connection through Oauth token

examples/getting_started/getting_started.py

@@ -5,6 +5,7 @@
     AddressHelper,
     AMQPMessagingHandler,
     Connection,
+    Converter,
     Environment,
     Event,
     ExchangeSpecification,
@@ -24,7 +25,11 @@ def __init__(self):
         self._count = 0
 
     def on_amqp_message(self, event: Event):
-        print("received message: " + str(event.message.body))
+        print(
+            "received message: {} ".format(
+                Converter.bytes_to_string(event.message.body)
+            )
+        )
 
         # accepting
         self.delivery_context.accept(event)
@@ -43,13 +48,11 @@ def on_amqp_message(self, event: Event):
         # in case of rejection with annotations added
         # self.delivery_context.discard_with_annotations(event)
 
-        print("count " + str(self._count))
-
         self._count = self._count + 1
+        print("count " + str(self._count))
 
         if self._count == MESSAGES_TO_PUBLISH:
-            print("closing receiver")
-            # if you want you can add cleanup operations here
+            print("received all messages")
 
     def on_connection_closed(self, event: Event):
         # if you want you can add cleanup operations here
@@ -79,7 +82,6 @@ def create_connection(environment: Environment) -> Connection:
 
 
 def main() -> None:
-
     exchange_name = "test-exchange"
     queue_name = "example-queue"
     routing_key = "routing-key"
@@ -122,8 +124,9 @@ def main() -> None:
 
     # publish 10 messages
     for i in range(MESSAGES_TO_PUBLISH):
-        print("publishing")
-        status = publisher.publish(Message(body="test"))
+        status = publisher.publish(
+            Message(body=Converter.string_to_bytes("test message {} ".format(i)))
+        )
         if status.remote_state == OutcomeState.ACCEPTED:
             print("message accepted")
         elif status.remote_state == OutcomeState.RELEASED:

examples/oauth/oaut.py renamed to examples/oauth/oAuth2.py

@@ -10,6 +10,7 @@
     AddressHelper,
     AMQPMessagingHandler,
     Connection,
+    Converter,
     Environment,
     Event,
     ExchangeSpecification,
@@ -30,7 +31,7 @@ def __init__(self):
         self._count = 0
 
     def on_amqp_message(self, event: Event):
-        print("received message: " + str(event.message.body))
+        print("received message: " + Converter.bytes_to_string(event.message.body))
 
         # accepting
         self.delivery_context.accept(event)
@@ -85,10 +86,9 @@ def create_connection(environment: Environment) -> Connection:
 
 
 def main() -> None:
-
-    exchange_name = "test-exchange"
-    queue_name = "example-queue"
-    routing_key = "routing-key"
+    exchange_name = "oAuth2-test-exchange"
+    queue_name = "oAuth2-example-queue"
+    routing_key = "oAuth2-routing-key"
 
     print("connection to amqp server")
     oaut_token = token(
@@ -144,14 +144,15 @@ def main() -> None:
 
     # publish 10 messages
     for i in range(MESSAGES_TO_PUBLISH):
-        print("publishing")
-        status = publisher.publish(Message(body="test"))
+        status = publisher.publish(
+            Message(body=Converter.string_to_bytes("test_{}".format(i)))
+        )
         if status.remote_state == OutcomeState.ACCEPTED:
-            print("message accepted")
+            print("message: test_{} accepted".format(i))
         elif status.remote_state == OutcomeState.RELEASED:
-            print("message not routed")
+            print("message: test_{} not routed".format(i))
         elif status.remote_state == OutcomeState.REJECTED:
-            print("message not rejected")
+            print("message: test_{} rejected".format(i))
 
     publisher.close()
 

examples/reconnection/reconnection_example.py

@@ -4,6 +4,7 @@
     AMQPMessagingHandler,
     Connection,
     ConnectionClosed,
+    Converter,
     Environment,
     Event,
     ExchangeSpecification,
@@ -15,7 +16,6 @@
 # here we keep track of the objects we need to reconnect
 MESSAGES_TO_PUBLISH = 50000
 
-
 environment = Environment(
     uri="amqp://guest:guest@localhost:5672/",
 )
@@ -29,7 +29,9 @@ def __init__(self):
 
     def on_message(self, event: Event):
         if self._count % 1000 == 0:
-            print("received 100 message: " + str(event.message.body))
+            print(
+                "received 100 message: " + Converter.bytes_to_string(event.message.body)
+            )
 
         # accepting
         self.delivery_context.accept(event)
@@ -79,7 +81,6 @@ def create_connection() -> Connection:
 
 
 def main() -> None:
-
     exchange_name = "test-exchange"
     queue_name = "example-queue"
     routing_key = "routing-key"
@@ -128,7 +129,7 @@ def main() -> None:
                 print("published 1000 messages...")
             try:
                 if publisher is not None:
-                    publisher.publish(Message(body="test"))
+                    publisher.publish(Message(body=Converter.string_to_bytes("test")))
             except ConnectionClosed:
                 print("publisher closing exception, resubmitting")
                 # publisher = connection.publisher(addr)