def handle_stream(self, stream):
        '''
        Override this to handle the streams as they arrive

        :param IOStream stream: An IOStream for processing

        See https://tornado.readthedocs.io/en/latest/iostream.html#tornado.iostream.IOStream
        for additional details.
        '''
        @tornado.gen.coroutine
        def _null(msg):
            raise tornado.gen.Return(None)

        def write_callback(stream, header):
            if header.get('mid'):
                @tornado.gen.coroutine
                def return_message(msg):
                    pack = salt.transport.frame.frame_msg_ipc(
                        msg,
                        header={'mid': header['mid']},
                        raw_body=True,
                    )
                    yield stream.write(pack)
                return return_message
            else:
                return _null
        if six.PY2:
            encoding = None
        else:
            encoding = 'utf-8'
        unpacker = msgpack.Unpacker(encoding=encoding)
        while not stream.closed():
            try:
                wire_bytes = yield stream.read_bytes(4096, partial=True)
                unpacker.feed(wire_bytes)
                for framed_msg in unpacker:
                    body = framed_msg['body']
                    self.io_loop.spawn_callback(self.payload_handler, body, write_callback(stream, framed_msg['head']))
            except tornado.iostream.StreamClosedError:
                log.trace('Client disconnected from IPC %s', self.socket_path)
                break
            except socket.error as exc:
                # On occasion an exception will occur with
                # an error code of 0, it's a spurious exception.
                if exc.errno == 0:
                    log.trace('Exception occured with error number 0, '
                              'spurious exception: %s', exc)
                else:
                    log.error('Exception occurred while '
                              'handling stream: %s', exc)
            except Exception as exc:
                log.error('Exception occurred while '
                          'handling stream: %s', exc)
def handle_stream ( self , stream ) : @ tornado . gen . coroutine def _null ( msg ) : raise tornado . gen . Return ( None ) def write_callback ( stream , header ) : if header . get ( 'mid' ) : @ tornado . gen . coroutine def return_message ( msg ) : pack = salt . transport . frame . frame_msg_ipc ( msg , header = { 'mid' : header [ 'mid' ] } , raw_body = True , ) yield stream . write ( pack ) return return_message else : return _null if six . PY2 : encoding = None else : encoding = 'utf-8' unpacker = msgpack . Unpacker ( encoding = encoding ) while not stream . closed ( ) : try : wire_bytes = yield stream . read_bytes ( 4096 , partial = True ) unpacker . feed ( wire_bytes ) for framed_msg in unpacker : body = framed_msg [ 'body' ] self . io_loop . spawn_callback ( self . payload_handler , body , write_callback ( stream , framed_msg [ 'head' ] ) ) except tornado . iostream . StreamClosedError : log . trace ( 'Client disconnected from IPC %s' , self . socket_path ) break except socket . error as exc : if exc . errno == 0 : log . trace ( 'Exception occured with error number 0, ' 'spurious exception: %s' , exc ) else : log . error ( 'Exception occurred while ' 'handling stream: %s' , exc ) except Exception as exc : log . error ( 'Exception occurred while ' 'handling stream: %s' , exc )
Override this to handle the streams as they arrive

        :param IOStream stream: An IOStream for processing

        See https://tornado.readthedocs.io/en/latest/iostream.html#tornado.iostream.IOStream
        for additional details.