#NEWLINE# def send( self , stream , msg_or_type , content = None , parent = None , ident = None , buffers = None , track = False , header = None , metadata = None ) : #NEWLINE# #INDENT# 'Build and send a message via stream or socket.\n\n        The message format used by this function internally is as follows:\n\n        [ident1,ident2,...,DELIM,HMAC,p_header,p_parent,p_content,\n         buffer1,buffer2,...]\n\n        The serialize/deserialize methods convert the nested message dict into this\n        format.\n\n        Parameters\n        ----------\n\n        stream : zmq.Socket or ZMQStream\n            The socket-like object used to send the data.\n        msg_or_type : str or Message/dict\n            Normally, msg_or_type will be a msg_type unless a message is being\n            sent more than once. If a header is supplied, this can be set to\n            None and the msg_type will be pulled from the header.\n\n        content : dict or None\n            The content of the message (ignored if msg_or_type is a message).\n        header : dict or None\n            The header dict for the message (ignored if msg_to_type is a message).\n        parent : Message or dict or None\n            The parent or parent header describing the parent of this message\n            (ignored if msg_or_type is a message).\n        ident : bytes or list of bytes\n            The zmq.IDENTITY routing path.\n        metadata : dict or None\n            The metadata describing the message\n        buffers : list or None\n            The already-serialized buffers to be appended to the message.\n        track : bool\n            Whether to track.  Only for use with Sockets, because ZMQStream\n            objects cannot track messages.\n\n\n        Returns\n        -------\n        msg : dict\n            The constructed message.\n        ' #NEWLINE# if ( not isinstance ( stream , zmq . Socket ) ) : #NEWLINE# #INDENT# track = False #NEWLINE# #UNINDENT# if isinstance ( msg_or_type , ( Message , dict ) ) : #NEWLINE# #INDENT# msg = msg_or_type #NEWLINE# buffers = ( buffers or msg . get ( 'buffers' , [ ] ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# msg = self . msg ( msg_or_type , content = content , parent = parent , header = header , metadata = metadata ) #NEWLINE# #UNINDENT# if ( self . check_pid and ( not ( os . getpid ( ) == self . pid ) ) ) : #NEWLINE# #INDENT# get_logger ( ) . warning ( 'WARNING: attempted to send message from fork\n%s' , msg ) #NEWLINE# return #NEWLINE# #UNINDENT# buffers = ( [ ] if ( buffers is None ) else buffers ) #NEWLINE# if self . adapt_version : #NEWLINE# #INDENT# msg = adapt ( msg , self . adapt_version ) #NEWLINE# #UNINDENT# to_send = self . serialize ( msg , ident ) #NEWLINE# to_send . extend ( buffers ) #NEWLINE# longest = max ( [ len ( s ) for s in to_send ] ) #NEWLINE# copy = ( longest < self . copy_threshold ) #NEWLINE# if ( buffers and track and ( not copy ) ) : #NEWLINE# #INDENT# tracker = stream . send_multipart ( to_send , copy = False , track = True ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# tracker = DONE #NEWLINE# stream . send_multipart ( to_send , copy = copy ) #NEWLINE# #UNINDENT# if self . debug : #NEWLINE# #INDENT# pprint . pprint ( msg ) #NEWLINE# pprint . pprint ( to_send ) #NEWLINE# pprint . pprint ( buffers ) #NEWLINE# #UNINDENT# msg [ 'tracker' ] = tracker #NEWLINE# return msg