| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578 |
- """Convert between bytestreams and higher-level AMQP types.
- 2007-11-05 Barry Pederson <bp@barryp.org>
- """
- # Copyright (C) 2007 Barry Pederson <bp@barryp.org>
- from __future__ import absolute_import, unicode_literals
- import calendar
- import sys
- from datetime import datetime
- from decimal import Decimal
- from io import BytesIO
- from .exceptions import FrameSyntaxError
- from .five import int_types, items, long_t, string, string_t
- from .platform import pack, unpack_from
- from .spec import Basic
- from .utils import bytes_to_str as pstr_t
- from .utils import str_to_bytes
- ftype_t = chr if sys.version_info[0] == 3 else None
- ILLEGAL_TABLE_TYPE = """\
- Table type {0!r} not handled by amqp.
- """
- ILLEGAL_TABLE_TYPE_WITH_KEY = """\
- Table type {0!r} for key {1!r} not handled by amqp. [value: {2!r}]
- """
- ILLEGAL_TABLE_TYPE_WITH_VALUE = """\
- Table type {0!r} not handled by amqp. [value: {1!r}]
- """
- def _read_item(buf, offset=0, unpack_from=unpack_from, ftype_t=ftype_t):
- ftype = ftype_t(buf[offset]) if ftype_t else buf[offset]
- offset += 1
- # 'S': long string
- if ftype == 'S':
- slen, = unpack_from('>I', buf, offset)
- offset += 4
- val = pstr_t(buf[offset:offset + slen])
- offset += slen
- # 's': short string
- elif ftype == 's':
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- val = pstr_t(buf[offset:offset + slen])
- offset += slen
- # 'x': Bytes Array
- elif ftype == 'x':
- blen, = unpack_from('>I', buf, offset)
- offset += 4
- val = buf[offset:offset + blen]
- offset += blen
- # 'b': short-short int
- elif ftype == 'b':
- val, = unpack_from('>B', buf, offset)
- offset += 1
- # 'B': short-short unsigned int
- elif ftype == 'B':
- val, = unpack_from('>b', buf, offset)
- offset += 1
- # 'U': short int
- elif ftype == 'U':
- val, = unpack_from('>h', buf, offset)
- offset += 2
- # 'u': short unsigned int
- elif ftype == 'u':
- val, = unpack_from('>H', buf, offset)
- offset += 2
- # 'I': long int
- elif ftype == 'I':
- val, = unpack_from('>i', buf, offset)
- offset += 4
- # 'i': long unsigned int
- elif ftype == 'i':
- val, = unpack_from('>I', buf, offset)
- offset += 4
- # 'L': long long int
- elif ftype == 'L':
- val, = unpack_from('>q', buf, offset)
- offset += 8
- # 'l': long long unsigned int
- elif ftype == 'l':
- val, = unpack_from('>Q', buf, offset)
- offset += 8
- # 'f': float
- elif ftype == 'f':
- val, = unpack_from('>f', buf, offset)
- offset += 4
- # 'd': double
- elif ftype == 'd':
- val, = unpack_from('>d', buf, offset)
- offset += 8
- # 'D': decimal
- elif ftype == 'D':
- d, = unpack_from('>B', buf, offset)
- offset += 1
- n, = unpack_from('>i', buf, offset)
- offset += 4
- val = Decimal(n) / Decimal(10 ** d)
- # 'F': table
- elif ftype == 'F':
- tlen, = unpack_from('>I', buf, offset)
- offset += 4
- limit = offset + tlen
- val = {}
- while offset < limit:
- keylen, = unpack_from('>B', buf, offset)
- offset += 1
- key = pstr_t(buf[offset:offset + keylen])
- offset += keylen
- val[key], offset = _read_item(buf, offset)
- # 'A': array
- elif ftype == 'A':
- alen, = unpack_from('>I', buf, offset)
- offset += 4
- limit = offset + alen
- val = []
- while offset < limit:
- v, offset = _read_item(buf, offset)
- val.append(v)
- # 't' (bool)
- elif ftype == 't':
- val, = unpack_from('>B', buf, offset)
- val = bool(val)
- offset += 1
- # 'T': timestamp
- elif ftype == 'T':
- val, = unpack_from('>Q', buf, offset)
- offset += 8
- val = datetime.utcfromtimestamp(val)
- # 'V': void
- elif ftype == 'V':
- val = None
- else:
- raise FrameSyntaxError(
- 'Unknown value in table: {0!r} ({1!r})'.format(
- ftype, type(ftype)))
- return val, offset
- def loads(format, buf, offset=0,
- ord=ord, unpack_from=unpack_from,
- _read_item=_read_item, pstr_t=pstr_t):
- """Deserialize amqp format.
- bit = b
- octet = o
- short = B
- long = l
- long long = L
- float = f
- shortstr = s
- longstr = S
- table = F
- array = A
- timestamp = T
- """
- bitcount = bits = 0
- values = []
- append = values.append
- format = pstr_t(format)
- for p in format:
- if p == 'b':
- if not bitcount:
- bits = ord(buf[offset:offset + 1])
- bitcount = 8
- val = (bits & 1) == 1
- bits >>= 1
- bitcount -= 1
- offset += 1
- elif p == 'o':
- bitcount = bits = 0
- val, = unpack_from('>B', buf, offset)
- offset += 1
- elif p == 'B':
- bitcount = bits = 0
- val, = unpack_from('>H', buf, offset)
- offset += 2
- elif p == 'l':
- bitcount = bits = 0
- val, = unpack_from('>I', buf, offset)
- offset += 4
- elif p == 'L':
- bitcount = bits = 0
- val, = unpack_from('>Q', buf, offset)
- offset += 8
- elif p == 'f':
- bitcount = bits = 0
- val, = unpack_from('>f', buf, offset)
- offset += 4
- elif p == 's':
- bitcount = bits = 0
- slen, = unpack_from('B', buf, offset)
- offset += 1
- val = buf[offset:offset + slen].decode('utf-8', 'surrogatepass')
- offset += slen
- elif p == 'S':
- bitcount = bits = 0
- slen, = unpack_from('>I', buf, offset)
- offset += 4
- val = buf[offset:offset + slen].decode('utf-8', 'surrogatepass')
- offset += slen
- elif p == 'x':
- blen, = unpack_from('>I', buf, offset)
- offset += 4
- val = buf[offset:offset + blen]
- offset += blen
- elif p == 'F':
- bitcount = bits = 0
- tlen, = unpack_from('>I', buf, offset)
- offset += 4
- limit = offset + tlen
- val = {}
- while offset < limit:
- keylen, = unpack_from('>B', buf, offset)
- offset += 1
- key = pstr_t(buf[offset:offset + keylen])
- offset += keylen
- val[key], offset = _read_item(buf, offset)
- elif p == 'A':
- bitcount = bits = 0
- alen, = unpack_from('>I', buf, offset)
- offset += 4
- limit = offset + alen
- val = []
- while offset < limit:
- aval, offset = _read_item(buf, offset)
- val.append(aval)
- elif p == 'T':
- bitcount = bits = 0
- val, = unpack_from('>Q', buf, offset)
- offset += 8
- val = datetime.utcfromtimestamp(val)
- else:
- raise FrameSyntaxError(ILLEGAL_TABLE_TYPE.format(p))
- append(val)
- return values, offset
- def _flushbits(bits, write, pack=pack):
- if bits:
- write(pack('B' * len(bits), *bits))
- bits[:] = []
- return 0
- def dumps(format, values):
- """Serialize AMQP arguments.
- Notes:
- bit = b
- octet = o
- short = B
- long = l
- long long = L
- shortstr = s
- longstr = S
- byte array = x
- table = F
- array = A
- """
- bitcount = 0
- bits = []
- out = BytesIO()
- write = out.write
- format = pstr_t(format)
- for i, val in enumerate(values):
- p = format[i]
- if p == 'b':
- val = 1 if val else 0
- shift = bitcount % 8
- if shift == 0:
- bits.append(0)
- bits[-1] |= (val << shift)
- bitcount += 1
- elif p == 'o':
- bitcount = _flushbits(bits, write)
- write(pack('B', val))
- elif p == 'B':
- bitcount = _flushbits(bits, write)
- write(pack('>H', int(val)))
- elif p == 'l':
- bitcount = _flushbits(bits, write)
- write(pack('>I', val))
- elif p == 'L':
- bitcount = _flushbits(bits, write)
- write(pack('>Q', val))
- elif p == 'f':
- bitcount = _flushbits(bits, write)
- write(pack('>f', val))
- elif p == 's':
- val = val or ''
- bitcount = _flushbits(bits, write)
- if isinstance(val, string):
- val = val.encode('utf-8', 'surrogatepass')
- write(pack('B', len(val)))
- write(val)
- elif p == 'S' or p == 'x':
- val = val or ''
- bitcount = _flushbits(bits, write)
- if isinstance(val, string):
- val = val.encode('utf-8', 'surrogatepass')
- write(pack('>I', len(val)))
- write(val)
- elif p == 'F':
- bitcount = _flushbits(bits, write)
- _write_table(val or {}, write, bits)
- elif p == 'A':
- bitcount = _flushbits(bits, write)
- _write_array(val or [], write, bits)
- elif p == 'T':
- write(pack('>Q', long_t(calendar.timegm(val.utctimetuple()))))
- _flushbits(bits, write)
- return out.getvalue()
- def _write_table(d, write, bits, pack=pack):
- out = BytesIO()
- twrite = out.write
- for k, v in items(d):
- if isinstance(k, string):
- k = k.encode('utf-8', 'surrogatepass')
- twrite(pack('B', len(k)))
- twrite(k)
- try:
- _write_item(v, twrite, bits)
- except ValueError:
- raise FrameSyntaxError(
- ILLEGAL_TABLE_TYPE_WITH_KEY.format(type(v), k, v))
- table_data = out.getvalue()
- write(pack('>I', len(table_data)))
- write(table_data)
- def _write_array(l, write, bits, pack=pack):
- out = BytesIO()
- awrite = out.write
- for v in l:
- try:
- _write_item(v, awrite, bits)
- except ValueError:
- raise FrameSyntaxError(
- ILLEGAL_TABLE_TYPE_WITH_VALUE.format(type(v), v))
- array_data = out.getvalue()
- write(pack('>I', len(array_data)))
- write(array_data)
- def _write_item(v, write, bits, pack=pack,
- string_t=string_t, bytes=bytes, string=string, bool=bool,
- float=float, int_types=int_types, Decimal=Decimal,
- datetime=datetime, dict=dict, list=list, tuple=tuple,
- None_t=None):
- if isinstance(v, (string_t, bytes)):
- if isinstance(v, string):
- v = v.encode('utf-8', 'surrogatepass')
- write(pack('>cI', b'S', len(v)))
- write(v)
- elif isinstance(v, bool):
- write(pack('>cB', b't', int(v)))
- elif isinstance(v, float):
- write(pack('>cd', b'd', v))
- elif isinstance(v, int_types):
- if v > 2147483647 or v < -2147483647:
- write(pack('>cq', b'L', v))
- else:
- write(pack('>ci', b'I', v))
- elif isinstance(v, Decimal):
- sign, digits, exponent = v.as_tuple()
- v = 0
- for d in digits:
- v = (v * 10) + d
- if sign:
- v = -v
- write(pack('>cBi', b'D', -exponent, v))
- elif isinstance(v, datetime):
- write(
- pack('>cQ', b'T', long_t(calendar.timegm(v.utctimetuple()))))
- elif isinstance(v, dict):
- write(b'F')
- _write_table(v, write, bits)
- elif isinstance(v, (list, tuple)):
- write(b'A')
- _write_array(v, write, bits)
- elif v is None_t:
- write(b'V')
- else:
- raise ValueError()
- def decode_properties_basic(buf, offset=0,
- unpack_from=unpack_from, pstr_t=pstr_t):
- """Decode basic properties."""
- properties = {}
- flags, = unpack_from('>H', buf, offset)
- offset += 2
- if flags & 0x8000:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['content_type'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- if flags & 0x4000:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['content_encoding'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- if flags & 0x2000:
- _f, offset = loads('F', buf, offset)
- properties['application_headers'], = _f
- if flags & 0x1000:
- properties['delivery_mode'], = unpack_from('>B', buf, offset)
- offset += 1
- if flags & 0x0800:
- properties['priority'], = unpack_from('>B', buf, offset)
- offset += 1
- if flags & 0x0400:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['correlation_id'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- if flags & 0x0200:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['reply_to'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- if flags & 0x0100:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['expiration'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- if flags & 0x0080:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['message_id'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- if flags & 0x0040:
- properties['timestamp'], = unpack_from('>Q', buf, offset)
- offset += 8
- if flags & 0x0020:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['type'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- if flags & 0x0010:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['user_id'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- if flags & 0x0008:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['app_id'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- if flags & 0x0004:
- slen, = unpack_from('>B', buf, offset)
- offset += 1
- properties['cluster_id'] = pstr_t(buf[offset:offset + slen])
- offset += slen
- return properties, offset
- PROPERTY_CLASSES = {
- Basic.CLASS_ID: decode_properties_basic,
- }
- class GenericContent(object):
- """Abstract base class for AMQP content.
- Subclasses should override the PROPERTIES attribute.
- """
- CLASS_ID = None
- PROPERTIES = [('dummy', 's')]
- def __init__(self, frame_method=None, frame_args=None, **props):
- self.frame_method = frame_method
- self.frame_args = frame_args
- self.properties = props
- self._pending_chunks = []
- self.body_received = 0
- self.body_size = 0
- self.ready = False
- def __getattr__(self, name):
- # Look for additional properties in the 'properties'
- # dictionary, and if present - the 'delivery_info' dictionary.
- if name == '__setstate__':
- # Allows pickling/unpickling to work
- raise AttributeError('__setstate__')
- if name in self.properties:
- return self.properties[name]
- raise AttributeError(name)
- def _load_properties(self, class_id, buf, offset=0,
- classes=PROPERTY_CLASSES, unpack_from=unpack_from):
- """Load AMQP properties.
- Given the raw bytes containing the property-flags and property-list
- from a content-frame-header, parse and insert into a dictionary
- stored in this object as an attribute named 'properties'.
- """
- # Read 16-bit shorts until we get one with a low bit set to zero
- props, offset = classes[class_id](buf, offset)
- self.properties = props
- return offset
- def _serialize_properties(self):
- """Serialize AMQP properties.
- Serialize the 'properties' attribute (a dictionary) into
- the raw bytes making up a set of property flags and a
- property list, suitable for putting into a content frame header.
- """
- shift = 15
- flag_bits = 0
- flags = []
- sformat, svalues = [], []
- props = self.properties
- for key, proptype in self.PROPERTIES:
- val = props.get(key, None)
- if val is not None:
- if shift == 0:
- flags.append(flag_bits)
- flag_bits = 0
- shift = 15
- flag_bits |= (1 << shift)
- if proptype != 'bit':
- sformat.append(str_to_bytes(proptype))
- svalues.append(val)
- shift -= 1
- flags.append(flag_bits)
- result = BytesIO()
- write = result.write
- for flag_bits in flags:
- write(pack('>H', flag_bits))
- write(dumps(b''.join(sformat), svalues))
- return result.getvalue()
- def inbound_header(self, buf, offset=0):
- class_id, self.body_size = unpack_from('>HxxQ', buf, offset)
- offset += 12
- self._load_properties(class_id, buf, offset)
- if not self.body_size:
- self.ready = True
- return offset
- def inbound_body(self, buf):
- chunks = self._pending_chunks
- self.body_received += len(buf)
- if self.body_received >= self.body_size:
- if chunks:
- chunks.append(buf)
- self.body = bytes().join(chunks)
- chunks[:] = []
- else:
- self.body = buf
- self.ready = True
- else:
- chunks.append(buf)
|