Skip to content

Commit 07a2617

Browse files
author
DanielePalaia
committed
improve examples / updating docs / adding credits to consumer
1 parent d08139f commit 07a2617

File tree

8 files changed

+239
-38
lines changed

8 files changed

+239
-38
lines changed

‎README.md

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ The client is distributed via [`PIP`](https://pypi.org/project/rabbitmq-amqp-pyt
1818

1919
## Getting Started
2020

21-
An example is provided in ./getting_started_main.py you can run it after starting a RabbitMQ 4.0 broker with:
21+
An example is provided [`here`](./examples/getting_started/basic_example.py) you can run it after starting a RabbitMQ 4.0 broker with:
2222

2323
poetry run python ./examples/getting_started/main.py
2424

@@ -109,6 +109,33 @@ Then from connection get a consumer object:
109109

110110
The consumer will run indefinitively waiting for messages to arrive.
111111

112+
### Support for streams
113+
114+
The client natively supports streams: https://www.rabbitmq.com/blog/2021/07/13/rabbitmq-streams-overview
115+
116+
You can consume from a given offset or specify a default starting point (FIRST, NEXT, LAST).
117+
118+
Streams filtering is also supported: https://www.rabbitmq.com/blog/2023/10/16/stream-filtering
119+
120+
You can check the [`stream example`](./examples/getting_started/example_with_streams.py) to see how to work with RabbitMQ streams.
121+
122+
### SSL
123+
124+
The client supports TLS/SSL connections.
125+
126+
You can check the [`ssl example`](./examples/getting_started/tls_example.py) to see how to establish a secured connection
127+
128+
129+
### Managing disconnection
130+
131+
At this stage the client doesn't support auto-reconnect but a callback is invoked everytime a remote disconnection is detected.
132+
You can use this callback to implement your own logic and eventually attempt a reconnection.
133+
134+
You can check the [`reconnection example`](./examples/getting_started/reconnection_example.py) to see how to manage disconnections and
135+
eventually attempt a reconnection
136+
137+
138+
112139

113140

114141

‎examples/getting_started/basic_example.py

Lines changed: 8 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -6,13 +6,14 @@
66
AMQPMessagingHandler,
77
BindingSpecification,
88
Connection,
9+
Disposition,
910
Event,
1011
ExchangeSpecification,
1112
Message,
1213
QuorumQueueSpecification,
1314
)
1415

15-
messages_to_publish = 100
16+
MESSAGES_TO_PUBLISH = 100
1617

1718

1819
class MyMessageHandler(AMQPMessagingHandler):
@@ -45,7 +46,7 @@ def on_message(self, event: Event):
4546

4647
self._count = self._count + 1
4748

48-
if self._count == messages_to_publish:
49+
if self._count == MESSAGES_TO_PUBLISH:
4950
print("closing receiver")
5051
# if you want you can add cleanup operations here
5152
# event.receiver.close()
@@ -62,17 +63,6 @@ def on_link_closed(self, event: Event) -> None:
6263

6364
def create_connection() -> Connection:
6465
connection = Connection("amqp://guest:guest@localhost:5672/")
65-
# in case of SSL enablement
66-
# ca_cert_file = ".ci/certs/ca_certificate.pem"
67-
# client_cert = ".ci/certs/client_certificate.pem"
68-
# client_key = ".ci/certs/client_key.pem"
69-
# connection = Connection(
70-
# "amqps://guest:guest@localhost:5671/",
71-
# ssl_context=SslConfigurationContext(
72-
# ca_cert=ca_cert_file,
73-
# client_cert=ClientCert(client_cert=client_cert, client_key=client_key),
74-
# ),
75-
# )
7666
connection.dial()
7767

7868
return connection
@@ -120,13 +110,14 @@ def main() -> None:
120110
# management.close()
121111

122112
# publish 10 messages
123-
for i in range(messages_to_publish):
113+
for i in range(MESSAGES_TO_PUBLISH):
114+
print("publishing")
124115
status = publisher.publish(Message(body="test"))
125-
if status.ACCEPTED:
116+
if status.remote_state == Disposition.ACCEPTED:
126117
print("message accepted")
127-
elif status.RELEASED:
118+
elif status.remote_state == Disposition.RELEASED:
128119
print("message not routed")
129-
elif status.REJECTED:
120+
elif status.remote_state == Disposition.REJECTED:
130121
print("message not rejected")
131122

132123
publisher.close()

‎examples/getting_started/example_with_streams.py

Lines changed: 21 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111
StreamSpecification,
1212
)
1313

14+
MESSAGES_TO_PUBLISH = 100
15+
1416

1517
class MyMessageHandler(AMQPMessagingHandler):
1618

@@ -19,6 +21,7 @@ def __init__(self):
1921
self._count = 0
2022

2123
def on_message(self, event: Event):
24+
# just messages with banana filters get received
2225
print(
2326
"received message from stream: "
2427
+ str(event.message.body)
@@ -47,7 +50,7 @@ def on_message(self, event: Event):
4750

4851
self._count = self._count + 1
4952

50-
if self._count == 100:
53+
if self._count == MESSAGES_TO_PUBLISH:
5154
print("closing receiver")
5255
# if you want you can add cleanup operations here
5356
# event.receiver.close()
@@ -64,25 +67,13 @@ def on_link_closed(self, event: Event) -> None:
6467

6568
def create_connection() -> Connection:
6669
connection = Connection("amqp://guest:guest@localhost:5672/")
67-
# in case of SSL enablement
68-
# ca_cert_file = ".ci/certs/ca_certificate.pem"
69-
# client_cert = ".ci/certs/client_certificate.pem"
70-
# client_key = ".ci/certs/client_key.pem"
71-
# connection = Connection(
72-
# "amqps://guest:guest@localhost:5671/",
73-
# ssl_context=SslConfigurationContext(
74-
# ca_cert=ca_cert_file,
75-
# client_cert=ClientCert(client_cert=client_cert, client_key=client_key),
76-
# ),
77-
# )
7870
connection.dial()
7971

8072
return connection
8173

8274

8375
def main() -> None:
8476
queue_name = "example-queue"
85-
messages_to_publish = 100
8677

8778
print("connection to amqp server")
8879
connection = create_connection()
@@ -99,6 +90,7 @@ def main() -> None:
9990
# can be first, last, next or an offset long
10091
# you can also specify stream filters with methods: apply_filters and filter_match_unfiltered
10192
stream_filter_options.offset(OffsetSpecification.first)
93+
stream_filter_options.apply_filters(["banana"])
10294

10395
consumer = consumer_connection.consumer(
10496
addr_queue,
@@ -112,8 +104,22 @@ def main() -> None:
112104
# print("create a publisher and publish a test message")
113105
publisher = connection.publisher(addr_queue)
114106

115-
for i in range(messages_to_publish):
116-
publisher.publish(Message(body="test: " + str(i)))
107+
# publish with a filter of apple
108+
for i in range(MESSAGES_TO_PUBLISH):
109+
publisher.publish(
110+
Message(
111+
body="apple: " + str(i), annotations={"x-stream-filter-value": "apple"}
112+
)
113+
)
114+
115+
# publish with a filter of banana
116+
for i in range(MESSAGES_TO_PUBLISH):
117+
publisher.publish(
118+
Message(
119+
body="banana: " + str(i),
120+
annotations={"x-stream-filter-value": "banana"},
121+
)
122+
)
117123

118124
publisher.close()
119125

‎examples/getting_started/reconnection_example.py

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ class ConnectionConfiguration:
3131

3232

3333
connection_configuration = ConnectionConfiguration()
34-
messages_to_publish = 50000
34+
MESSAGES_TO_PUBLSH = 50000
3535

3636

3737
# disconnection callback
@@ -95,7 +95,7 @@ def on_message(self, event: Event):
9595

9696
self._count = self._count + 1
9797

98-
if self._count == messages_to_publish:
98+
if self._count == MESSAGES_TO_PUBLSH:
9999
print("closing receiver")
100100
# if you want you can add cleanup operations here
101101
# event.receiver.close()
@@ -181,7 +181,7 @@ def main() -> None:
181181

182182
# publishing messages
183183
while True:
184-
for i in range(messages_to_publish):
184+
for i in range(MESSAGES_TO_PUBLSH):
185185

186186
if i % 1000 == 0:
187187
print("published 1000 messages...")

‎examples/getting_started/tls_example.py

Lines changed: 168 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,168 @@
1+
# type: ignore
2+
3+
4+
from rabbitmq_amqp_python_client import ( # SSlConfigurationContext,; SslConfigurationContext,; ClientCert,
5+
AddressHelper,
6+
AMQPMessagingHandler,
7+
BindingSpecification,
8+
ClientCert,
9+
Connection,
10+
Event,
11+
ExchangeSpecification,
12+
Message,
13+
QuorumQueueSpecification,
14+
SslConfigurationContext,
15+
)
16+
17+
messages_to_publish = 100
18+
19+
20+
class MyMessageHandler(AMQPMessagingHandler):
21+
22+
def __init__(self):
23+
super().__init__()
24+
self._count = 0
25+
26+
def on_message(self, event: Event):
27+
print("received message: " + str(event.message.body))
28+
29+
# accepting
30+
self.delivery_context.accept(event)
31+
32+
# in case of rejection (+eventually deadlettering)
33+
# self.delivery_context.discard(event)
34+
35+
# in case of requeuing
36+
# self.delivery_context.requeue(event)
37+
38+
# annotations = {}
39+
# annotations[symbol('x-opt-string')] = 'x-test1'
40+
# in case of requeuing with annotations added
41+
# self.delivery_context.requeue_with_annotations(event, annotations)
42+
43+
# in case of rejection with annotations added
44+
# self.delivery_context.discard_with_annotations(event)
45+
46+
print("count " + str(self._count))
47+
48+
self._count = self._count + 1
49+
50+
if self._count == messages_to_publish:
51+
print("closing receiver")
52+
# if you want you can add cleanup operations here
53+
# event.receiver.close()
54+
# event.connection.close()
55+
56+
def on_connection_closed(self, event: Event):
57+
# if you want you can add cleanup operations here
58+
print("connection closed")
59+
60+
def on_link_closed(self, event: Event) -> None:
61+
# if you want you can add cleanup operations here
62+
print("link closed")
63+
64+
65+
def create_connection() -> Connection:
66+
# in case of SSL enablement
67+
ca_cert_file = ".ci/certs/ca_certificate.pem"
68+
client_cert = ".ci/certs/client_certificate.pem"
69+
client_key = ".ci/certs/client_key.pem"
70+
connection = Connection(
71+
"amqps://guest:guest@localhost:5671/",
72+
ssl_context=SslConfigurationContext(
73+
ca_cert=ca_cert_file,
74+
client_cert=ClientCert(client_cert=client_cert, client_key=client_key),
75+
),
76+
)
77+
connection.dial()
78+
79+
return connection
80+
81+
82+
def main() -> None:
83+
84+
exchange_name = "test-exchange"
85+
queue_name = "example-queue"
86+
routing_key = "routing-key"
87+
88+
print("connection to amqp server")
89+
connection = create_connection()
90+
91+
management = connection.management()
92+
93+
print("declaring exchange and queue")
94+
management.declare_exchange(ExchangeSpecification(name=exchange_name, arguments={}))
95+
96+
management.declare_queue(
97+
QuorumQueueSpecification(name=queue_name)
98+
# QuorumQueueSpecification(name=queue_name, dead_letter_exchange="dead-letter")
99+
)
100+
101+
print("binding queue to exchange")
102+
bind_name = management.bind(
103+
BindingSpecification(
104+
source_exchange=exchange_name,
105+
destination_queue=queue_name,
106+
binding_key=routing_key,
107+
)
108+
)
109+
110+
addr = AddressHelper.exchange_address(exchange_name, routing_key)
111+
112+
addr_queue = AddressHelper.queue_address(queue_name)
113+
114+
print("create a publisher and publish a test message")
115+
publisher = connection.publisher(addr)
116+
117+
print("purging the queue")
118+
messages_purged = management.purge_queue(queue_name)
119+
120+
print("messages purged: " + str(messages_purged))
121+
# management.close()
122+
123+
# publish 10 messages
124+
for i in range(messages_to_publish):
125+
status = publisher.publish(Message(body="test"))
126+
if status.ACCEPTED:
127+
print("message accepted")
128+
elif status.RELEASED:
129+
print("message not routed")
130+
elif status.REJECTED:
131+
print("message not rejected")
132+
133+
publisher.close()
134+
135+
print(
136+
"create a consumer and consume the test message - press control + c to terminate to consume"
137+
)
138+
consumer = connection.consumer(addr_queue, handler=MyMessageHandler())
139+
140+
try:
141+
consumer.run()
142+
except KeyboardInterrupt:
143+
pass
144+
145+
print("cleanup")
146+
consumer.close()
147+
# once we finish consuming if we close the connection we need to create a new one
148+
# connection = create_connection()
149+
# management = connection.management()
150+
151+
print("unbind")
152+
management.unbind(bind_name)
153+
154+
print("delete queue")
155+
management.delete_queue(queue_name)
156+
157+
print("delete exchange")
158+
management.delete_exchange(exchange_name)
159+
160+
print("closing connections")
161+
management.close()
162+
print("after management closing")
163+
connection.close()
164+
print("after connection closing")
165+
166+
167+
if __name__ == "__main__":
168+
main()

0 commit comments

Comments
 (0)