1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
from qpid.testlib import TestBase
from qpid.content import Content
import qpid.client
class EchoTests(TestBase):
"""Verify that messages can be sent and received retaining fidelity"""
def test_small_message(self):
channel = self.channel
self.queue_declare(queue="q")
channel.tx_select()
consumer = self.consume("q", no_ack=False)
body = self.uniqueString()
channel.basic_publish(
content=Content(body),
routing_key="q")
channel.tx_commit()
msg = consumer.get(timeout=1)
channel.basic_ack(delivery_tag=msg.delivery_tag)
channel.tx_commit()
self.assertEqual(body, msg.content.body)
def test_large_message(self):
channel = self.channel
self.queue_declare(queue="q")
channel.tx_select()
consumer = self.consume("q", no_ack=False)
# This is default maximum frame size supported by the Java Broker. Python
# currently does not support framing of oversized messages in multiple frames.
body = self.randomLongString()
channel.basic_publish(
content=Content(body),
routing_key="q")
channel.tx_commit()
msg = consumer.get(timeout=1)
channel.basic_ack(delivery_tag=msg.delivery_tag)
channel.tx_commit()
self.assertEqual(len(body), len(msg.content.body))
self.assertEqual(body, msg.content.body)
def test_large_message_received_in_many_content_frames(self):
channel = self.channel
queue_name = "q"
self.queue_declare(queue=queue_name)
channel.tx_select()
body = self.randomLongString()
channel.basic_publish(
content=Content(body),
routing_key=queue_name)
channel.tx_commit()
consuming_client = None
try:
# Create a second connection with minimum framesize. The Broker will then be forced to chunk
# the content in order to send it to us.
consuming_client = qpid.client.Client(self.config.broker.host, self.config.broker.port)
tune_params = { "channel_max" : 256, "frame_max" : 4096 }
consuming_client.start(username = self.config.broker.user, password = self.config.broker.password, tune_params = tune_params)
consuming_channel = consuming_client.channel(1)
consuming_channel.channel_open()
consuming_channel.tx_select()
consumer_reply = consuming_channel.basic_consume(queue=queue_name, no_ack=False)
consumer = consuming_client.queue(consumer_reply.consumer_tag)
msg = consumer.get(timeout=1)
consuming_channel.basic_ack(delivery_tag=msg.delivery_tag)
consuming_channel.tx_commit()
self.assertEqual(len(body), len(msg.content.body))
self.assertEqual(body, msg.content.body)
finally:
if consuming_client:
consuming_client.close()
def test_commit_ok_possibly_interleaved_with_message_delivery(self):
"""This test exposes an defect on the Java Broker (QPID-6094). The Java Client
can contravene the AMQP spec by sending other frames between the message header/frames.
As this is a long standing defect in the Java Broker, QPID-6082 changed
the Python client to allow it to tolerate such illegal interleaving.
"""
channel = self.channel
queue_name = "q"
self.queue_declare(queue=queue_name)
count = 25
channel.basic_qos(prefetch_count=count)
channel.tx_select()
bodies = []
for i in range(count):
body = self.randomLongString()
bodies.append(body)
channel.basic_publish(
content=Content(bodies[i]),
routing_key=queue_name)
channel.tx_commit()
# Start consuming. Prefetch will mean the Broker will start to send us
# all the messages accumulating them in the client.
consumer = self.consume("q", no_ack=False)
# Get and ack/commit the first message
msg = consumer.get(timeout=1)
channel.basic_ack(delivery_tag=msg.delivery_tag)
channel.tx_commit()
# In the problematic case, the Broker interleaves our commit-ok response amongst the content
# frames of message. QPID-6082 means the Python client now tolerates this
# problem and all messages should arrive correctly.
expectedBody = bodies[0]
self.assertEqual(len(expectedBody), len(msg.content.body))
self.assertEqual(expectedBody, msg.content.body)
for i in range(1, len(bodies)):
msg = consumer.get(timeout=5)
expectedBody = bodies[i]
self.assertEqual(len(expectedBody), len(msg.content.body))
self.assertEqual(expectedBody, msg.content.body)
|
