session.py
716 lines
| 25.1 KiB
| text/x-python
|
PythonLexer
|
r4017 | """Session object for building, serializing, sending, and receiving messages in | ||
IPython. The Session object supports serialization, HMAC signatures, and | ||||
metadata on messages. | ||||
Also defined here are utilities for working with Sessions: | ||||
* A SessionFactory to be used as a base class for configurables that work with | ||||
Sessions. | ||||
* A Message object for convenience that allows attribute-access to the msg dict. | ||||
|
r4018 | |||
Authors: | ||||
* Min RK | ||||
* Brian Granger | ||||
* Fernando Perez | ||||
|
r4006 | """ | ||
#----------------------------------------------------------------------------- | ||||
# Copyright (C) 2010-2011 The IPython Development Team | ||||
# | ||||
# Distributed under the terms of the BSD License. The full license is in | ||||
# the file COPYING, distributed as part of this software. | ||||
#----------------------------------------------------------------------------- | ||||
#----------------------------------------------------------------------------- | ||||
# Imports | ||||
#----------------------------------------------------------------------------- | ||||
import hmac | ||||
|
r4007 | import logging | ||
|
r2597 | import os | ||
import pprint | ||||
|
r4006 | import uuid | ||
from datetime import datetime | ||||
try: | ||||
import cPickle | ||||
pickle = cPickle | ||||
except: | ||||
cPickle = None | ||||
import pickle | ||||
|
r2597 | |||
import zmq | ||||
|
r4006 | from zmq.utils import jsonapi | ||
|
r4007 | from zmq.eventloop.ioloop import IOLoop | ||
|
r4006 | from zmq.eventloop.zmqstream import ZMQStream | ||
|
r4017 | from IPython.config.configurable import Configurable, LoggingConfigurable | ||
|
r4006 | from IPython.utils.importstring import import_item | ||
|
r4008 | from IPython.utils.jsonutil import extract_dates, squash_dates, date_default | ||
|
r4734 | from IPython.utils.py3compat import str_to_bytes | ||
|
r4091 | from IPython.utils.traitlets import (CBytes, Unicode, Bool, Any, Instance, Set, | ||
|
r4735 | DottedObjectName, CUnicode) | ||
|
r4006 | |||
#----------------------------------------------------------------------------- | ||||
# utility functions | ||||
#----------------------------------------------------------------------------- | ||||
def squash_unicode(obj): | ||||
"""coerce unicode back to bytestrings.""" | ||||
if isinstance(obj,dict): | ||||
for key in obj.keys(): | ||||
obj[key] = squash_unicode(obj[key]) | ||||
if isinstance(key, unicode): | ||||
obj[squash_unicode(key)] = obj.pop(key) | ||||
elif isinstance(obj, list): | ||||
for i,v in enumerate(obj): | ||||
obj[i] = squash_unicode(v) | ||||
elif isinstance(obj, unicode): | ||||
obj = obj.encode('utf8') | ||||
return obj | ||||
#----------------------------------------------------------------------------- | ||||
# globals and defaults | ||||
#----------------------------------------------------------------------------- | ||||
|
r4008 | key = 'on_unknown' if jsonapi.jsonmod.__name__ == 'jsonlib' else 'default' | ||
json_packer = lambda obj: jsonapi.dumps(obj, **{key:date_default}) | ||||
|
r4035 | json_unpacker = lambda s: extract_dates(jsonapi.loads(s)) | ||
|
r2597 | |||
|
r4006 | pickle_packer = lambda o: pickle.dumps(o,-1) | ||
pickle_unpacker = pickle.loads | ||||
default_packer = json_packer | ||||
default_unpacker = json_unpacker | ||||
|
r4110 | DELIM=b"<IDS|MSG>" | ||
|
r4006 | |||
#----------------------------------------------------------------------------- | ||||
# Classes | ||||
#----------------------------------------------------------------------------- | ||||
|
r3269 | |||
|
r4017 | class SessionFactory(LoggingConfigurable): | ||
|
r4007 | """The Base class for configurables that have a Session, Context, logger, | ||
and IOLoop. | ||||
""" | ||||
logname = Unicode('') | ||||
def _logname_changed(self, name, old, new): | ||||
self.log = logging.getLogger(new) | ||||
# not configurable: | ||||
context = Instance('zmq.Context') | ||||
def _context_default(self): | ||||
return zmq.Context.instance() | ||||
session = Instance('IPython.zmq.session.Session') | ||||
loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False) | ||||
def _loop_default(self): | ||||
return IOLoop.instance() | ||||
def __init__(self, **kwargs): | ||||
super(SessionFactory, self).__init__(**kwargs) | ||||
if self.session is None: | ||||
# construct the session | ||||
self.session = Session(**kwargs) | ||||
|
r2597 | class Message(object): | ||
"""A simple message object that maps dict keys to attributes. | ||||
A Message can be created from a dict and a dict from a Message instance | ||||
simply by calling dict(msg_obj).""" | ||||
def __init__(self, msg_dict): | ||||
dct = self.__dict__ | ||||
|
r4006 | for k, v in dict(msg_dict).iteritems(): | ||
|
r2597 | if isinstance(v, dict): | ||
v = Message(v) | ||||
dct[k] = v | ||||
# Having this iterator lets dict(msg_obj) work out of the box. | ||||
def __iter__(self): | ||||
return iter(self.__dict__.iteritems()) | ||||
def __repr__(self): | ||||
return repr(self.__dict__) | ||||
def __str__(self): | ||||
return pprint.pformat(self.__dict__) | ||||
def __contains__(self, k): | ||||
return k in self.__dict__ | ||||
def __getitem__(self, k): | ||||
return self.__dict__[k] | ||||
|
r4006 | def msg_header(msg_id, msg_type, username, session): | ||
|
r4008 | date = datetime.now() | ||
|
r4006 | return locals() | ||
|
r2597 | |||
def extract_header(msg_or_header): | ||||
"""Given a message or header, return the header.""" | ||||
if not msg_or_header: | ||||
return {} | ||||
try: | ||||
# See if msg_or_header is the entire message. | ||||
h = msg_or_header['header'] | ||||
except KeyError: | ||||
try: | ||||
# See if msg_or_header is just the header | ||||
h = msg_or_header['msg_id'] | ||||
except KeyError: | ||||
raise | ||||
else: | ||||
h = msg_or_header | ||||
if not isinstance(h, dict): | ||||
h = dict(h) | ||||
return h | ||||
|
r4006 | class Session(Configurable): | ||
|
r4017 | """Object for handling serialization and sending of messages. | ||
The Session object handles building messages and sending them | ||||
with ZMQ sockets or ZMQStream objects. Objects can communicate with each | ||||
other over the network via Session objects, and only need to work with the | ||||
dict-based IPython message spec. The Session will handle | ||||
serialization/deserialization, security, and metadata. | ||||
Sessions support configurable serialiization via packer/unpacker traits, | ||||
and signing with HMAC digests via the key/keyfile traits. | ||||
Parameters | ||||
---------- | ||||
debug : bool | ||||
whether to trigger extra debugging statements | ||||
packer/unpacker : str : 'json', 'pickle' or import_string | ||||
importstrings for methods to serialize message parts. If just | ||||
'json' or 'pickle', predefined JSON and pickle packers will be used. | ||||
Otherwise, the entire importstring must be used. | ||||
The functions must accept at least valid JSON input, and output *bytes*. | ||||
For example, to use msgpack: | ||||
packer = 'msgpack.packb', unpacker='msgpack.unpackb' | ||||
pack/unpack : callables | ||||
You can also set the pack/unpack callables for serialization directly. | ||||
session : bytes | ||||
the ID of this Session object. The default is to generate a new UUID. | ||||
username : unicode | ||||
username added to message headers. The default is to ask the OS. | ||||
key : bytes | ||||
The key used to initialize an HMAC signature. If unset, messages | ||||
will not be signed or checked. | ||||
keyfile : filepath | ||||
The file containing a key. If this is set, `key` will be initialized | ||||
to the contents of the file. | ||||
""" | ||||
|
r4006 | debug=Bool(False, config=True, help="""Debug output in the Session""") | ||
|
r4017 | |||
|
r4055 | packer = DottedObjectName('json',config=True, | ||
|
r4006 | help="""The name of the packer for serializing messages. | ||
Should be one of 'json', 'pickle', or an import name | ||||
|
r4008 | for a custom callable serializer.""") | ||
|
r4006 | def _packer_changed(self, name, old, new): | ||
if new.lower() == 'json': | ||||
self.pack = json_packer | ||||
self.unpack = json_unpacker | ||||
elif new.lower() == 'pickle': | ||||
self.pack = pickle_packer | ||||
self.unpack = pickle_unpacker | ||||
|
r2606 | else: | ||
|
r4008 | self.pack = import_item(str(new)) | ||
|
r2597 | |||
|
r4055 | unpacker = DottedObjectName('json', config=True, | ||
|
r4006 | help="""The name of the unpacker for unserializing messages. | ||
Only used with custom functions for `packer`.""") | ||||
def _unpacker_changed(self, name, old, new): | ||||
if new.lower() == 'json': | ||||
self.pack = json_packer | ||||
self.unpack = json_unpacker | ||||
elif new.lower() == 'pickle': | ||||
self.pack = pickle_packer | ||||
self.unpack = pickle_unpacker | ||||
else: | ||||
|
r4008 | self.unpack = import_item(str(new)) | ||
|
r4006 | |||
|
r4735 | session = CUnicode(u'', config=True, | ||
|
r4006 | help="""The UUID identifying this session.""") | ||
def _session_default(self): | ||||
|
r4770 | u = unicode(uuid.uuid4()) | ||
self.bsession = u.encode('ascii') | ||||
return u | ||||
def _session_changed(self, name, old, new): | ||||
self.bsession = self.session.encode('ascii') | ||||
# bsession is the session as bytes | ||||
bsession = CBytes(b'') | ||||
|
r4017 | |||
|
r4234 | username = Unicode(os.environ.get('USER',u'username'), config=True, | ||
|
r4006 | help="""Username for the Session. Default is your system username.""") | ||
# message signature related traits: | ||||
|
r4091 | key = CBytes(b'', config=True, | ||
|
r4006 | help="""execution key, for extra authentication.""") | ||
def _key_changed(self, name, old, new): | ||||
if new: | ||||
self.auth = hmac.HMAC(new) | ||||
else: | ||||
self.auth = None | ||||
auth = Instance(hmac.HMAC) | ||||
digest_history = Set() | ||||
keyfile = Unicode('', config=True, | ||||
help="""path to file containing execution key.""") | ||||
def _keyfile_changed(self, name, old, new): | ||||
with open(new, 'rb') as f: | ||||
self.key = f.read().strip() | ||||
|
r2597 | |||
|
r4006 | pack = Any(default_packer) # the actual packer function | ||
def _pack_changed(self, name, old, new): | ||||
if not callable(new): | ||||
raise TypeError("packer must be callable, not %s"%type(new)) | ||||
|
r4008 | |||
|
r4006 | unpack = Any(default_unpacker) # the actual packer function | ||
def _unpack_changed(self, name, old, new): | ||||
|
r4008 | # unpacker is not checked - it is assumed to be | ||
|
r4006 | if not callable(new): | ||
|
r4008 | raise TypeError("unpacker must be callable, not %s"%type(new)) | ||
|
r4006 | |||
def __init__(self, **kwargs): | ||||
|
r4017 | """create a Session object | ||
Parameters | ||||
---------- | ||||
debug : bool | ||||
whether to trigger extra debugging statements | ||||
packer/unpacker : str : 'json', 'pickle' or import_string | ||||
importstrings for methods to serialize message parts. If just | ||||
'json' or 'pickle', predefined JSON and pickle packers will be used. | ||||
Otherwise, the entire importstring must be used. | ||||
The functions must accept at least valid JSON input, and output | ||||
*bytes*. | ||||
For example, to use msgpack: | ||||
packer = 'msgpack.packb', unpacker='msgpack.unpackb' | ||||
pack/unpack : callables | ||||
You can also set the pack/unpack callables for serialization | ||||
directly. | ||||
|
r4770 | session : unicode (must be ascii) | ||
|
r4017 | the ID of this Session object. The default is to generate a new | ||
UUID. | ||||
|
r4770 | bsession : bytes | ||
The session as bytes | ||||
|
r4017 | username : unicode | ||
username added to message headers. The default is to ask the OS. | ||||
key : bytes | ||||
The key used to initialize an HMAC signature. If unset, messages | ||||
will not be signed or checked. | ||||
keyfile : filepath | ||||
The file containing a key. If this is set, `key` will be | ||||
initialized to the contents of the file. | ||||
""" | ||||
|
r4006 | super(Session, self).__init__(**kwargs) | ||
|
r4008 | self._check_packers() | ||
|
r4006 | self.none = self.pack({}) | ||
|
r4770 | # ensure self._session_default() if necessary, so bsession is defined: | ||
self.session | ||||
|
r4006 | |||
@property | ||||
def msg_id(self): | ||||
"""always return new uuid""" | ||||
return str(uuid.uuid4()) | ||||
|
r4008 | def _check_packers(self): | ||
"""check packers for binary data and datetime support.""" | ||||
pack = self.pack | ||||
unpack = self.unpack | ||||
# check simple serialization | ||||
msg = dict(a=[1,'hi']) | ||||
try: | ||||
packed = pack(msg) | ||||
except Exception: | ||||
raise ValueError("packer could not serialize a simple message") | ||||
# ensure packed message is bytes | ||||
if not isinstance(packed, bytes): | ||||
raise ValueError("message packed to %r, but bytes are required"%type(packed)) | ||||
# check that unpack is pack's inverse | ||||
try: | ||||
unpacked = unpack(packed) | ||||
except Exception: | ||||
raise ValueError("unpacker could not handle the packer's output") | ||||
# check datetime support | ||||
msg = dict(t=datetime.now()) | ||||
try: | ||||
unpacked = unpack(pack(msg)) | ||||
except Exception: | ||||
self.pack = lambda o: pack(squash_dates(o)) | ||||
self.unpack = lambda s: extract_dates(unpack(s)) | ||||
|
r4006 | def msg_header(self, msg_type): | ||
return msg_header(self.msg_id, msg_type, self.username, self.session) | ||||
|
r4232 | def msg(self, msg_type, content=None, parent=None, subheader=None, header=None): | ||
|
r4220 | """Return the nested message dict. | ||
This format is different from what is sent over the wire. The | ||||
|
r4232 | serialize/unserialize methods converts this nested message dict to the wire | ||
format, which is a list of message parts. | ||||
|
r4220 | """ | ||
|
r2597 | msg = {} | ||
|
r4711 | header = self.msg_header(msg_type) if header is None else header | ||
msg['header'] = header | ||||
msg['msg_id'] = header['msg_id'] | ||||
msg['msg_type'] = header['msg_type'] | ||||
|
r2597 | msg['parent_header'] = {} if parent is None else extract_header(parent) | ||
msg['content'] = {} if content is None else content | ||||
|
r4006 | sub = {} if subheader is None else subheader | ||
msg['header'].update(sub) | ||||
|
r2597 | return msg | ||
|
r4220 | def sign(self, msg_list): | ||
"""Sign a message with HMAC digest. If no auth, return b''. | ||||
Parameters | ||||
---------- | ||||
msg_list : list | ||||
The [p_header,p_parent,p_content] part of the message list. | ||||
""" | ||||
|
r4006 | if self.auth is None: | ||
return b'' | ||||
h = self.auth.copy() | ||||
|
r4220 | for m in msg_list: | ||
|
r4006 | h.update(m) | ||
|
r4732 | return str_to_bytes(h.hexdigest()) | ||
|
r4008 | |||
|
r4006 | def serialize(self, msg, ident=None): | ||
|
r4017 | """Serialize the message components to bytes. | ||
|
r4220 | |||
|
r4231 | This is roughly the inverse of unserialize. The serialize/unserialize | ||
methods work with full message lists, whereas pack/unpack work with | ||||
the individual message parts in the message list. | ||||
|
r4220 | Parameters | ||
---------- | ||||
msg : dict or Message | ||||
The nexted message dict as returned by the self.msg method. | ||||
|
r4017 | Returns | ||
------- | ||||
|
r4220 | msg_list : list | ||
The list of bytes objects to be sent with the format: | ||||
[ident1,ident2,...,DELIM,HMAC,p_header,p_parent,p_content, | ||||
buffer1,buffer2,...]. In this list, the p_* entities are | ||||
the packed or serialized versions, so if JSON is used, these | ||||
|
r4733 | are utf8 encoded JSON strings. | ||
|
r4017 | """ | ||
|
r4006 | content = msg.get('content', {}) | ||
if content is None: | ||||
content = self.none | ||||
elif isinstance(content, dict): | ||||
content = self.pack(content) | ||||
elif isinstance(content, bytes): | ||||
# content is already packed, as in a relayed message | ||||
pass | ||||
elif isinstance(content, unicode): | ||||
# should be bytes, but JSON often spits out unicode | ||||
content = content.encode('utf8') | ||||
else: | ||||
raise TypeError("Content incorrect type: %s"%type(content)) | ||||
real_message = [self.pack(msg['header']), | ||||
self.pack(msg['parent_header']), | ||||
content | ||||
] | ||||
to_send = [] | ||||
if isinstance(ident, list): | ||||
# accept list of idents | ||||
to_send.extend(ident) | ||||
elif ident is not None: | ||||
to_send.append(ident) | ||||
to_send.append(DELIM) | ||||
signature = self.sign(real_message) | ||||
to_send.append(signature) | ||||
to_send.extend(real_message) | ||||
return to_send | ||||
|
r4233 | def send(self, stream, msg_or_type, content=None, parent=None, ident=None, | ||
|
r4232 | buffers=None, subheader=None, track=False, header=None): | ||
|
r4006 | """Build and send a message via stream or socket. | ||
|
r4220 | |||
The message format used by this function internally is as follows: | ||||
[ident1,ident2,...,DELIM,HMAC,p_header,p_parent,p_content, | ||||
buffer1,buffer2,...] | ||||
|
r4232 | The serialize/unserialize methods convert the nested message dict into this | ||
|
r4220 | format. | ||
|
r3271 | Parameters | ||
---------- | ||||
|
r4006 | |||
stream : zmq.Socket or ZMQStream | ||||
|
r4232 | The socket-like object used to send the data. | ||
|
r4006 | msg_or_type : str or Message/dict | ||
|
r4017 | Normally, msg_or_type will be a msg_type unless a message is being | ||
|
r4234 | sent more than once. If a header is supplied, this can be set to | ||
None and the msg_type will be pulled from the header. | ||||
|
r4006 | |||
content : dict or None | ||||
|
r4232 | The content of the message (ignored if msg_or_type is a message). | ||
header : dict or None | ||||
The header dict for the message (ignores if msg_to_type is a message). | ||||
|
r4006 | parent : Message or dict or None | ||
|
r4232 | The parent or parent header describing the parent of this message | ||
(ignored if msg_or_type is a message). | ||||
|
r4006 | ident : bytes or list of bytes | ||
|
r4232 | The zmq.IDENTITY routing path. | ||
|
r4006 | subheader : dict or None | ||
|
r4232 | Extra header keys for this message's header (ignored if msg_or_type | ||
is a message). | ||||
|
r4006 | buffers : list or None | ||
|
r4232 | The already-serialized buffers to be appended to the message. | ||
|
r4006 | track : bool | ||
|
r4232 | Whether to track. Only for use with Sockets, because ZMQStream | ||
objects cannot track messages. | ||||
|
r3271 | |||
Returns | ||||
------- | ||||
|
r4232 | msg : dict | ||
The constructed message. | ||||
(msg,tracker) : (dict, MessageTracker) | ||||
|
r4006 | if track=True, then a 2-tuple will be returned, | ||
the first element being the constructed | ||||
message, and the second being the MessageTracker | ||||
|
r3271 | """ | ||
|
r4006 | |||
if not isinstance(stream, (zmq.Socket, ZMQStream)): | ||||
raise TypeError("stream must be Socket or ZMQStream, not %r"%type(stream)) | ||||
elif track and isinstance(stream, ZMQStream): | ||||
raise TypeError("ZMQStream cannot track messages") | ||||
|
r3290 | if isinstance(msg_or_type, (Message, dict)): | ||
|
r4232 | # We got a Message or message dict, not a msg_type so don't | ||
# build a new Message. | ||||
|
r4006 | msg = msg_or_type | ||
|
r3269 | else: | ||
|
r4232 | msg = self.msg(msg_or_type, content=content, parent=parent, | ||
subheader=subheader, header=header) | ||||
|
r4006 | buffers = [] if buffers is None else buffers | ||
to_send = self.serialize(msg, ident) | ||||
flag = 0 | ||||
if buffers: | ||||
flag = zmq.SNDMORE | ||||
_track = False | ||||
else: | ||||
_track=track | ||||
if track: | ||||
tracker = stream.send_multipart(to_send, flag, copy=False, track=_track) | ||||
else: | ||||
tracker = stream.send_multipart(to_send, flag, copy=False) | ||||
for b in buffers[:-1]: | ||||
stream.send(b, flag, copy=False) | ||||
if buffers: | ||||
if track: | ||||
tracker = stream.send(buffers[-1], copy=False, track=track) | ||||
else: | ||||
tracker = stream.send(buffers[-1], copy=False) | ||||
|
r4220 | |||
|
r4006 | # omsg = Message(msg) | ||
if self.debug: | ||||
pprint.pprint(msg) | ||||
pprint.pprint(to_send) | ||||
pprint.pprint(buffers) | ||||
|
r3271 | |||
|
r4006 | msg['tracker'] = tracker | ||
|
r3271 | |||
|
r4006 | return msg | ||
|
r4232 | |||
|
r4220 | def send_raw(self, stream, msg_list, flags=0, copy=True, ident=None): | ||
|
r4006 | """Send a raw message via ident path. | ||
|
r4220 | |||
This method is used to send a already serialized message. | ||||
|
r4006 | Parameters | ||
---------- | ||||
|
r4220 | stream : ZMQStream or Socket | ||
The ZMQ stream or socket to use for sending the message. | ||||
msg_list : list | ||||
The serialized list of messages to send. This only includes the | ||||
[p_header,p_parent,p_content,buffer1,buffer2,...] portion of | ||||
the message. | ||||
ident : ident or list | ||||
A single ident or a list of idents to use in sending. | ||||
""" | ||||
|
r4006 | to_send = [] | ||
if isinstance(ident, bytes): | ||||
ident = [ident] | ||||
if ident is not None: | ||||
to_send.extend(ident) | ||||
|
r4232 | |||
|
r4006 | to_send.append(DELIM) | ||
|
r4220 | to_send.append(self.sign(msg_list)) | ||
to_send.extend(msg_list) | ||||
stream.send_multipart(msg_list, flags, copy=copy) | ||||
|
r4006 | |||
def recv(self, socket, mode=zmq.NOBLOCK, content=True, copy=True): | ||||
|
r4220 | """Receive and unpack a message. | ||
Parameters | ||||
---------- | ||||
socket : ZMQStream or Socket | ||||
The socket or stream to use in receiving. | ||||
Returns | ||||
------- | ||||
[idents], msg | ||||
[idents] is a list of idents and msg is a nested message dict of | ||||
same format as self.msg returns. | ||||
""" | ||||
|
r4006 | if isinstance(socket, ZMQStream): | ||
socket = socket.socket | ||||
|
r2597 | try: | ||
|
r4220 | msg_list = socket.recv_multipart(mode) | ||
|
r4006 | except zmq.ZMQError as e: | ||
|
r2597 | if e.errno == zmq.EAGAIN: | ||
# We can convert EAGAIN to None as we know in this case | ||||
|
r4006 | # recv_multipart won't return None. | ||
|
r3269 | return None,None | ||
|
r2597 | else: | ||
raise | ||||
|
r4017 | # split multipart message into identity list and message dict | ||
# invalid large messages can cause very expensive string comparisons | ||||
|
r4220 | idents, msg_list = self.feed_identities(msg_list, copy) | ||
|
r4006 | try: | ||
|
r4231 | return idents, self.unserialize(msg_list, content=content, copy=copy) | ||
|
r4006 | except Exception as e: | ||
# TODO: handle it | ||||
raise e | ||||
|
r4220 | def feed_identities(self, msg_list, copy=True): | ||
"""Split the identities from the rest of the message. | ||||
Feed until DELIM is reached, then return the prefix as idents and | ||||
remainder as msg_list. This is easily broken by setting an IDENT to DELIM, | ||||
|
r4017 | but that would be silly. | ||
|
r4006 | |||
Parameters | ||||
---------- | ||||
|
r4220 | msg_list : a list of Message or bytes objects | ||
The message to be split. | ||||
|
r4006 | copy : bool | ||
flag determining whether the arguments are bytes or Messages | ||||
Returns | ||||
------- | ||||
|
r4231 | (idents, msg_list) : two lists | ||
idents will always be a list of bytes, each of which is a ZMQ | ||||
identity. msg_list will be a list of bytes or zmq.Messages of the | ||||
form [HMAC,p_header,p_parent,p_content,buffer1,buffer2,...] and | ||||
should be unpackable/unserializable via self.unserialize at this | ||||
point. | ||||
|
r4006 | """ | ||
if copy: | ||||
|
r4220 | idx = msg_list.index(DELIM) | ||
return msg_list[:idx], msg_list[idx+1:] | ||||
|
r4006 | else: | ||
failed = True | ||||
|
r4220 | for idx,m in enumerate(msg_list): | ||
|
r4006 | if m.bytes == DELIM: | ||
failed = False | ||||
break | ||||
if failed: | ||||
|
r4220 | raise ValueError("DELIM not in msg_list") | ||
idents, msg_list = msg_list[:idx], msg_list[idx+1:] | ||||
return [m.bytes for m in idents], msg_list | ||||
|
r4006 | |||
|
r4231 | def unserialize(self, msg_list, content=True, copy=True): | ||
"""Unserialize a msg_list to a nested message dict. | ||||
This is roughly the inverse of serialize. The serialize/unserialize | ||||
methods work with full message lists, whereas pack/unpack work with | ||||
the individual message parts in the message list. | ||||
|
r4006 | Parameters: | ||
----------- | ||||
|
r4231 | msg_list : list of bytes or Message objects | ||
The list of message parts of the form [HMAC,p_header,p_parent, | ||||
p_content,buffer1,buffer2,...]. | ||||
|
r4006 | content : bool (True) | ||
|
r4231 | Whether to unpack the content dict (True), or leave it packed | ||
(False). | ||||
|
r4006 | copy : bool (True) | ||
|
r4231 | Whether to return the bytes (True), or the non-copying Message | ||
object in each place (False). | ||||
Returns | ||||
------- | ||||
msg : dict | ||||
The nested message dict with top-level keys [header, parent_header, | ||||
content, buffers]. | ||||
|
r4006 | """ | ||
minlen = 4 | ||||
message = {} | ||||
if not copy: | ||||
for i in range(minlen): | ||||
|
r4220 | msg_list[i] = msg_list[i].bytes | ||
|
r4006 | if self.auth is not None: | ||
|
r4220 | signature = msg_list[0] | ||
|
r4522 | if not signature: | ||
raise ValueError("Unsigned Message") | ||||
|
r4006 | if signature in self.digest_history: | ||
raise ValueError("Duplicate Signature: %r"%signature) | ||||
self.digest_history.add(signature) | ||||
|
r4220 | check = self.sign(msg_list[1:4]) | ||
|
r4006 | if not signature == check: | ||
raise ValueError("Invalid Signature: %r"%signature) | ||||
|
r4220 | if not len(msg_list) >= minlen: | ||
|
r4006 | raise TypeError("malformed message, must have at least %i elements"%minlen) | ||
|
r4711 | header = self.unpack(msg_list[1]) | ||
message['header'] = header | ||||
message['msg_id'] = header['msg_id'] | ||||
message['msg_type'] = header['msg_type'] | ||||
|
r4220 | message['parent_header'] = self.unpack(msg_list[2]) | ||
|
r4006 | if content: | ||
|
r4220 | message['content'] = self.unpack(msg_list[3]) | ||
|
r3269 | else: | ||
|
r4220 | message['content'] = msg_list[3] | ||
|
r3269 | |||
|
r4220 | message['buffers'] = msg_list[4:] | ||
|
r4006 | return message | ||
|
r2597 | |||
def test_msg2obj(): | ||||
am = dict(x=1) | ||||
ao = Message(am) | ||||
assert ao.x == am['x'] | ||||
am['y'] = dict(z=1) | ||||
ao = Message(am) | ||||
assert ao.y.z == am['y']['z'] | ||||
k1, k2 = 'y', 'z' | ||||
assert ao[k1][k2] == am[k1][k2] | ||||
am2 = dict(ao) | ||||
assert am['x'] == am2['x'] | ||||
assert am['y']['z'] == am2['y']['z'] | ||||
|
r4017 | |||