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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
|
#-------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
#--------------------------------------------------------------------------
# pylint: disable=super-init-not-called,arguments-differ
from uamqp import c_uamqp, compat, utils
class AMQPType(object):
"""Base type for specific AMQP encoded type definitions.
:ivar value: The Python value of the AMQP type.
:ivar c_data: The C AMQP encoded object.
"""
def __init__(self, value):
self._c_type = self._c_wrapper(value)
@property
def value(self):
return self._c_type.value
@property
def c_data(self):
return self._c_type
def _c_wrapper(self, value):
raise NotImplementedError()
class AMQPSymbol(AMQPType):
"""An AMQP symbol object.
:ivar value: The Python value of the AMQP type.
:vartype value: bytes
:ivar c_data: The C AMQP encoded object.
:vartype c_data: c_uamqp.SymbolValue
:param value: The value to encode as an AMQP symbol.
:type value: bytes or str
:param encoding: The encoding to be used if a str is provided.
The default is 'UTF-8'.
:type encoding: str
"""
def __init__(self, value, encoding='UTF-8'):
self._c_type = self._c_wrapper(value, encoding)
def _c_wrapper(self, value, encoding='UTF-8'):
value = value.encode(encoding) if isinstance(value, str) else value
return c_uamqp.symbol_value(value)
class AMQPChar(AMQPType):
"""An AMQP char object.
:ivar value: The Python value of the AMQP type.
:vartype value: bytes
:ivar c_data: The C AMQP encoded object.
:vartype c_data: c_uamqp.CharValue
:param value: The value to encode as an AMQP char.
:type value: bytes or str
:param encoding: The encoding to be used if a str is provided.
The default is 'UTF-8'.
:type encoding: str
"""
def __init__(self, value, encoding='UTF-8'):
self._c_type = self._c_wrapper(value, encoding)
def _c_wrapper(self, value, encoding='UTF-8'):
if len(value) > 1:
raise ValueError("Value must be a single character.")
value = value.encode(encoding) if isinstance(value, str) else value
return c_uamqp.char_value(value)
class AMQPLong(AMQPType):
"""An AMQP long object.
:ivar value: The Python value of the AMQP type.
:vartype value: int
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.LongValue
:param value: The value to encode as an AMQP ulong.
:type value: int
:raises: ValueError if value is not within allowed range.
"""
def _c_wrapper(self, value):
try:
return c_uamqp.long_value(compat.long(value))
except TypeError:
raise ValueError("Value must be an integer")
except OverflowError:
raise ValueError("Value {} is too large for a Long value.".format(value))
class AMQPuLong(AMQPType):
"""An AMQP unsigned long object.
:ivar value: The Python value of the AMQP uLong.
:vartype value: int
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.ULongValue
:param value: The value to encode as an AMQP unsigned Long.
:type value: list
:raises: ValueError if value is not within allowed range.
"""
def _c_wrapper(self, value):
try:
return c_uamqp.ulong_value(compat.long(value))
except TypeError:
raise ValueError("Value must be an integer")
except OverflowError:
raise ValueError("Value {} is too large for an unsigned Long value.".format(value))
class AMQPByte(AMQPType):
"""An AMQP byte object.
:ivar value: The Python value of the AMQP type.
:vartype value: int
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.ByteValue
:param value: The value to encode as an AMQP ulong.
:type value: int
:raises: ValueError if value is not within allowed range.
"""
def _c_wrapper(self, value):
try:
return c_uamqp.byte_value(int(value))
except TypeError:
raise ValueError("Value must be an integer")
except OverflowError:
raise ValueError("Value {} is too large for a Byte value.".format(value))
class AMQPuByte(AMQPType):
"""An AMQP unsigned byte object.
:ivar value: The Python value of the AMQP uByte.
:vartype value: int
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.UByteValue
:param value: The value to encode as an AMQP unsigned Byte.
:type value: list
:raises: ValueError if value is not within allowed range.
"""
def _c_wrapper(self, value):
try:
return c_uamqp.ubyte_value(int(value))
except TypeError:
raise ValueError("Value must be an integer")
except OverflowError:
raise ValueError("Value {} is too large for an unsigned Byte value.".format(value))
class AMQPInt(AMQPType):
"""An AMQP int object.
:ivar value: The Python value of the AMQP type.
:vartype value: int
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.IntValue
:param value: The value to encode as an AMQP int.
:type value: int
:raises: ValueError if value is not within allowed range.
"""
def _c_wrapper(self, value):
try:
return c_uamqp.int_value(int(value))
except TypeError:
raise ValueError("Value must be an integer")
except OverflowError:
raise ValueError("Value {} is too large for an Int value.".format(value))
class AMQPuInt(AMQPType):
"""An AMQP unsigned int object.
:ivar value: The Python value of the AMQP uInt.
:vartype value: int
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.UIntValue
:param value: The value to encode as an AMQP unsigned int.
:type value: list
:raises: ValueError if value is not within allowed range.
"""
def _c_wrapper(self, value):
try:
return c_uamqp.uint_value(int(value))
except TypeError:
raise ValueError("Value must be an integer")
except OverflowError:
raise ValueError("Value {} is too large for an unsigned int value.".format(value))
class AMQPShort(AMQPType):
"""An AMQP short object.
:ivar value: The Python value of the AMQP type.
:vartype value: int
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.ShortValue
:param value: The value to encode as an AMQP short.
:type value: int
:raises: ValueError if value is not within allowed range.
"""
def _c_wrapper(self, value):
try:
return c_uamqp.short_value(int(value))
except TypeError:
raise ValueError("Value must be an integer")
except OverflowError:
raise ValueError("Value {} is too large for a short value.".format(value))
class AMQPuShort(AMQPType):
"""An AMQP unsigned short object.
:ivar value: The Python value of the AMQP uInt.
:vartype value: int
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.UShortValue
:param value: The value to encode as an AMQP unsigned short.
:type value: int
:raises: ValueError if value is not within allowed range.
"""
def _c_wrapper(self, value):
try:
return c_uamqp.ushort_value(int(value))
except TypeError:
raise ValueError("Value must be an integer")
except OverflowError:
raise ValueError("Value {} is too large for an unsigned short value.".format(value))
class AMQPArray(AMQPType):
"""An AMQP Array object. All the values in the array
must be of the same type.
:ivar value: The Python values of the AMQP array.
:vartype value: list
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.ArrayValue
:param value: The value to encode as an AMQP array.
:type value: int
:raises: ValueError if all values are not the same type.
"""
def _c_wrapper(self, value_array):
if value_array:
value_type = type(value_array[0])
if not all(isinstance(x, value_type) for x in value_array):
raise ValueError("All Array values must be the same type.")
c_array = c_uamqp.array_value()
for value in value_array:
c_array.append(utils.data_factory(value))
return c_array
class AMQPDescribed(AMQPType):
"""An AMQP Described object. All the values in the array
must be of the same type.
:ivar value: The Python values of the AMQP array.
:vartype value: list
:ivar c_data: The C AMQP encoded object.
:vartype c_data: uamqp.c_uamqp.ArrayValue
:param value: The value to encode as an AMQP array.
:type value: int
:raises: ValueError if all values are not the same type.
"""
def __init__(self, descriptor, described):
self._c_type = self._c_wrapper(descriptor, described)
def _c_wrapper(self, descriptor, described):
descriptor = utils.data_factory(descriptor)
described = utils.data_factory(described)
return c_uamqp.described_value(descriptor, described)
|
