Process a stream type message.
337975
def handle_stream(msg):
    """"""
    if not msg.gateway.is_sensor(msg.node_id):
        return None
    stream = msg.gateway.const.Stream(msg.sub_type)
    handler = stream.get_handler(msg.gateway.handlers)
    if handler is None:
        return None
    return handler(msg)
1123935
def start_stream(self, stream_type, **stream_parameters):
        """"""
        if stream_type:
            self.subscribe(stream_type, **stream_parameters)
        else:
            self.handle_error("You need to set a stream type")
1174699
async def publish(self, message):
        """"""
        try:
            self.write('data: {}\n\n'.format(message))
            await self.flush()
        except StreamClosedError:
            self.finished = True
487552
def receive_message(self):
        """"""
        with self.lock:
            assert self.can_receive_messages()
            message_type = self._read_message_type(self._file)
            message = message_type(self._file, self)
            self._message_received(message)
1114773
def parse(self, stream):
        """"""
        (result, _) = (self << eof).parse_partial(stream)
        return result
215312
def process_stream_line(line)
      return if @mutex.synchronize { @subscriptions.any? { |subscription| subscription.process_stream_data line } }
      return if process_send_message_outcome line

      warn "Lightstreamer: unprocessed stream data '#{line}'"
    end
404369
def _process_message(self, message: amqp.Message) -> None:
        """"""
        if self.shutdown_pending.is_set():
            return

        try:
            if isinstance(message.body, bytes):
                message.body = message.body.decode()
            description = json.loads(message.body)
        except Exception:
            logger.error("Cannot decode message. Dropping. Message: %r", message.body)
            traceback.print_exc()
            message.channel.basic_reject(message.delivery_tag, requeue=False)
        else:
            logger.info("Processing task: %r", description)
            self._process_description(message, description)
1021450
function onMessage(message) {
    if (message.type != "utf8") return

    DEBUGMSG("recv: " + message.utf8Data)
    message = JSON.parse(message.utf8Data)

    if      (message.type == "request")  session.onRequest(channel, message)
    else if (message.type == "response") session.onResponse(channel, message)
    else if (message.type == "event")    session.onEvent(channel, message)
  }
343548
def handle_stream(self, stream, address):
        '''
        
        '''
        log.trace('Req client %s connected', address)
        self.clients.append((stream, address))
        unpacker = msgpack.Unpacker()
        try:
            while True:
                wire_bytes = yield stream.read_bytes(4096, partial=True)
                unpacker.feed(wire_bytes)
                for framed_msg in unpacker:
                    if six.PY3:
                        framed_msg = salt.transport.frame.decode_embedded_strs(
                            framed_msg
                        )
                    header = framed_msg['head']
                    self.io_loop.spawn_callback(self.message_handler, stream, header, framed_msg['body'])

        except StreamClosedError:
            log.trace('req client disconnected %s', address)
            self.clients.remove((stream, address))
        except Exception as e:
            log.trace('other master-side exception: %s', e)
            self.clients.remove((stream, address))
            stream.close()
244301
function processMessageNode (message) {
      if (
        (message.type === 'Literal' && typeof message.value === 'string' && !pattern.test(message.value)) ||
        (message.type === 'TemplateLiteral' && message.quasis.length === 1 && !pattern.test(message.quasis[0].value.cooked))
      ) {
        context.report({
          node: message,
          message: "Report message does not match the pattern '{{pattern}}'.",
          data: { pattern: context.options[0] || '' },
        });
      }
    }
490561
def receive_message(self, message, data):
        
        if data[MESSAGE_TYPE] == TYPE_STATUS:
            self._process_status(data.get("data"))
            return True

        return False
632917
public static void process(InputStream inputStream, StreamHandler handler) throws IOException {
    LOGGER.trace("process(InputStream, StreamHandler)");
    process(inputStream, handler, new byte[BUFFER_SIZE]);
  }
243173
protected void finishCompoundMessage ()
    {
        if (--_compoundDepth == 0) {
            CompoundDownstreamMessage downstream = _compound;
            _compound = null;
            if (!downstream.msgs.isEmpty()) {
                postMessage(downstream, null);
            }
        }
    }
812900
void processAck(Message msg) {
        PubAck pa;
        Exception ex = null;
        try {
            pa = PubAck.parseFrom(msg.getData());
        } catch (InvalidProtocolBufferException e) {
            // If we are speaking to a server we don't understand, let the
            // user know.
            System.err.println("Protocol error: " +  e.getStackTrace());
            return;
        }

        // Remove
        AckClosure ackClosure = removeAck(pa.getGuid());
        if (ackClosure != null) {
            // Capture error if it exists.
            String ackError = pa.getError();

            if (ackClosure.ah != null) {
                if (!ackError.isEmpty()) {
                    ex = new IOException(ackError);
                }
                // Perform the ackHandler callback
                ackClosure.ah.onAck(pa.getGuid(), ex);
            } else if (ackClosure.ch != null) {
                try {
                    ackClosure.ch.put(ackError);
                } catch (InterruptedException e) {
                    // ignore
                }
            }
        }
    }
506199
def process_message(self, peer, mailfrom, rcpttos, data, **kwargs):
        """"""
        inheaders = 1
        lines = data.split('\n')
        logger.info('---------- MESSAGE FOLLOWS ----------')
        for line in lines:
            # headers first
            if inheaders and not line:
                logger.info('X-Peer: %s' % peer[0])
                inheaders = 0
            logger.info(line)
        logger.info('------------ END MESSAGE ------------')
28401
@Override
    public void handleMessage(final Message message) throws Fault {
        final InputStream is = message.getContent(InputStream.class);
        if (WireTapHelper.isMessageContentToBeLogged(message, logMessageContent, logMessageContentOverride)) {
            if (null == is) {
                Reader reader = message.getContent(Reader.class);
                if (null != reader) {
                    String encoding = (String) message.get(Message.ENCODING);
                    if (encoding == null) {
                        encoding = "UTF-8";
                    }
                    try {
                        final CachedOutputStream cos = new CachedOutputStream();
                        final Writer writer = new OutputStreamWriter(cos, encoding);
                        IOUtils.copy(reader, writer, 1024);
                        reader.reset();
                        writer.flush();
                        message.setContent(InputStream.class, cos.getInputStream());
                        message.setContent(Reader.class, null);
                        message.setContent(CachedOutputStream.class, cos);
                        closeCachedOutputStream(message, cos);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    } finally {
                        try {
                            reader.reset();
                        } catch (IOException e) {
                        }
                    }
                }
            } else {
                try {
                    final CachedOutputStream cos = new CachedOutputStream();
                    // TODO: We should try to make this streaming
                    //WireTapInputStream wtis = new WireTapInputStream(is, cos);
                    //message.setContent(InputStream.class, wtis);
                    IOUtils.copyAndCloseInput(is, cos);
                    message.setContent(InputStream.class, cos.getInputStream());
                    message.setContent(CachedOutputStream.class, cos);
                    closeCachedOutputStream(message, cos);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }else {
            try {
                final CachedOutputStream cos = new CachedOutputStream();
                cos.write(WireTapHelper.CONTENT_LOGGING_IS_DISABLED.getBytes(Charset.forName("UTF-8")));
                message.setContent(CachedOutputStream.class, cos);
                message.setContent(InputStream.class, is);
                closeCachedOutputStream(message, cos);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }
148606
def process_message(self, message):
        
        try:
            self.logger.debug("Received message %s with id %r.", message, message.message_id)
            self.broker.emit_before("process_message", message)

            res = None
            if not message.failed:
                actor = self.broker.get_actor(message.actor_name)
                res = actor(*message.args, **message.kwargs)

            self.broker.emit_after("process_message", message, result=res)

        except SkipMessage:
            self.logger.warning("Message %s was skipped.", message)
            self.broker.emit_after("skip_message", message)

        except BaseException as e:
            # Stuff the exception into the message [proxy] so that it
            # may be used by the stub broker to provide a nicer
            # testing experience.
            message.stuff_exception(e)

            if isinstance(e, RateLimitExceeded):
                self.logger.warning("Rate limit exceeded in message %s: %s.", message, e)
            else:
                self.logger.warning("Failed to process message %s with unhandled exception.", message, exc_info=True)

            self.broker.emit_after("process_message", message, exception=e)

        finally:
            # NOTE: There is no race here as any message that was
            # processed must have come off of a consumer.  Therefore,
            # there has to be a consumer for that message's queue so
            # this is safe.  Probably.
            self.consumers[message.queue_name].post_process_message(message)
            self.work_queue.task_done()