Not yet processed messages - for resiliency
180198
function() {
    var collection = [];
    for (let id of this.store._keys) {
      if (id.startsWith("unprocessed")) {
        collection.push(this.get(id));
      }
    }
    return Promise.resolve(collection);
  }
1365634
def received_message(self, msg):
        
        logger.debug("Received message: %s", msg)
        if msg.is_binary:
            raise ValueError("Binary messages not supported")
        resps = json.loads(msg.data)

        cmd_group = _get_cmds_id(*resps)
        if cmd_group:
            (cmds, promise) = self._cmd_groups[cmd_group]
            promise.fulfill((cmds, resps))
        else:
            try:
                self.received_unsolicited(resps)
            except:
                logger.exception("Error in unsolicited msg handler")
                raise
1202252
protected void notifyAllPendingSenders(String message) {
        for (ChannelFuture future : _futureMap.values()) {
            future.setFailure(new Error(message));
        }
    }
1301186
def _on_status_message(self, sequence, topic, message):
        

        self._logger.debug("Received message on (topic=%s): %s" % (topic, message))

        try:
            conn_key = self._find_connection(topic)
        except ArgumentError:
            self._logger.warn("Dropping message that does not correspond with a known connection, message=%s", message)
            return

        if messages.ConnectionResponse.matches(message):
            if self.name != message['client']:
                self._logger.debug("Connection response received for a different client, client=%s, name=%s", message['client'], self.name)
                return

            self.conns.finish_connection(conn_key, message['success'], message.get('failure_reason', None))
        else:
            self._logger.warn("Dropping message that did not correspond with a known schema, message=%s", message)
419180
private void doFail(long id) {
    if (log.isDebugEnabled()) {
      log.debug(String.format("%s - Received resend request for messages starting at %d", this, id));
    }

    // Ack all the entries before the given ID.
    doAck(id);

    // Now that all the entries before the given ID have been removed,
    // just iterate over the messages map and resend all the messages.
    Iterator<Map.Entry<Long, JsonObject>> iter = messages.entrySet().iterator();
    while (iter.hasNext()) {
      eventBus.send(inAddress, iter.next().getValue());
    }
  }
859989
def post_message(self, msg):
        ''''''
        if '_posted' in msg.__dict__:
            return
        msg._posted = True
        msg._timestamp = time.time()
        type = msg.get_type()
        if type != 'HEARTBEAT' or (msg.type != mavlink.MAV_TYPE_GCS and msg.type != mavlink.MAV_TYPE_GIMBAL):
            self.messages[type] = msg

        if 'usec' in msg.__dict__:
            self.uptime = msg.usec * 1.0e-6
        if 'time_boot_ms' in msg.__dict__:
            self.uptime = msg.time_boot_ms * 1.0e-3

        if self._timestamp is not None:
            if self.notimestamps:
                msg._timestamp = self.uptime
            else:
                msg._timestamp = self._timestamp

        src_system = msg.get_srcSystem()
        src_component = msg.get_srcComponent()
        src_tuple = (src_system, src_component)
        radio_tuple = (ord('3'), ord('D'))
        if not (src_tuple == radio_tuple or msg.get_type() == 'BAD_DATA'):
            if not src_tuple in self.last_seq:
                last_seq = -1
            else:
                last_seq = self.last_seq[src_tuple]
            seq = (last_seq+1) % 256
            seq2 = msg.get_seq()
            if seq != seq2 and last_seq != -1:
                diff = (seq2 - seq) % 256
                self.mav_loss += diff
                #print("lost %u seq=%u seq2=%u last_seq=%u src_system=%u %s" % (diff, seq, seq2, last_seq, src_system, msg.get_type()))
            self.last_seq[src_tuple] = seq2
            self.mav_count += 1
        
        self.timestamp = msg._timestamp
        if type == 'HEARTBEAT' and msg.get_srcComponent() != mavlink.MAV_COMP_ID_GIMBAL:
            self.target_system = msg.get_srcSystem()
            self.target_component = msg.get_srcComponent()
            if float(mavlink.WIRE_PROTOCOL_VERSION) >= 1 and msg.type != mavlink.MAV_TYPE_GCS:
                self.flightmode = mode_string_v10(msg)
                self.mav_type = msg.type
                self.base_mode = msg.base_mode
        elif type == 'PARAM_VALUE':
            s = str(msg.param_id)
            self.params[str(msg.param_id)] = msg.param_value
            if msg.param_index+1 == msg.param_count:
                self.param_fetch_in_progress = False
                self.param_fetch_complete = True
        elif type == 'SYS_STATUS' and mavlink.WIRE_PROTOCOL_VERSION == '0.9':
            self.flightmode = mode_string_v09(msg)
        elif type == 'GPS_RAW':
            if self.messages['HOME'].fix_type < 2:
                self.messages['HOME'] = msg
        elif type == 'GPS_RAW_INT':
            if self.messages['HOME'].fix_type < 3:
                self.messages['HOME'] = msg
        for hook in self.message_hooks:
            hook(self, msg)

        if (msg.get_signed() and
            self.mav.signing.link_id == 0 and
            msg.get_link_id() != 0 and
            self.target_system == msg.get_srcSystem() and
            self.target_component == msg.get_srcComponent()):
            # change to link_id from incoming packet
            self.mav.signing.link_id = msg.get_link_id()
740420
def on_message(self, message):
        """"""
        json_rpc = json.loads(message)
        logging.log(logging.DEBUG, json_rpc)

        if self.pool is None:
            self.pool = multiprocessing.Pool(processes=args.workers)

        # Spawn a process to protect the server against segfaults
        async = self.pool.apply_async(_worker_process, [json_rpc])
        try:
            result = async.get(timeout=args.timeout)
            error = 0
        except multiprocessing.TimeoutError:
            result = ("File format conversion timed out! This is due "
                      "either to a large input file or a segmentation "
                      "fault in the underlying open babel library.")
            error = 1
            self.pool.terminate()
            self.pool = multiprocessing.Pool(processes=args.workers)
        except Exception:
            result = traceback.format_exc()
            error = 1
        logging.log(logging.DEBUG, result)

        self.write_message(json.dumps({'result': result, 'error': error,
                                       'id': json_rpc['id']},
                                      separators=(',', ':')))
476622
private void deliverReadyTxns() {
        // First, pull all the sequenced messages, if any.
        VoltMessage m = m_replaySequencer.poll();
        while(m != null) {
            deliver(m);
            m = m_replaySequencer.poll();
        }
        // Then, try to pull all the drainable messages, if any.
        m = m_replaySequencer.drain();
        while (m != null) {
            if (m instanceof Iv2InitiateTaskMessage) {
                // Send IGNORED response for all SPs
                Iv2InitiateTaskMessage task = (Iv2InitiateTaskMessage) m;
                final InitiateResponseMessage response = new InitiateResponseMessage(task);
                response.setResults(new ClientResponseImpl(ClientResponse.UNEXPECTED_FAILURE,
                            new VoltTable[0],
                            ClientResponseImpl.IGNORED_TRANSACTION));
                m_mailbox.send(response.getInitiatorHSId(), response);
            }
            m = m_replaySequencer.drain();
        }
    }
630695
def _got_message(self, uid, text):
        
        #TODO: use try
        e = Event(uid, E_MESSAGE, text)
        self.queue.put_nowait(e)

        self.controllers[uid][2] = time.time()
196966
private void updateReadyness()
	{
		StringBuffer _buffer = new StringBuffer();
		if (null == myMessageProvider)
		{
			_buffer.append(NO_MESSAGE_PROVIDER);
		}
		if (null == myIconProvider)
		{
			if (_buffer.length() > 0)
			{
				_buffer.append(COMMA);
			}
			_buffer.append(NO_ICON_PROVIDER);
		}
		if (_buffer.length() > 0)
		{
			_buffer.append(DOT);
		}
		myReasonWhyNotReady = _buffer.toString();
		myIsReady = (_buffer.length() == 0);

	}
720177
async def wait_message(self):
        """"""
        if not self._queue.empty():
            return True
        if self._queue.closed:
            return False
        await self._queue.wait()
        return self.is_active
420533
function processWaitingPublishPromises( self, newPromiseResolves ) {
   const waitingResolves = self.waitingPromiseResolves_;
   self.waitingPromiseResolves_ = newPromiseResolves;

   waitingResolves.forEach( resolve => resolve() );

   if( self.eventQueue_.length === 0 ) {
      // nothing was queued by any subscriber. The publishers can instantly be notified of delivery.
      newPromiseResolves.forEach( resolve => resolve() );
      self.waitingPromiseResolves_ = [];
   }
}
829108
def ProcessMessage(self, message):
    """"""
    # Only accept authenticated messages
    if (message.auth_state !=
        rdf_flows.GrrMessage.AuthorizationState.AUTHENTICATED):
      return

    now = time.time()

    # Maintain a cache of the foreman
    with self.lock:
      if (self.foreman_cache is None or
          now > self.foreman_cache.age + self.cache_refresh_time):
        self.foreman_cache = aff4.FACTORY.Open(
            "aff4:/foreman", mode="rw", token=self.token)
        self.foreman_cache.age = now

    if message.source:
      self.foreman_cache.AssignTasksToClient(message.source.Basename())
1082851
def process_message_notification(request, messages_path):
    """"""
    if not messages_path:
        return

    global _MESSAGES_CACHE
    global _MESSAGES_MTIME

    # NOTE (lhcheng): Cache the processed messages to avoid parsing
    # the files every time. Check directory modification time if
    # reload is necessary.
    if (_MESSAGES_CACHE is None or
            _MESSAGES_MTIME != os.path.getmtime(messages_path)):
        _MESSAGES_CACHE = _get_processed_messages(messages_path)
        _MESSAGES_MTIME = os.path.getmtime(messages_path)

    for msg in _MESSAGES_CACHE:
        msg.send_message(request)
1009130
def message_handlers(self):
        """"""
        index = lib.EnvGetNextDefmessageHandler(self._env, self._cls, 0)

        while index != 0:
            yield MessageHandler(self._env, self._cls, index)

            index = lib.EnvGetNextDefmessageHandler(self._env, self._cls, index)
1226467
def process_next_message(self, timeout):
        """"""
        message = self.worker_manager.receive(timeout)

        if isinstance(message, Acknowledgement):
            self.task_manager.task_start(message.task, message.worker)
        elif isinstance(message, Result):
            self.task_manager.task_done(message.task, message.result)
231693
def process_messages_loop(self):
        
        self.receiving_messages = True
        try:
            self.process_messages_loop_internal()
        except pika.exceptions.ConnectionClosed as ex:
            logging.error("Connection closed {}.".format(ex))
            raise