see all of these messages and control their transmission.
2142
public void handleInitiateResponseMessage(InitiateResponseMessage message)
    {
        final VoltTrace.TraceEventBatch traceLog = VoltTrace.log(VoltTrace.Category.MPI);
        if (traceLog != null) {
            traceLog.add(() -> VoltTrace.endAsync("initmp", message.getTxnId()));
        }

        DuplicateCounter counter = m_duplicateCounters.get(message.getTxnId());

        // A transaction may be routed back here for EveryPartitionTask via leader migration
        if (counter != null && message.isMisrouted()) {
            tmLog.info("The message on the partition is misrouted. TxnID: " + TxnEgo.txnIdToString(message.getTxnId()));
            Long newLeader = m_leaderMigrationMap.get(message.m_sourceHSId);
            if (newLeader != null) {
                // Update the DuplicateCounter with new replica
                counter.updateReplica(message.m_sourceHSId, newLeader);
                m_leaderMigrationMap.remove(message.m_sourceHSId);

                // Leader migration has updated the leader, send the request to the new leader
                m_mailbox.send(newLeader, counter.getOpenMessage());
            } else {
                // Leader migration not done yet.
                m_mailbox.send(message.m_sourceHSId, counter.getOpenMessage());
            }
            return;
        }

        if (counter != null) {
            int result = counter.offer(message);
            if (result == DuplicateCounter.DONE) {
                m_duplicateCounters.remove(message.getTxnId());
                // Only advance the truncation point on committed transactions that sent fragments to SPIs.
                // See ENG-4211 & ENG-14563
                if (message.shouldCommit() && message.haveSentMpFragment()) {
                    m_repairLogTruncationHandle = m_repairLogAwaitingCommit;
                    m_repairLogAwaitingCommit = message.getTxnId();
                }
                m_outstandingTxns.remove(message.getTxnId());

                m_mailbox.send(counter.m_destinationId, message);
            }
            else if (result == DuplicateCounter.MISMATCH) {
                VoltDB.crashLocalVoltDB("HASH MISMATCH running every-site system procedure.", true, null);
            } else if (result == DuplicateCounter.ABORT) {
                VoltDB.crashLocalVoltDB("PARTIAL ROLLBACK/ABORT running every-site system procedure.", true, null);
            }
            // doing duplicate suppresion: all done.
        }
        else {
            // Only advance the truncation point on committed transactions that sent fragments to SPIs.
            if (message.shouldCommit() && message.haveSentMpFragment()) {
                m_repairLogTruncationHandle = m_repairLogAwaitingCommit;
                m_repairLogAwaitingCommit = message.getTxnId();
            }
            MpTransactionState txn = (MpTransactionState)m_outstandingTxns.remove(message.getTxnId());
            assert(txn != null);
            // the initiatorHSId is the ClientInterface mailbox. Yeah. I know.
            m_mailbox.send(message.getInitiatorHSId(), message);
            // We actually completed this MP transaction.  Create a fake CompleteTransactionMessage
            // to send to our local repair log so that the fate of this transaction is never forgotten
            // even if all the masters somehow die before forwarding Complete on to their replicas.
            CompleteTransactionMessage ctm = new CompleteTransactionMessage(m_mailbox.getHSId(),
                    message.m_sourceHSId, message.getTxnId(), message.isReadOnly(), 0,
                    !message.shouldCommit(), false, false, false, txn.isNPartTxn(),
                    message.m_isFromNonRestartableSysproc, false);
            ctm.setTruncationHandle(m_repairLogTruncationHandle);
            // dump it in the repair log
            // hacky castage
            ((MpInitiatorMailbox)m_mailbox).deliverToRepairLog(ctm);
        }
    }
841244
def allow(self, channel, message):
		
		return all(
			filter(channel, message)
			for filter in _load_filters()
		)
601727
def listen(self):
        
        with Consumer(self.connection, queues=self.queue, on_message=self.on_message,
                      auto_declare=False):
            for _ in eventloop(self.connection, timeout=1, ignore_timeouts=True):
                    pass
45816
def consume(self, limit=None):
        """"""
        for total_message_count in count():
            if limit and total_message_count >= limit:
                raise StopIteration
            self.drain_events()
            yield True
938990
def _send(self, messages):
        """"""
        if len(messages) == 1:
            to_send = self._build_message(messages[0])
            if to_send is False:
                # The message was missing recipients.
                # Bail.
                return False
        else:
            pm_messages = list(map(self._build_message, messages))
            pm_messages = [m for m in pm_messages if m]
            if len(pm_messages) == 0:
                # If after filtering, there aren't any messages
                # to send, bail.
                return False
            to_send = PMBatchMail(messages=pm_messages)
        try:
            to_send.send(test=self.test_mode)
        except:
            if self.fail_silently:
                return False
            raise
        return True
1074975
def watch(self):
        """"""
        return bool(lib.EnvGetDefmessageHandlerWatch(
            self._env, self._cls, self._idx))
11163
function(message) {
    var pack = {
      toMaster: true,
      toWorkers: false,
      toSource: false,
      message: message,
      source: process.pid
    }
    dispatch(pack)
  }
490561
def receive_message(self, message, data):
        
        if data[MESSAGE_TYPE] == TYPE_STATUS:
            self._process_status(data.get("data"))
            return True

        return False
1224205
void broadcast(Iterator<String> peers, Message message) {
    transport.broadcast(peers, message);
  }
831358
def _read_message(self):
        
        job = gevent.spawn(super(GConnection, self)._read_message)
        job.join()
1223841
def send_messages(self, messages):
		""""""
		msg_count = 0
		for message in messages:  # .message() triggers header validation
			message.message()
			msg_count += 1
		mail.outbox.extend(messages)
		return msg_count
577977
def listen(self):
        ''''''
        try:
            self._pubsub.subscribe(self._channels)
            for message in self._pubsub.listen():
                if message['type'] == 'message':
                    yield message
        finally:
            self._channels = []
231587
def on_messages(self, msg_list):
        
        self.stats.on_pack_recv(len(msg_list))

        for msg in msg_list:
            if self.state == OPEN:
                self.conn.on_message(msg)
305048
def control(self, key):
        """"""
        if not self.connection:
            raise exceptions.ConnectionClosed()

        payload = b"\x00\x00\x00" + self._serialize_string(key)
        packet = b"\x00\x00\x00" + self._serialize_string(payload, True)

        logging.info("Sending control command: %s", key)
        self.connection.send(packet)
        self._read_response()
        time.sleep(self._key_interval)
8701
def listen(self):
        """"""
        while True:
            message = self.pull.recv()
            logger.debug("received message of length %d" % len(message))
            uuid, message = message[:32], message[32:]
            response = uuid + self.handle(message)
            self.push.send(response)
436218
public void deliverTell (BodyObject target, UserMessage message)
    {
        SpeakUtil.sendMessage(target, message);

        // note that the teller "heard" what they said
        SpeakUtil.noteMessage(target, message.speaker, message);
    }
853808
function message(msg, rinfo) {
    self._open = true;
    msg = self._readEncoding ?
      msg.toString(self._readEncoding) :
      msg;

    self.emit('data', msg, {
      port: rinfo.port,
      host: rinfo.address
    },
    rinfo.family.toLowerCase());
  }