| # |
| # 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. |
| # |
| |
| import socket, struct, sys, time |
| from logging import getLogger, DEBUG |
| from qpid import compat |
| from qpid import sasl |
| from qpid.concurrency import synchronized |
| from qpid.datatypes import RangedSet, Serial |
| from qpid.framing import OpEncoder, SegmentEncoder, FrameEncoder, \ |
| FrameDecoder, SegmentDecoder, OpDecoder |
| from qpid.messaging import address, transports |
| from qpid.messaging.constants import UNLIMITED, REJECTED, RELEASED |
| from qpid.messaging.exceptions import * |
| from qpid.messaging.message import get_codec, Disposition, Message |
| from qpid.messaging.endpoints import MangledString |
| from qpid.ops import * |
| from qpid.selector import Selector |
| from qpid.util import URL, default,get_client_properties_with_defaults |
| from qpid.validator import And, Context, List, Map, Types, Values |
| from threading import Condition, Thread |
| |
| log = getLogger("qpid.messaging") |
| rawlog = getLogger("qpid.messaging.io.raw") |
| opslog = getLogger("qpid.messaging.io.ops") |
| |
| def addr2reply_to(addr): |
| name, subject, options = address.parse(addr) |
| if options: |
| type = options.get("node", {}).get("type") |
| else: |
| type = None |
| |
| if type == "topic": |
| return ReplyTo(name, subject) |
| else: |
| return ReplyTo(None, name) |
| |
| def reply_to2addr(reply_to): |
| if reply_to.exchange in (None, ""): |
| return reply_to.routing_key |
| elif reply_to.routing_key is None: |
| return "%s; {node: {type: topic}}" % reply_to.exchange |
| else: |
| return "%s/%s; {node: {type: topic}}" % (reply_to.exchange, reply_to.routing_key) |
| |
| class Attachment: |
| |
| def __init__(self, target): |
| self.target = target |
| |
| # XXX |
| |
| DURABLE_DEFAULT=False |
| |
| # XXX |
| |
| class Pattern: |
| """ |
| The pattern filter matches the supplied wildcard pattern against a |
| message subject. |
| """ |
| |
| def __init__(self, value): |
| self.value = value |
| |
| # XXX: this should become part of the driver |
| def _bind(self, sst, exchange, queue): |
| from qpid.ops import ExchangeBind |
| |
| sst.write_cmd(ExchangeBind(exchange=exchange, queue=queue, |
| binding_key=self.value.replace("*", "#"))) |
| |
| SUBJECT_DEFAULTS = { |
| "topic": "#" |
| } |
| |
| def noop(): pass |
| def sync_noop(): pass |
| |
| class SessionState: |
| |
| def __init__(self, driver, session, name, channel): |
| self.driver = driver |
| self.session = session |
| self.name = name |
| self.channel = channel |
| self.detached = False |
| self.committing = False |
| self.aborting = False |
| |
| # sender state |
| self.sent = Serial(0) |
| self.acknowledged = RangedSet() |
| self.actions = {} |
| self.min_completion = self.sent |
| self.max_completion = self.sent |
| self.results = {} |
| self.need_sync = False |
| |
| # receiver state |
| self.received = None |
| self.executed = RangedSet() |
| |
| # XXX: need to periodically exchange completion/known_completion |
| |
| self.destinations = {} |
| |
| def write_query(self, query, handler, obj): |
| id = self.sent |
| self.write_cmd(query, lambda: handler(self.results.pop(id), obj)) |
| |
| def apply_overrides(self, cmd, overrides): |
| for k, v in overrides.items(): |
| cmd[k.replace('-', '_')] = v |
| |
| def write_cmd(self, cmd, action=noop, overrides=None, sync=True): |
| if overrides: |
| self.apply_overrides(cmd, overrides) |
| |
| if action != noop: |
| cmd.sync = sync |
| if self.detached: |
| raise Exception("detached") |
| cmd.id = self.sent |
| self.sent += 1 |
| self.actions[cmd.id] = action |
| self.max_completion = cmd.id |
| self.write_op(cmd) |
| self.need_sync = not cmd.sync |
| |
| def write_cmds(self, cmds, action=noop): |
| if cmds: |
| for cmd in cmds[:-1]: |
| self.write_cmd(cmd) |
| self.write_cmd(cmds[-1], action) |
| else: |
| action() |
| |
| def write_op(self, op): |
| op.channel = self.channel |
| self.driver.write_op(op) |
| |
| POLICIES = Values("always", "sender", "receiver", "never") |
| RELIABILITY = Values("unreliable", "at-most-once", "at-least-once", |
| "exactly-once") |
| |
| DECLARE = Map({}, restricted=False) |
| BINDINGS = List(Map({ |
| "exchange": Types(basestring), |
| "queue": Types(basestring), |
| "key": Types(basestring), |
| "arguments": Map({}, restricted=False) |
| })) |
| |
| COMMON_OPTS = { |
| "create": POLICIES, |
| "delete": POLICIES, |
| "assert": POLICIES, |
| "node": Map({ |
| "type": Values("queue", "topic"), |
| "durable": Types(bool), |
| "x-declare": DECLARE, |
| "x-bindings": BINDINGS |
| }), |
| "link": Map({ |
| "name": Types(basestring), |
| "durable": Types(bool), |
| "reliability": RELIABILITY, |
| "x-declare": DECLARE, |
| "x-bindings": BINDINGS, |
| "x-subscribe": Map({}, restricted=False) |
| }) |
| } |
| |
| RECEIVE_MODES = Values("browse", "consume") |
| |
| SOURCE_OPTS = COMMON_OPTS.copy() |
| SOURCE_OPTS.update({ |
| "mode": RECEIVE_MODES |
| }) |
| |
| TARGET_OPTS = COMMON_OPTS.copy() |
| |
| class LinkIn: |
| |
| ADDR_NAME = "source" |
| DIR_NAME = "receiver" |
| VALIDATOR = Map(SOURCE_OPTS) |
| |
| def init_link(self, sst, rcv, _rcv): |
| _rcv.destination = str(rcv.id) |
| sst.destinations[_rcv.destination] = _rcv |
| _rcv.draining = False |
| _rcv.bytes_open = False |
| _rcv.on_unlink = [] |
| |
| def do_link(self, sst, rcv, _rcv, type, subtype, action): |
| link_opts = _rcv.options.get("link", {}) |
| if type == "topic": |
| default_reliability = "unreliable" |
| else: |
| default_reliability = "at-least-once" |
| reliability = link_opts.get("reliability", default_reliability) |
| declare = link_opts.get("x-declare", {}) |
| subscribe = link_opts.get("x-subscribe", {}) |
| acq_mode = acquire_mode.pre_acquired |
| if reliability in ("unreliable", "at-most-once"): |
| rcv._accept_mode = accept_mode.none |
| else: |
| rcv._accept_mode = accept_mode.explicit |
| |
| if type == "topic": |
| default_name = "%s.%s" % (rcv.session.name, _rcv.destination) |
| _rcv._queue = link_opts.get("name", default_name) |
| sst.write_cmd(QueueDeclare(queue=_rcv._queue, |
| durable=link_opts.get("durable", False), |
| exclusive=True, |
| auto_delete=(reliability == "unreliable")), |
| overrides=declare) |
| if declare.get("exclusive", True): _rcv.on_unlink = [QueueDelete(_rcv._queue)] |
| subject = _rcv.subject or SUBJECT_DEFAULTS.get(subtype) |
| bindings = get_bindings(link_opts, _rcv._queue, _rcv.name, subject) |
| if not bindings: |
| sst.write_cmd(ExchangeBind(_rcv._queue, _rcv.name, subject)) |
| |
| elif type == "queue": |
| _rcv._queue = _rcv.name |
| if _rcv.options.get("mode", "consume") == "browse": |
| acq_mode = acquire_mode.not_acquired |
| bindings = get_bindings(link_opts, queue=_rcv._queue) |
| |
| |
| sst.write_cmds(bindings) |
| sst.write_cmd(MessageSubscribe(queue=_rcv._queue, |
| destination=_rcv.destination, |
| acquire_mode = acq_mode, |
| accept_mode = rcv._accept_mode), |
| overrides=subscribe) |
| sst.write_cmd(MessageSetFlowMode(_rcv.destination, flow_mode.credit), action) |
| |
| def do_unlink(self, sst, rcv, _rcv, action=noop): |
| link_opts = _rcv.options.get("link", {}) |
| reliability = link_opts.get("reliability") |
| cmds = [MessageCancel(_rcv.destination)] |
| cmds.extend(_rcv.on_unlink) |
| msgs = [] #release back messages for the closing receiver |
| msg = rcv.session._pop(rcv) |
| while msg is not None: |
| msgs.append(msg) |
| msg = rcv.session._pop(rcv) |
| if len(msgs) > 0: |
| ids = RangedSet(*[m._transfer_id for m in msgs]) |
| log.debug("releasing back messages: %s, as receiver is closing", ids) |
| cmds.append(MessageRelease(ids, True)) |
| sst.write_cmds(cmds, action) |
| |
| def del_link(self, sst, rcv, _rcv): |
| del sst.destinations[_rcv.destination] |
| |
| class LinkOut: |
| |
| ADDR_NAME = "target" |
| DIR_NAME = "sender" |
| VALIDATOR = Map(TARGET_OPTS) |
| |
| def init_link(self, sst, snd, _snd): |
| _snd.closing = False |
| _snd.pre_ack = False |
| |
| def do_link(self, sst, snd, _snd, type, subtype, action): |
| link_opts = _snd.options.get("link", {}) |
| reliability = link_opts.get("reliability", "at-least-once") |
| _snd.pre_ack = reliability in ("unreliable", "at-most-once") |
| if type == "topic": |
| _snd._exchange = _snd.name |
| _snd._routing_key = _snd.subject |
| bindings = get_bindings(link_opts, exchange=_snd.name, key=_snd.subject) |
| elif type == "queue": |
| _snd._exchange = "" |
| _snd._routing_key = _snd.name |
| bindings = get_bindings(link_opts, queue=_snd.name) |
| sst.write_cmds(bindings, action) |
| |
| def do_unlink(self, sst, snd, _snd, action=noop): |
| action() |
| |
| def del_link(self, sst, snd, _snd): |
| pass |
| |
| class Cache: |
| |
| def __init__(self, ttl): |
| self.ttl = ttl |
| self.entries = {} |
| |
| def __setitem__(self, key, value): |
| self.entries[key] = time.time(), value |
| |
| def __getitem__(self, key): |
| tstamp, value = self.entries[key] |
| if time.time() - tstamp >= self.ttl: |
| del self.entries[key] |
| raise KeyError(key) |
| else: |
| return value |
| |
| def __delitem__(self, key): |
| del self.entries[key] |
| |
| # XXX |
| HEADER="!4s4B" |
| |
| EMPTY_DP = DeliveryProperties() |
| EMPTY_MP = MessageProperties() |
| |
| SUBJECT = "qpid.subject" |
| |
| CLOSED = "CLOSED" |
| READ_ONLY = "READ_ONLY" |
| WRITE_ONLY = "WRITE_ONLY" |
| OPEN = "OPEN" |
| |
| class Driver: |
| |
| def __init__(self, connection): |
| self.connection = connection |
| self.log_id = "%x" % id(self.connection) |
| self._lock = self.connection._lock |
| |
| self._selector = Selector.default() |
| self._attempts = 0 |
| self._delay = self.connection.reconnect_interval_min |
| self._reconnect_log = self.connection.reconnect_log |
| self._host = 0 |
| self._retrying = False |
| self._next_retry = None |
| self._transport = None |
| |
| self._timeout = None |
| |
| self.engine = None |
| |
| def _next_host(self): |
| urls = [URL(u) for u in self.connection.reconnect_urls] |
| hosts = [(self.connection.host, default(self.connection.port, 5672))] + \ |
| [(u.host, default(u.port, 5672)) for u in urls] |
| if self._host >= len(hosts): |
| self._host = 0 |
| self._last_host = hosts[self._host] |
| if self._host == 0: |
| self._attempts += 1 |
| self._host = self._host + 1 |
| return self._last_host |
| |
| def _num_hosts(self): |
| return len(self.connection.reconnect_urls) + 1 |
| |
| @synchronized |
| def wakeup(self): |
| self.dispatch() |
| self._selector.wakeup() |
| |
| def start(self): |
| self._selector.register(self) |
| |
| def stop(self): |
| self._selector.unregister(self) |
| if self._transport: |
| self.st_closed() |
| |
| def fileno(self): |
| return self._transport.fileno() |
| |
| @synchronized |
| def reading(self): |
| """Called by the Selector I/O thread to determine if the driver needs to |
| wait on the arrival of network data (call self.readable() callback) |
| """ |
| return self._transport is not None and \ |
| self._transport.reading(True) |
| |
| @synchronized |
| def writing(self): |
| """Called by the Selector I/O thread to determine if it should block |
| waiting for output bandwidth (call the self.writeable() callback) |
| """ |
| return self._transport is not None and \ |
| self._transport.writing(self.engine.pending()) |
| |
| @synchronized |
| def timing(self): |
| """Called by the Selector I/O thread to determine if it should wake up the |
| driver (call the timeout() callback |
| """ |
| return self._timeout |
| |
| @synchronized |
| def abort(self, exc, info): |
| """Called if the Selector I/O thread hits an unrecoverable error and fails. |
| """ |
| try: |
| self.connection.error = exc |
| log.error("I/O Thread Fatal error: %s\n%s" % (str(exc), info)) |
| except: |
| pass |
| |
| def _check_retry_ok(self): |
| """We consider a reconnect to have suceeded only when we have received |
| open-ok from the peer. |
| |
| If we declared success as soon as the transport connected, then we could get |
| into an infinite heartbeat loop if the remote process is hung and never |
| sends us any data. We would fail the connection after 2 missed heartbeats, |
| reconnect the transport, declare the reconnect ok, then fail again after 2 |
| missed heartbeats and so on. |
| """ |
| if self._retrying and self.engine._connected: # Means we have received open-ok. |
| if self._reconnect_log: |
| log.warn("reconnect succeeded: %s:%s", *self._last_host) |
| self._next_retry = None |
| self._attempts = 0 |
| self._delay = self.connection.reconnect_interval_min |
| self._retrying = False |
| |
| @synchronized |
| def readable(self): |
| try: |
| data = self._transport.recv(64*1024) |
| if data is None: |
| return |
| elif data: |
| rawlog.debug("READ[%s]: %r", self.log_id, data) |
| self.engine.write(data) |
| self._check_retry_ok() |
| else: |
| self.close_engine() |
| except socket.error, e: |
| self.close_engine(ConnectionError(text=str(e))) |
| |
| self.update_status() |
| |
| self._notify() |
| |
| def _notify(self): |
| if self.connection.error: |
| self.connection._condition.gc() |
| self.connection._waiter.notifyAll() |
| |
| def close_engine(self, e=None): |
| if e is None: |
| e = ConnectionError(text="connection aborted") |
| |
| if (self.connection.reconnect and |
| (self.connection.reconnect_limit is None or |
| self.connection.reconnect_limit <= 0 or |
| self._attempts <= self.connection.reconnect_limit)): |
| if self._host < self._num_hosts(): |
| delay = 0 |
| else: |
| delay = self._delay |
| self._delay = min(2*self._delay, |
| self.connection.reconnect_interval_max) |
| self._next_retry = time.time() + delay |
| if self._reconnect_log: |
| log.warn("recoverable error[attempt %s]: %s" % (self._attempts, e)) |
| if delay > 0: |
| log.warn("sleeping %s seconds" % delay) |
| self._retrying = True |
| self.engine.close() |
| else: |
| self.engine.close(e) |
| |
| self.schedule() |
| |
| def update_status(self): |
| if not self.engine: return False |
| status = self.engine.status() |
| return getattr(self, "st_%s" % status.lower())() |
| |
| def st_closed(self): |
| # XXX: this log statement seems to sometimes hit when the socket is not connected |
| # XXX: rawlog.debug("CLOSE[%s]: %s", self.log_id, self._socket.getpeername()) |
| if self._transport: self._transport.close() |
| self._transport = None |
| self.engine = None |
| return True |
| |
| def st_open(self): |
| return False |
| |
| @synchronized |
| def writeable(self): |
| notify = False |
| try: |
| n = self._transport.send(self.engine.peek()) |
| if n == 0: return |
| sent = self.engine.read(n) |
| rawlog.debug("SENT[%s]: %r", self.log_id, sent) |
| except socket.error, e: |
| self.close_engine(e) |
| notify = True |
| |
| if self.update_status() or notify: |
| self._notify() |
| |
| @synchronized |
| def timeout(self): |
| self.dispatch() |
| self.update_status() |
| self._notify() |
| self.schedule() |
| |
| def schedule(self): |
| times = [] |
| if self.connection.heartbeat: |
| times.append(time.time() + self.connection.heartbeat) |
| if self._next_retry: |
| times.append(self._next_retry) |
| if times: |
| self._timeout = min(times) |
| else: |
| self._timeout = None |
| |
| def dispatch(self): |
| try: |
| if self._transport is None: |
| if self.connection._connected and not self.connection.error: |
| self.connect() |
| else: |
| self.engine.dispatch() |
| except HeartbeatTimeout, e: |
| self.close_engine(e) |
| except ContentError, e: |
| msg = compat.format_exc() |
| self.connection.error = ContentError(text=msg) |
| except: |
| # XXX: Does socket get leaked if this occurs? |
| msg = compat.format_exc() |
| self.connection.error = InternalError(text=msg) |
| |
| def connect(self): |
| if self._retrying and time.time() < self._next_retry: |
| return |
| |
| try: |
| # XXX: should make this non blocking |
| host, port = self._next_host() |
| if self._retrying and self._reconnect_log: |
| log.warn("trying: %s:%s", host, port) |
| self.engine = Engine(self.connection) |
| self.engine.open() |
| rawlog.debug("OPEN[%s]: %s:%s", self.log_id, host, port) |
| trans = transports.TRANSPORTS.get(self.connection.transport) |
| if trans: |
| self._transport = trans(self.connection, host, port) |
| else: |
| raise ConnectError("no such transport: %s" % self.connection.transport) |
| self.schedule() |
| except socket.error, e: |
| self.close_engine(ConnectError(text=str(e))) |
| |
| DEFAULT_DISPOSITION = Disposition(None) |
| |
| def get_bindings(opts, queue=None, exchange=None, key=None): |
| bindings = opts.get("x-bindings", []) |
| cmds = [] |
| for b in bindings: |
| exchange = b.get("exchange", exchange) |
| queue = b.get("queue", queue) |
| key = b.get("key", key) |
| args = b.get("arguments", {}) |
| cmds.append(ExchangeBind(queue, exchange, key, args)) |
| return cmds |
| |
| CONNECTION_ERRS = { |
| # anythong not here (i.e. everything right now) will default to |
| # connection error |
| } |
| |
| SESSION_ERRS = { |
| # anything not here will default to session error |
| error_code.unauthorized_access: UnauthorizedAccess, |
| error_code.not_found: NotFound, |
| error_code.resource_locked: ReceiverError, |
| error_code.resource_limit_exceeded: TargetCapacityExceeded, |
| error_code.internal_error: ServerError |
| } |
| |
| class Engine: |
| |
| def __init__(self, connection): |
| self.connection = connection |
| self.log_id = "%x" % id(self.connection) |
| self._closing = False |
| self._connected = False |
| self._reconnecting = bool(connection.sessions) |
| self._attachments = {} |
| |
| self._in = LinkIn() |
| self._out = LinkOut() |
| |
| self._channel_max = 65536 |
| self._channels = 0 |
| self._sessions = {} |
| |
| self.address_cache = Cache(self.connection.address_ttl) |
| |
| self._status = CLOSED |
| self._buf = "" |
| self._hdr = "" |
| # Set _last_in and _last_out here so heartbeats will be timed from the |
| # beginning of connection if no data is sent/received. |
| self._last_in = time.time() |
| self._last_out = time.time() |
| self._op_enc = OpEncoder() |
| self._seg_enc = SegmentEncoder() |
| self._frame_enc = FrameEncoder() |
| self._frame_dec = FrameDecoder() |
| self._seg_dec = SegmentDecoder() |
| self._op_dec = OpDecoder() |
| |
| self._sasl = sasl.Client() |
| if self.connection.username: |
| self._sasl.setAttr("username", self.connection.username) |
| if self.connection.password: |
| self._sasl.setAttr("password", self.connection.password) |
| if self.connection.host: |
| self._sasl.setAttr("host", self.connection.host) |
| self._sasl.setAttr("service", self.connection.sasl_service) |
| if self.connection.sasl_min_ssf is not None: |
| self._sasl.setAttr("minssf", self.connection.sasl_min_ssf) |
| if self.connection.sasl_max_ssf is not None: |
| self._sasl.setAttr("maxssf", self.connection.sasl_max_ssf) |
| self._sasl.init() |
| self._sasl_encode = False |
| self._sasl_decode = False |
| |
| def _reset(self): |
| self.connection._transport_connected = False |
| |
| for ssn in self.connection.sessions.values(): |
| for m in ssn.acked + ssn.unacked + ssn.incoming: |
| m._transfer_id = None |
| for snd in ssn.senders: |
| snd.linked = False |
| for rcv in ssn.receivers: |
| rcv.impending = rcv.received |
| rcv.linked = False |
| |
| def status(self): |
| return self._status |
| |
| def write(self, data): |
| self._last_in = time.time() |
| try: |
| if self._sasl_decode: |
| data = self._sasl.decode(data) |
| |
| if len(self._hdr) < 8: |
| r = 8 - len(self._hdr) |
| self._hdr += data[:r] |
| data = data[r:] |
| |
| if len(self._hdr) == 8: |
| self.do_header(self._hdr) |
| |
| self._frame_dec.write(data) |
| self._seg_dec.write(*self._frame_dec.read()) |
| self._op_dec.write(*self._seg_dec.read()) |
| for op in self._op_dec.read(): |
| self.assign_id(op) |
| opslog.debug("RCVD[%s]: %r", self.log_id, op) |
| op.dispatch(self) |
| self.dispatch() |
| except MessagingError, e: |
| self.close(e) |
| except: |
| self.close(InternalError(text=compat.format_exc())) |
| |
| def close(self, e=None): |
| self._reset() |
| # We cannot re-establish transactional sessions, they must be aborted. |
| # We could re-do transactional enqueues, but not dequeues. |
| for ssn in self.connection.sessions.values(): |
| if ssn.transactional: |
| if ssn.committing: |
| ssn.error = TransactionUnknown(text="Transaction outcome unknown due to transport failure") |
| else: |
| ssn.error = TransactionAborted(text="Transaction aborted due to transport failure") |
| ssn.closed = True |
| if e: |
| self.connection.error = e |
| self._status = CLOSED |
| |
| def assign_id(self, op): |
| if isinstance(op, Command): |
| sst = self.get_sst(op) |
| op.id = sst.received |
| sst.received += 1 |
| |
| def pending(self): |
| return len(self._buf) |
| |
| def read(self, n): |
| result = self._buf[:n] |
| self._buf = self._buf[n:] |
| return result |
| |
| def peek(self): |
| return self._buf |
| |
| def write_op(self, op): |
| opslog.debug("SENT[%s]: %r", self.log_id, op) |
| self._op_enc.write(op) |
| self._seg_enc.write(*self._op_enc.read()) |
| self._frame_enc.write(*self._seg_enc.read()) |
| bytes = self._frame_enc.read() |
| if self._sasl_encode: |
| bytes = self._sasl.encode(bytes) |
| self._buf += bytes |
| self._last_out = time.time() |
| |
| def do_header(self, hdr): |
| cli_major = 0; cli_minor = 10 |
| magic, _, _, major, minor = struct.unpack(HEADER, hdr) |
| if major != cli_major or minor != cli_minor: |
| raise VersionError(text="client: %s-%s, server: %s-%s" % |
| (cli_major, cli_minor, major, minor)) |
| |
| def do_connection_start(self, start): |
| if self.connection.sasl_mechanisms: |
| permitted = self.connection.sasl_mechanisms.split() |
| mechs = [m for m in start.mechanisms if m in permitted] |
| else: |
| mechs = start.mechanisms |
| try: |
| mech, initial = self._sasl.start(" ".join(mechs)) |
| except sasl.SASLError, e: |
| if "ANONYMOUS" not in mechs and self.connection.username is None: |
| _text="Anonymous connections disabled, missing credentials" |
| else: |
| _text=str(e) |
| raise AuthenticationFailure(text=_text) |
| |
| client_properties = get_client_properties_with_defaults(provided_client_properties=self.connection.client_properties); |
| self.write_op(ConnectionStartOk(client_properties=client_properties, |
| mechanism=mech, response=initial)) |
| |
| def do_connection_secure(self, secure): |
| resp = self._sasl.step(secure.challenge) |
| self.write_op(ConnectionSecureOk(response=resp)) |
| |
| def do_connection_tune(self, tune): |
| # XXX: is heartbeat protocol specific? |
| if tune.channel_max is not None: |
| self.channel_max = tune.channel_max |
| self.write_op(ConnectionTuneOk(heartbeat=self.connection.heartbeat, |
| channel_max=self.channel_max)) |
| self.write_op(ConnectionOpen()) |
| self._sasl_encode = True |
| |
| def do_connection_open_ok(self, open_ok): |
| self.connection.auth_username = self._sasl.auth_username() |
| self._connected = True |
| self._sasl_decode = True |
| self.connection._transport_connected = True |
| |
| def do_connection_heartbeat(self, hrt): |
| pass |
| |
| def do_connection_close(self, close): |
| self.write_op(ConnectionCloseOk()) |
| if close.reply_code != close_code.normal: |
| exc = CONNECTION_ERRS.get(close.reply_code, ConnectionError) |
| self.connection.error = exc(close.reply_code, close.reply_text) |
| # XXX: should we do a half shutdown on the socket here? |
| # XXX: we really need to test this, we may end up reporting a |
| # connection abort after this, if we were to do a shutdown on read |
| # and stop reading, then we wouldn't report the abort, that's |
| # probably the right thing to do |
| |
| def do_connection_close_ok(self, close_ok): |
| self.close() |
| |
| def do_session_attached(self, atc): |
| pass |
| |
| def do_session_command_point(self, cp): |
| sst = self.get_sst(cp) |
| sst.received = cp.command_id |
| |
| def do_session_completed(self, sc): |
| sst = self.get_sst(sc) |
| for r in sc.commands: |
| sst.acknowledged.add(r.lower, r.upper) |
| |
| if not sc.commands.empty(): |
| while sst.min_completion in sc.commands: |
| if sst.actions.has_key(sst.min_completion): |
| sst.actions.pop(sst.min_completion)() |
| sst.min_completion += 1 |
| |
| def session_known_completed(self, kcmp): |
| sst = self.get_sst(kcmp) |
| executed = RangedSet() |
| for e in sst.executed.ranges: |
| for ke in kcmp.ranges: |
| if e.lower in ke and e.upper in ke: |
| break |
| else: |
| executed.add_range(e) |
| sst.executed = completed |
| |
| def do_session_flush(self, sf): |
| sst = self.get_sst(sf) |
| if sf.expected: |
| if sst.received is None: |
| exp = None |
| else: |
| exp = RangedSet(sst.received) |
| sst.write_op(SessionExpected(exp)) |
| if sf.confirmed: |
| sst.write_op(SessionConfirmed(sst.executed)) |
| if sf.completed: |
| sst.write_op(SessionCompleted(sst.executed)) |
| |
| def do_session_request_timeout(self, rt): |
| sst = self.get_sst(rt) |
| sst.write_op(SessionTimeout(timeout=0)) |
| |
| def do_execution_result(self, er): |
| sst = self.get_sst(er) |
| sst.results[er.command_id] = er.value |
| sst.executed.add(er.id) |
| |
| def do_execution_exception(self, ex): |
| sst = self.get_sst(ex) |
| exc = SESSION_ERRS.get(ex.error_code, SessionError) |
| sst.session.error = exc(ex.error_code, ex.description) |
| |
| def dispatch(self): |
| if not self.connection._connected and not self._closing and self._status != CLOSED: |
| self.disconnect() |
| |
| if self._connected and not self._closing: |
| for ssn in self.connection.sessions.values(): |
| self.attach(ssn) |
| self.process(ssn) |
| |
| # We need to check heartbeat even if not self._connected since we may have |
| # heartbeat timeout before receiving an open-ok |
| if self.connection.heartbeat and self._status != CLOSED and not self._closing: |
| now = time.time() |
| if now - self._last_in > 2*self.connection.heartbeat: |
| raise HeartbeatTimeout(text="heartbeat timeout") |
| # Only send heartbeats if we are connected. |
| if self._connected and now - self._last_out >= self.connection.heartbeat/2.0: |
| self.write_op(ConnectionHeartbeat()) |
| |
| def open(self): |
| self._reset() |
| self._status = OPEN |
| self._buf += struct.pack(HEADER, "AMQP", 1, 1, 0, 10) |
| |
| def disconnect(self): |
| self.write_op(ConnectionClose(close_code.normal)) |
| self._closing = True |
| |
| def attach(self, ssn): |
| if ssn.closed: return |
| sst = self._attachments.get(ssn) |
| if sst is None: |
| for i in xrange(0, self.channel_max): |
| if not self._sessions.has_key(i): |
| ch = i |
| break |
| else: |
| raise RuntimeError("all channels used") |
| sst = SessionState(self, ssn, ssn.name, ch) |
| sst.write_op(SessionAttach(name=ssn.name, force=self._reconnecting)) |
| sst.write_op(SessionCommandPoint(sst.sent, 0)) |
| self._reconnecting = False |
| sst.outgoing_idx = 0 |
| sst.acked = [] |
| sst.acked_idx = 0 |
| if ssn.transactional: |
| sst.write_cmd(TxSelect()) |
| self._attachments[ssn] = sst |
| self._sessions[sst.channel] = sst |
| |
| for snd in ssn.senders: |
| self.link(snd, self._out, snd.target) |
| for rcv in ssn.receivers: |
| self.link(rcv, self._in, rcv.source) |
| |
| if sst is not None and ssn.closing and not sst.detached: |
| sst.detached = True |
| sst.write_op(SessionDetach(name=ssn.name)) |
| |
| def get_sst(self, op): |
| return self._sessions[op.channel] |
| |
| def do_session_detached(self, dtc): |
| sst = self._sessions.pop(dtc.channel) |
| ssn = sst.session |
| del self._attachments[ssn] |
| ssn.closed = True |
| |
| def do_session_detach(self, dtc): |
| sst = self.get_sst(dtc) |
| sst.write_op(SessionDetached(name=dtc.name)) |
| self.do_session_detached(dtc) |
| |
| def link(self, lnk, dir, addr): |
| sst = self._attachments.get(lnk.session) |
| _lnk = self._attachments.get(lnk) |
| |
| if _lnk is None and not lnk.closed: |
| _lnk = Attachment(lnk) |
| _lnk.closing = False |
| dir.init_link(sst, lnk, _lnk) |
| |
| err = self.parse_address(_lnk, dir, addr) or self.validate_options(_lnk, dir) |
| if err: |
| lnk.error = err |
| lnk.closed = True |
| return |
| |
| def linked(): |
| lnk.linked = True |
| |
| def resolved(type, subtype): |
| dir.do_link(sst, lnk, _lnk, type, subtype, linked) |
| |
| self.resolve_declare(sst, _lnk, dir.DIR_NAME, resolved) |
| self._attachments[lnk] = _lnk |
| |
| if lnk.linked and lnk.closing and not lnk.closed: |
| if not _lnk.closing: |
| def unlinked(): |
| dir.del_link(sst, lnk, _lnk) |
| del self._attachments[lnk] |
| lnk.closed = True |
| if _lnk.options.get("delete") in ("always", dir.DIR_NAME): |
| dir.do_unlink(sst, lnk, _lnk) |
| requested_type = _lnk.options.get("node", {}).get("type") |
| self.delete(sst, _lnk.name, unlinked, node_type=requested_type) |
| else: |
| dir.do_unlink(sst, lnk, _lnk, unlinked) |
| _lnk.closing = True |
| elif not lnk.linked and lnk.closing and not lnk.closed: |
| if lnk.error: lnk.closed = True |
| |
| def parse_address(self, lnk, dir, addr): |
| if addr is None: |
| return MalformedAddress(text="%s is None" % dir.ADDR_NAME) |
| else: |
| try: |
| lnk.name, lnk.subject, lnk.options = address.parse(addr) |
| # XXX: subject |
| if lnk.options is None: |
| lnk.options = {} |
| if isinstance(addr, MangledString): |
| lnk.options['create'] = "always" |
| if 'node' not in lnk.options: |
| lnk.options['node'] = {} |
| if 'x-declare' not in lnk.options['node']: |
| lnk.options['node']['x-declare'] = {} |
| xdeclare = lnk.options['node']['x-declare'] |
| if 'auto-delete' not in xdeclare: |
| xdeclare['auto-delete'] = "True" |
| if 'exclusive' not in xdeclare: |
| xdeclare['exclusive'] = "True" |
| except address.LexError, e: |
| return MalformedAddress(text=str(e)) |
| except address.ParseError, e: |
| return MalformedAddress(text=str(e)) |
| |
| def validate_options(self, lnk, dir): |
| ctx = Context() |
| err = dir.VALIDATOR.validate(lnk.options, ctx) |
| if err: return InvalidOption(text="error in options: %s" % err) |
| |
| def resolve_declare(self, sst, lnk, dir, action): |
| declare = lnk.options.get("create") in ("always", dir) |
| assrt = lnk.options.get("assert") in ("always", dir) |
| requested_type = lnk.options.get("node", {}).get("type") |
| def do_resolved(type, subtype): |
| err = None |
| if type is None: |
| if declare: |
| err = self.declare(sst, lnk, action, True) |
| else: |
| err = NotFound(text="no such %s: %s" % (requested_type or "queue", lnk.name)) |
| else: |
| if assrt: |
| expected = lnk.options.get("node", {}).get("type") |
| if expected and type != expected: |
| if declare: |
| err = self.declare(sst, lnk, action, True) |
| else: |
| err = AssertionFailed(text="expected %s, got %s" % (expected, type)) |
| if "node" in lnk.options and "x-bindings" in lnk.options["node"]: |
| err = self.declare(sst, lnk, action, False) |
| if err is None: |
| action(type, subtype) |
| |
| if err: |
| tgt = lnk.target |
| tgt.error = err |
| del self._attachments[tgt] |
| tgt.closed = True |
| return |
| self.resolve(sst, lnk.name, do_resolved, node_type=requested_type, force=declare) |
| |
| def resolve(self, sst, name, action, force=False, node_type=None, delete=False): |
| if not force and not node_type: |
| try: |
| type, subtype = self.address_cache[name] |
| action(type, subtype) |
| return |
| except KeyError: |
| pass |
| |
| args = { "topic":None, "queue":None } |
| def do_result(r, obj): |
| args[obj] = r |
| def do_action(): |
| er = args["topic"] |
| qr = args["queue"] |
| if node_type == "topic" and er and not er.not_found: |
| type, subtype = "topic", er.type |
| elif node_type == "queue" and qr and qr.queue: |
| type, subtype = "queue", None |
| elif (er and er.not_found) and qr and not qr.queue: |
| type, subtype = None, None |
| elif (qr and qr.queue): |
| if node_type == "topic" and force: |
| type, subtype = None, None |
| else: |
| type, subtype = "queue", None |
| elif (er and not er.not_found): |
| if node_type == "queue" and force: |
| type, subtype = None, None |
| else: |
| type, subtype = "topic", er.type |
| elif er: |
| if er.not_found: |
| type, subtype = None, None |
| else: |
| type, subtype = "topic", er.type |
| else: |
| type, subtype = None, None |
| if type is not None: |
| self.address_cache[name] = (type, subtype) |
| action(type, subtype) |
| def do_result_and_action(r, obj): |
| do_result(r, obj) |
| do_action() |
| if (node_type is None): # we don't know the type, let check broker |
| sst.write_query(ExchangeQuery(name), do_result, "topic") |
| sst.write_query(QueueQuery(name), do_result_and_action, "queue") |
| elif force and not delete: # we forcefully declare known type, dont ask broker |
| do_action() |
| elif node_type == "topic": |
| sst.write_query(ExchangeQuery(name), do_result_and_action, "topic") |
| else: |
| sst.write_query(QueueQuery(name), do_result_and_action, "queue") |
| |
| def declare(self, sst, lnk, action, create_node): |
| name = lnk.name |
| props = lnk.options.get("node", {}) |
| durable = props.get("durable", DURABLE_DEFAULT) |
| type = props.get("type", "queue") |
| declare = props.get("x-declare", {}) |
| |
| cmd = None |
| if type == "topic": |
| if create_node: cmd = ExchangeDeclare(exchange=name, durable=durable) |
| bindings = get_bindings(props, exchange=name) |
| elif type == "queue": |
| if create_node: cmd = QueueDeclare(queue=name, durable=durable) |
| bindings = get_bindings(props, queue=name) |
| else: |
| raise ValueError(type) |
| |
| if cmd is not None: |
| sst.apply_overrides(cmd, declare) |
| if type == "topic": |
| if cmd.type is None: |
| cmd.type = "topic" |
| subtype = cmd.type |
| else: |
| subtype = None |
| cmds = [cmd] |
| else: |
| cmds = [] |
| |
| cmds.extend(bindings) |
| |
| def declared(): |
| if create_node: |
| self.address_cache[name] = (type, subtype) |
| action(type, subtype) |
| |
| sst.write_cmds(cmds, declared) |
| |
| def delete(self, sst, name, action, node_type=None): |
| def deleted(): |
| del self.address_cache[name] |
| action() |
| |
| def do_delete(type, subtype): |
| if type == "topic": |
| sst.write_cmd(ExchangeDelete(name), deleted) |
| elif type == "queue": |
| sst.write_cmd(QueueDelete(name), deleted) |
| elif type is None: |
| action() |
| else: |
| raise ValueError(type) |
| self.resolve(sst, name, do_delete, force=True, node_type=node_type, delete=True) |
| |
| def process(self, ssn): |
| if ssn.closed or ssn.closing: return |
| |
| sst = self._attachments[ssn] |
| |
| while sst.outgoing_idx < len(ssn.outgoing): |
| msg = ssn.outgoing[sst.outgoing_idx] |
| snd = msg._sender |
| # XXX: should check for sender error here |
| _snd = self._attachments.get(snd) |
| if _snd and snd.linked: |
| self.send(snd, msg) |
| sst.outgoing_idx += 1 |
| else: |
| break |
| |
| for snd in ssn.senders: |
| # XXX: should included snd.acked in this |
| if snd.synced >= snd.queued and sst.need_sync: |
| sst.write_cmd(ExecutionSync(), sync_noop) |
| |
| for rcv in ssn.receivers: |
| self.process_receiver(rcv) |
| |
| if ssn.acked: |
| messages = ssn.acked[sst.acked_idx:] |
| if messages: |
| ids = RangedSet() |
| |
| disposed = [(DEFAULT_DISPOSITION, [])] |
| acked = [] |
| for m in messages: |
| # XXX: we're ignoring acks that get lost when disconnected, |
| # could we deal this via some message-id based purge? |
| if m._transfer_id is None: |
| acked.append(m) |
| continue |
| ids.add(m._transfer_id) |
| if m._receiver._accept_mode is accept_mode.explicit: |
| disp = m._disposition or DEFAULT_DISPOSITION |
| last, msgs = disposed[-1] |
| if disp.type is last.type and disp.options == last.options: |
| msgs.append(m) |
| else: |
| disposed.append((disp, [m])) |
| else: |
| acked.append(m) |
| |
| for range in ids: |
| sst.executed.add_range(range) |
| sst.write_op(SessionCompleted(sst.executed)) |
| |
| def ack_acker(msgs): |
| def ack_ack(): |
| for m in msgs: |
| ssn.acked.remove(m) |
| sst.acked_idx -= 1 |
| # XXX: should this check accept_mode too? |
| if not ssn.transactional: |
| sst.acked.remove(m) |
| return ack_ack |
| |
| for disp, msgs in disposed: |
| if not msgs: continue |
| if disp.type is None: |
| op = MessageAccept |
| elif disp.type is RELEASED: |
| op = MessageRelease |
| elif disp.type is REJECTED: |
| op = MessageReject |
| sst.write_cmd(op(RangedSet(*[m._transfer_id for m in msgs]), |
| **disp.options), |
| ack_acker(msgs)) |
| if log.isEnabledFor(DEBUG): |
| for m in msgs: |
| log.debug("SACK[%s]: %s, %s", ssn.log_id, m, m._disposition) |
| |
| sst.acked.extend(messages) |
| sst.acked_idx += len(messages) |
| ack_acker(acked)() |
| |
| if ssn.committing and not sst.committing: |
| def commit_ok(): |
| del sst.acked[:] |
| ssn.committing = False |
| ssn.committed = True |
| ssn.aborting = False |
| ssn.aborted = False |
| sst.committing = False |
| sst.write_cmd(TxCommit(), commit_ok) |
| sst.committing = True |
| |
| if ssn.aborting and not sst.aborting: |
| sst.aborting = True |
| def do_rb(): |
| messages = sst.acked + ssn.unacked + ssn.incoming |
| ids = RangedSet(*[m._transfer_id for m in messages]) |
| for range in ids: |
| sst.executed.add_range(range) |
| sst.write_op(SessionCompleted(sst.executed)) |
| sst.write_cmd(MessageRelease(ids, True)) |
| sst.write_cmd(TxRollback(), do_rb_ok) |
| |
| def do_rb_ok(): |
| del ssn.incoming[:] |
| del ssn.unacked[:] |
| del sst.acked[:] |
| |
| for rcv in ssn.receivers: |
| rcv.impending = rcv.received |
| rcv.returned = rcv.received |
| # XXX: do we need to update granted here as well? |
| |
| for rcv in ssn.receivers: |
| self.process_receiver(rcv) |
| |
| ssn.aborting = False |
| ssn.aborted = True |
| ssn.committing = False |
| ssn.committed = False |
| sst.aborting = False |
| |
| for rcv in ssn.receivers: |
| _rcv = self._attachments[rcv] |
| sst.write_cmd(MessageStop(_rcv.destination)) |
| sst.write_cmd(ExecutionSync(), do_rb) |
| |
| def grant(self, rcv): |
| sst = self._attachments[rcv.session] |
| _rcv = self._attachments.get(rcv) |
| if _rcv is None or not rcv.linked or _rcv.closing or _rcv.draining: |
| return |
| |
| if rcv.granted is UNLIMITED: |
| if rcv.impending is UNLIMITED: |
| delta = 0 |
| else: |
| delta = UNLIMITED |
| elif rcv.impending is UNLIMITED: |
| delta = -1 |
| else: |
| delta = max(rcv.granted, rcv.received) - rcv.impending |
| |
| if delta is UNLIMITED: |
| if not _rcv.bytes_open: |
| sst.write_cmd(MessageFlow(_rcv.destination, credit_unit.byte, UNLIMITED.value)) |
| _rcv.bytes_open = True |
| sst.write_cmd(MessageFlow(_rcv.destination, credit_unit.message, UNLIMITED.value)) |
| rcv.impending = UNLIMITED |
| elif delta > 0: |
| if not _rcv.bytes_open: |
| sst.write_cmd(MessageFlow(_rcv.destination, credit_unit.byte, UNLIMITED.value)) |
| _rcv.bytes_open = True |
| sst.write_cmd(MessageFlow(_rcv.destination, credit_unit.message, delta)) |
| rcv.impending += delta |
| elif delta < 0 and not rcv.draining: |
| _rcv.draining = True |
| def do_stop(): |
| rcv.impending = rcv.received |
| _rcv.draining = False |
| _rcv.bytes_open = False |
| self.grant(rcv) |
| sst.write_cmd(MessageStop(_rcv.destination), do_stop) |
| |
| if rcv.draining: |
| _rcv.draining = True |
| def do_flush(): |
| rcv.impending = rcv.received |
| rcv.granted = rcv.impending |
| _rcv.draining = False |
| _rcv.bytes_open = False |
| rcv.draining = False |
| sst.write_cmd(MessageFlush(_rcv.destination), do_flush) |
| |
| |
| def process_receiver(self, rcv): |
| if rcv.closed: return |
| self.grant(rcv) |
| |
| def send(self, snd, msg): |
| sst = self._attachments[snd.session] |
| _snd = self._attachments[snd] |
| |
| if msg.subject is None or _snd._exchange == "": |
| rk = _snd._routing_key |
| else: |
| rk = msg.subject |
| |
| if msg.subject is None: |
| subject = _snd.subject |
| else: |
| subject = msg.subject |
| |
| # XXX: do we need to query to figure out how to create the reply-to interoperably? |
| if msg.reply_to: |
| rt = addr2reply_to(msg.reply_to) |
| else: |
| rt = None |
| content_encoding = msg.properties.get("x-amqp-0-10.content-encoding") |
| dp = DeliveryProperties(routing_key=rk) |
| mp = MessageProperties(message_id=msg.id, |
| user_id=msg.user_id, |
| reply_to=rt, |
| correlation_id=msg.correlation_id, |
| app_id = msg.properties.get("x-amqp-0-10.app-id"), |
| content_type=msg.content_type, |
| content_encoding=content_encoding, |
| application_headers=msg.properties) |
| if subject is not None: |
| if mp.application_headers is None: |
| mp.application_headers = {} |
| mp.application_headers[SUBJECT] = subject |
| if msg.durable is not None: |
| if msg.durable: |
| dp.delivery_mode = delivery_mode.persistent |
| else: |
| dp.delivery_mode = delivery_mode.non_persistent |
| if msg.priority is not None: |
| dp.priority = msg.priority |
| if msg.ttl is not None: |
| dp.ttl = long(msg.ttl*1000) |
| enc, dec = get_codec(msg.content_type) |
| try: |
| body = enc(msg.content) |
| except AttributeError, e: |
| # convert to non-blocking EncodeError |
| raise EncodeError(e) |
| |
| # XXX: this is not safe for out of order, can this be triggered by pre_ack? |
| def msg_acked(): |
| # XXX: should we log the ack somehow too? |
| snd.acked += 1 |
| m = snd.session.outgoing.pop(0) |
| sst.outgoing_idx -= 1 |
| log.debug("RACK[%s]: %s", sst.session.log_id, msg) |
| assert msg == m |
| |
| xfr = MessageTransfer(destination=_snd._exchange, headers=(dp, mp), |
| payload=body) |
| |
| if _snd.pre_ack: |
| sst.write_cmd(xfr) |
| else: |
| sst.write_cmd(xfr, msg_acked, sync=msg._sync) |
| |
| log.debug("SENT[%s]: %s", sst.session.log_id, msg) |
| |
| if _snd.pre_ack: |
| msg_acked() |
| |
| def do_message_transfer(self, xfr): |
| sst = self.get_sst(xfr) |
| ssn = sst.session |
| |
| msg = self._decode(xfr) |
| rcv = sst.destinations[xfr.destination].target |
| msg._receiver = rcv |
| if rcv.closing or rcv.closed: # release message to a closing receiver |
| ids = RangedSet(*[msg._transfer_id]) |
| log.debug("releasing back %s message: %s, as receiver is closing", ids, msg) |
| sst.write_cmd(MessageRelease(ids, True)) |
| return |
| if rcv.impending is not UNLIMITED: |
| assert rcv.received < rcv.impending, "%s, %s" % (rcv.received, rcv.impending) |
| rcv.received += 1 |
| log.debug("RCVD[%s]: %s", ssn.log_id, msg) |
| ssn._notify_message_received(msg) |
| |
| |
| def _decode(self, xfr): |
| dp = EMPTY_DP |
| mp = EMPTY_MP |
| |
| for h in xfr.headers: |
| if isinstance(h, DeliveryProperties): |
| dp = h |
| elif isinstance(h, MessageProperties): |
| mp = h |
| |
| ap = mp.application_headers |
| enc, dec = get_codec(mp.content_type) |
| try: |
| content = dec(xfr.payload) |
| except Exception, e: |
| raise DecodeError(e) |
| msg = Message(content) |
| msg.id = mp.message_id |
| if ap is not None: |
| msg.subject = ap.get(SUBJECT) |
| msg.user_id = mp.user_id |
| if mp.reply_to is not None: |
| msg.reply_to = reply_to2addr(mp.reply_to) |
| msg.correlation_id = mp.correlation_id |
| if dp.delivery_mode is not None: |
| msg.durable = dp.delivery_mode == delivery_mode.persistent |
| msg.priority = dp.priority |
| if dp.ttl is not None: |
| msg.ttl = dp.ttl/1000.0 |
| msg.redelivered = dp.redelivered |
| msg.properties = mp.application_headers or {} |
| if mp.app_id is not None: |
| msg.properties["x-amqp-0-10.app-id"] = mp.app_id |
| if mp.content_encoding is not None: |
| msg.properties["x-amqp-0-10.content-encoding"] = mp.content_encoding |
| if dp.routing_key is not None: |
| msg.properties["x-amqp-0-10.routing-key"] = dp.routing_key |
| if dp.timestamp is not None: |
| msg.properties["x-amqp-0-10.timestamp"] = dp.timestamp |
| msg.content_type = mp.content_type |
| msg._transfer_id = xfr.id |
| return msg |
