Skip to content

Commit 6f3ee6a

Browse files
author
DanielePalaia
committed
improvements
1 parent 658af34 commit 6f3ee6a

File tree

6 files changed

+174
-23
lines changed

6 files changed

+174
-23
lines changed

rabbitmq_amqp_python_client/amqp_message.py

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
from typing import Optional, Union, cast
1+
from typing import Union, cast
22
from uuid import UUID
33

44
from proton._data import Described
@@ -16,13 +16,14 @@ def __init__( # type: ignore
1616
**kwargs,
1717
):
1818
super().__init__(body=body, **kwargs)
19-
self._addr: Optional[str] = None
19+
self._addr: str = ""
2020
self._native_message = None
2121

2222
def to_address(self, addr: str) -> None:
23+
self.address = addr
2324
self._addr = addr
2425

25-
def address(self) -> Optional[str]:
26+
def get_address(self) -> str:
2627
return self._addr
2728

2829
def native_message(self) -> Message:

rabbitmq_amqp_python_client/connection.py

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -82,11 +82,12 @@ def close(self) -> None:
8282
self._conn.close()
8383
self._connections.remove(self)
8484

85-
def publisher(self, destination: str) -> Publisher:
86-
if validate_address(destination) is False:
87-
raise ArgumentOutOfRangeException(
88-
"destination address must start with /queues or /exchanges"
89-
)
85+
def publisher(self, destination: str = "") -> Publisher:
86+
if destination != "":
87+
if validate_address(destination) is False:
88+
raise ArgumentOutOfRangeException(
89+
"destination address must start with /queues or /exchanges"
90+
)
9091
publisher = Publisher(self._conn, destination)
9192
return publisher
9293

rabbitmq_amqp_python_client/publisher.py

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,9 @@
11
import logging
22
from typing import Optional
33

4+
from .address_helper import validate_address
45
from .amqp_message import AmqpMessage
6+
from .exceptions import ArgumentOutOfRangeException
57
from .options import SenderOptionUnseattle
68
from .qpid.proton._delivery import Delivery
79
from .qpid.proton.utils import (
@@ -25,8 +27,18 @@ def _open(self) -> None:
2527
self._sender = self._create_sender(self._addr)
2628

2729
def publish(self, message: AmqpMessage) -> Delivery:
28-
if self._sender is not None:
29-
return self._sender.send(message.native_message())
30+
if self._addr != "":
31+
if self._sender is not None:
32+
return self._sender.send(message.native_message())
33+
else:
34+
if message.get_address() != "":
35+
if validate_address(message.get_address()) is False:
36+
raise ArgumentOutOfRangeException(
37+
"destination address must start with /queues or /exchanges"
38+
)
39+
if self._sender is not None:
40+
delivery = self._sender.send(message.native_message())
41+
return delivery
3042

3143
def close(self) -> None:
3244
logger.debug("Closing Sender")

rabbitmq_amqp_python_client/qpid/proton/_message.py

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -140,7 +140,6 @@ def _check(self, err: int) -> int:
140140
def _check_property_keys(self) -> None:
141141
"""
142142
AMQP allows only string keys for properties. This function checks that this requirement is met
143-
and raises a MessageException if not. However, in certain cases, conversions to string are
144143
automatically performed:
145144
146145
1. When a key is a user-defined (non-AMQP) subclass of str.

tests/test_publisher.py

Lines changed: 121 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,13 @@
99
ConnectionClosed,
1010
Environment,
1111
ExchangeSpecification,
12+
OutcomeState,
1213
QuorumQueueSpecification,
1314
StreamSpecification,
1415
)
1516

1617
from .http_requests import delete_all_connections
18+
from .utils import publish_per_message, create_binding
1719

1820

1921
def test_publish_queue(connection: Connection) -> None:
@@ -31,18 +33,112 @@ def test_publish_queue(connection: Connection) -> None:
3133
try:
3234
publisher = connection.publisher("/queues/" + queue_name)
3335
status = publisher.publish(AmqpMessage(body="test"))
34-
if status.ACCEPTED:
36+
if status.remote_state == OutcomeState.ACCEPTED:
37+
accepted = True
38+
except Exception:
39+
raised = True
40+
41+
if publisher is not None:
42+
publisher.close()
43+
44+
management.delete_queue(queue_name)
45+
management.close()
46+
47+
assert accepted is True
48+
assert raised is False
49+
50+
51+
def test_publish_per_message(connection: Connection) -> None:
52+
53+
queue_name = "test-queue-1"
54+
queue_name_2 = "test-queue-2"
55+
management = connection.management()
56+
57+
management.declare_queue(QuorumQueueSpecification(name=queue_name))
58+
management.declare_queue(QuorumQueueSpecification(name=queue_name_2))
59+
60+
raised = False
61+
62+
publisher = None
63+
accepted = False
64+
accepted_2 = True
65+
66+
try:
67+
publisher = connection.publisher()
68+
status = publish_per_message(
69+
publisher, addr=AddressHelper.queue_address(queue_name)
70+
)
71+
if status.remote_state == OutcomeState.ACCEPTED:
72+
accepted = True
73+
status = publish_per_message(
74+
publisher, addr=AddressHelper.queue_address(queue_name_2)
75+
)
76+
if status.remote_state == OutcomeState.ACCEPTED:
77+
accepted_2 = True
78+
except Exception:
79+
raised = True
80+
81+
if publisher is not None:
82+
publisher.close()
83+
84+
purged_messages_queue_1 = management.purge_queue(queue_name)
85+
purged_messages_queue_2 = management.purge_queue(queue_name_2)
86+
management.delete_queue(queue_name)
87+
management.delete_queue(queue_name_2)
88+
management.close()
89+
90+
assert accepted is True
91+
assert accepted_2 is True
92+
assert purged_messages_queue_1 == 1
93+
assert purged_messages_queue_2 == 1
94+
assert raised is False
95+
96+
97+
def test_publish_per_message_to_exchange(connection: Connection) -> None:
98+
99+
exchange_name = "test-bind-exchange-to-queue-exchange"
100+
queue_name = "test-bind-exchange-to-queue-queue"
101+
routing_key = "routing-key"
102+
103+
management = connection.management()
104+
105+
binding_exchange_queue_path = create_binding(
106+
management, exchange_name, queue_name, routing_key
107+
)
108+
109+
raised = False
110+
111+
publisher = None
112+
accepted = False
113+
accepted_2 = True
114+
115+
try:
116+
publisher = connection.publisher()
117+
status = publish_per_message(
118+
publisher, addr=AddressHelper.exchange_address(exchange_name, routing_key)
119+
)
120+
if status.remote_state == OutcomeState.ACCEPTED:
35121
accepted = True
122+
status = publish_per_message(
123+
publisher, addr=AddressHelper.queue_address(queue_name)
124+
)
125+
if status.remote_state == OutcomeState.ACCEPTED:
126+
accepted_2 = True
36127
except Exception:
37128
raised = True
38129

39130
if publisher is not None:
40131
publisher.close()
41132

133+
purged_messages_queue = management.purge_queue(queue_name)
134+
management.unbind(binding_exchange_queue_path)
135+
management.delete_exchange(exchange_name)
42136
management.delete_queue(queue_name)
43137
management.close()
44138

45139
assert accepted is True
140+
assert accepted_2 is True
141+
assert purged_messages_queue == 2
46142
assert raised is False
47143

48144

@@ -90,24 +186,37 @@ def test_publish_to_invalid_destination(connection: Connection) -> None:
90186
assert raised is True
91187

92188

189+
def test_publish_per_message_to_invalid_destination(connection: Connection) -> None:
190+
191+
queue_name = "test-queue-1"
192+
raised = False
193+
194+
publisher = None
195+
message = AmqpMessage(body="test")
196+
message.to_address("/invalid_destination/" + queue_name)
197+
198+
try:
199+
publisher = connection.publisher()
200+
publisher.publish(message)
201+
except ArgumentOutOfRangeException:
202+
raised = True
203+
except Exception:
204+
raised = False
205+
206+
if publisher is not None:
207+
publisher.close()
208+
209+
assert raised is True
210+
211+
93212
def test_publish_exchange(connection: Connection) -> None:
94213

95214
exchange_name = "test-exchange"
96215
queue_name = "test-queue"
97216
management = connection.management()
98217
routing_key = "routing-key"
99218

100-
management.declare_exchange(ExchangeSpecification(name=exchange_name))
101-
102-
management.declare_queue(QuorumQueueSpecification(name=queue_name))
103-
104-
management.bind(
105-
BindingSpecification(
106-
source_exchange=exchange_name,
107-
destination_queue=queue_name,
108-
binding_key=routing_key,
109-
)
110-
)
219+
create_binding(management, exchange_name, queue_name, routing_key)
111220

112221
addr = AddressHelper.exchange_address(exchange_name, routing_key)
113222

tests/utils.py

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,13 @@
44
AmqpMessage,
55
BindingSpecification,
66
Connection,
7+
Delivery,
78
ExchangeSpecification,
89
ExchangeType,
910
Management,
11+
Publisher,
1012
QuorumQueueSpecification,
13+
AddressHelper,
1114
)
1215

1316

@@ -29,6 +32,13 @@ def publish_messages(
2932
publisher.close()
3033

3134

35+
def publish_per_message(publisher: Publisher, addr: str) -> Delivery:
36+
message = AmqpMessage(body="test")
37+
message.to_address(addr)
38+
status = publisher.publish(message)
39+
return status
40+
41+
3242
def setup_dead_lettering(management: Management) -> str:
3343

3444
exchange_dead_lettering = "exchange-dead-letter"
@@ -55,6 +65,25 @@ def setup_dead_lettering(management: Management) -> str:
5565
return bind_path
5666

5767

68+
def create_binding(
69+
management: Management, exchange_name: str, queue_name: str, routing_key: str
70+
) -> str:
71+
72+
management.declare_exchange(ExchangeSpecification(name=exchange_name))
73+
74+
management.declare_queue(QuorumQueueSpecification(name=queue_name))
75+
76+
bind_name = management.bind(
77+
BindingSpecification(
78+
source_exchange=exchange_name,
79+
destination_queue=queue_name,
80+
binding_key=routing_key,
81+
)
82+
)
83+
84+
return bind_name
85+
86+
5887
def cleanup_dead_lettering(management: Management, bind_path: str) -> None:
5988

6089
exchange_dead_lettering = "exchange-dead-letter"

0 commit comments

Comments
 (0)