CONTRACT: message should be delivered to the caller of MessageReceiver.receive() only via Poll on prefetchqueue
979950
private Message pollPrefetchQueue() {
        final Message message = this.prefetchedMessages.poll();
        if (message != null) {
            // message lastReceivedOffset should be up-to-date upon each poll - as recreateLink will depend on this
            this.lastReceivedMessage = message;
            this.sendFlow(1);
        }

        return message;
    }
1153165
private Message receive(TestContext context) {
        Endpoint messageEndpoint = getOrCreateEndpoint(context);
        return receiveTimeout > 0 ? messageEndpoint.createConsumer().receive(context, receiveTimeout) :
                messageEndpoint.createConsumer().receive(context, messageEndpoint.getEndpointConfiguration().getTimeout());
    }
1349123
@Override
	public final Message receive(long timeout) throws JMSException
    {
        if (messageListener != null)
            throw new FFMQException("Cannot receive messages while a listener is active","INVALID_OPERATION"); 
        
        AbstractMessage message = receiveFromDestination(timeout,true);
        if (message != null)
        {
        	message.ensureDeserializationLevel(MessageSerializationLevel.FULL);
        	
            message.setSession(session);
            
            // Auto acknowledge message
            if (autoAcknowledge)
                session.acknowledge();
        }
        return message;
    }
275330
@Override
    public void messageReceived(NextFilter nextFilter, S session,
            AmqpSecureMessage message) throws Exception {
        nextFilter.messageReceived(session, message); 
    }
38672
@Override
    public void deliver(VoltMessage message)
    {
        if (!(message instanceof Iv2RepairLogResponseMessage)) {
            return;
        }
        Iv2RepairLogResponseMessage response = (Iv2RepairLogResponseMessage)message;
        if (response.getRequestId() != m_requestId) {
            if (repairLogger.isTraceEnabled()) {
                repairLogger.trace(m_whoami + "rejecting stale repair response."
                        + " Current request id is: " + m_requestId
                        + " Received response for request id: " + response.getRequestId());
            }
            return;
        }

        ReplicaRepairStruct rrs = m_replicaRepairStructs.get(response.m_sourceHSId);
        if (rrs.m_expectedResponses < 0 ) {
            if (repairLogger.isDebugEnabled()) {
                repairLogger.debug(m_whoami + "collecting " + response.getOfTotal()
                + " repair log entries from "
                + CoreUtils.hsIdToString(response.m_sourceHSId));
            }
        }
        // Long.MAX_VALUE has rejoin semantics
        if (response.getHandle() != Long.MAX_VALUE) {
            m_maxSeenTxnId = Math.max(m_maxSeenTxnId, response.getHandle());
        }

        if (response.getPayload() != null) {
            m_repairLogUnion.add(response);
            if (repairLogger.isDebugEnabled()) {
                repairLogger.debug(m_whoami + " collected from " + CoreUtils.hsIdToString(response.m_sourceHSId) +
                        ", message: " + response.getPayload());
            }
        }
        if (rrs.update(response)) {
            if (repairLogger.isDebugEnabled()) {
                repairLogger.debug(m_whoami + "collected " + rrs.m_receivedResponses
                        + " responses for " + rrs.m_expectedResponses
                        + " repair log entries from " + CoreUtils.hsIdToString(response.m_sourceHSId));
            }
            if (areRepairLogsComplete()) {

                //no repair needed for MigratePartitionLeader
                if (m_isMigratePartitionLeader) {
                    m_promotionResult.set(new RepairResult(m_maxSeenTxnId));
                } else {
                    repairSurvivors();
                }
            }
        }
    }
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);
    }
684670
public ReceiveMessageResult receiveMessageSync(ReceiveMessageRequest rq) {
        Future<ReceiveMessageResult> future = receiveMessage(rq, null);
        return waitForFuture(future);
    }
907061
void deliverSingleDestinationMessage(Message msg, MessageID messageID) {
        synchronized (deliverySet) {
            long sequenceNumber = sequenceNumberManager.get();
            MessageInfo messageInfo = new MessageInfo(messageID, msg, sequenceNumber);
            messageInfo.updateAndMarkReadyToDeliver(sequenceNumber);
            deliverySet.add(messageInfo);
            notifyIfNeeded();
        }
    }
1190835
function postMessage(channelState, messageJson) {
  var writePromise = writeMessage(channelState.channelName, channelState.uuid, messageJson, channelState.paths);
  channelState.writeBlockPromise = channelState.writeBlockPromise.then(
  /*#__PURE__*/
  _asyncToGenerator(
  /*#__PURE__*/
  _regeneratorRuntime.mark(function _callee3() {
    var _ref4, msgObj, pingStr, writeToReadersPromise;

    return _regeneratorRuntime.wrap(function _callee3$(_context3) {
      while (1) {
        switch (_context3.prev = _context3.next) {
          case 0:
            _context3.next = 2;
            return new Promise(function (res) {
              return setTimeout(res, 0);
            });

          case 2:
            _context3.next = 4;
            return Promise.all([writePromise, refreshReaderClients(channelState)]);

          case 4:
            _ref4 = _context3.sent;
            msgObj = _ref4[0];
            emitOverFastPath(channelState, msgObj, messageJson);
            pingStr = '{"t":' + msgObj.time + ',"u":"' + msgObj.uuid + '","to":"' + msgObj.token + '"}|';
            writeToReadersPromise = Promise.all(Object.values(channelState.otherReaderClients).filter(function (client) {
              return client.writable;
            }) // client might have closed in between
            .map(function (client) {
              return new Promise(function (res) {
                client.write(pingStr, res);
              });
            }));
            /**
             * clean up old messages
             * to not waste resources on cleaning up,
             * only if random-int matches, we clean up old messages
             */

            if (randomInt(0, 20) === 0) {
              /* await */
              getAllMessages(channelState.channelName, channelState.paths).then(function (allMessages) {
                return cleanOldMessages(allMessages, channelState.options.node.ttl);
              });
            }

            return _context3.abrupt("return", writeToReadersPromise);

          case 11:
          case "end":
            return _context3.stop();
        }
      }
    }, _callee3);
  })));
  return channelState.writeBlockPromise;
}
24219
@Override
	public void receivePong(Tree message) {

		// Verify protocol version
		if (checkVersion) {
			String ver = message.get("ver", "unknown");
			if (!PROTOCOL_VERSION.equals(ver)) {
				logger.warn("Invalid protocol version (" + ver + ")!");
				return;
			}
		}

		// Get response's unique ID
		String id = message.get("id", (String) null);
		if (id == null || id.isEmpty()) {
			logger.warn("Missing \"id\" property!", message);
			return;
		}

		// Get stored promise
		PendingPromise pending = promises.remove(id);
		if (pending == null) {
			logger.warn("Unknown (maybe timeouted) response received!", message);
			return;
		}

		// Resolve Promise
		pending.promise.complete(message);
	}
179515
@Override
    public void deliver(VoltMessage message)
    {
        if (message instanceof Iv2RepairLogResponseMessage) {
            Iv2RepairLogResponseMessage response = (Iv2RepairLogResponseMessage)message;
            if (response.getRequestId() != m_requestId) {
                if (repairLogger.isTraceEnabled()) {
                    repairLogger.trace(m_whoami + "rejecting stale repair response."
                            + " Current request id is: " + m_requestId
                            + " Received response for request id: " + response.getRequestId());
                }
                return;
            }

            // Step 1: if the msg has a known (not MAX VALUE) handle, update m_maxSeen.
            if (response.getTxnId() != Long.MAX_VALUE) {
                m_maxSeenTxnId = Math.max(m_maxSeenTxnId, response.getTxnId());
            }

            // Step 2: track hashinator versions

            if (response.hasHashinatorConfig()) {
                Pair<Long,byte[]> proposed = response.getHashinatorVersionedConfig();
                if (proposed.getFirst() > m_newestHashinatorConfig.getFirst()) {
                    m_newestHashinatorConfig = proposed;
                }
            }

            // Step 3: offer to the union
            addToRepairLog(response);
            if (repairLogger.isDebugEnabled()) {
                repairLogger.debug(m_whoami + " collected from " + CoreUtils.hsIdToString(response.m_sourceHSId) +
                        ", message: " + response.getPayload());
            }

            // Step 4: update the corresponding replica repair struct.
            ReplicaRepairStruct rrs = m_replicaRepairStructs.get(response.m_sourceHSId);
            if (rrs.m_expectedResponses < 0) {
                if (repairLogger.isDebugEnabled()) {
                    repairLogger.debug(m_whoami + "collecting " + response.getOfTotal()
                    + " repair log entries from "
                    + CoreUtils.hsIdToString(response.m_sourceHSId));
                }
            }

            if (rrs.update(response)) {
                if (repairLogger.isDebugEnabled()) {
                    repairLogger.debug(m_whoami + "collected " + rrs.m_receivedResponses
                            + " responses for " + rrs.m_expectedResponses
                            + " repair log entries from " + CoreUtils.hsIdToString(response.m_sourceHSId));
                }

                if (areRepairLogsComplete()) {

                    TheHashinator.updateHashinator(ElasticHashinator.class,
                            m_newestHashinatorConfig.getFirst(), m_newestHashinatorConfig.getSecond(), true);

                    //no real transaction repair when triggered with MigratePartitionLeader. Theoretically it should not be here.
                    //MigratePartitionLeader should not trigger MP promotion.
                    if (m_isMigratePartitionLeader) {
                        m_promotionResult.set(new RepairResult(m_maxSeenTxnId));
                    } else {
                        repairSurvivors();
                    }
                }
            }
        }
    }
673580
private void sendPending() {
    while (outstanding.get() < concurrency) {
      final TopicQueue queue = pendingTopics.poll();
      if (queue == null) {
        return;
      }

      queue.pending = false;
      final int sent = queue.sendBatch();

      // Did we send a whole batch? Then there might be more messages in the queue. Mark as pending again.
      if (sent == batchSize) {
        queue.pending = true;
        pendingTopics.offer(queue);
      }
    }
  }
1082273
private Message receiveSelected(TestContext context, String selectorString) {
        if (log.isDebugEnabled()) {
            log.debug("Setting message selector: '" + selectorString + "'");
        }

        Endpoint messageEndpoint = getOrCreateEndpoint(context);
        Consumer consumer = messageEndpoint.createConsumer();
        if (consumer instanceof SelectiveConsumer) {
            if (receiveTimeout > 0) {
                return ((SelectiveConsumer) messageEndpoint.createConsumer()).receive(
                        context.replaceDynamicContentInString(selectorString),
                        context, receiveTimeout);
            } else {
                return ((SelectiveConsumer) messageEndpoint.createConsumer()).receive(
                        context.replaceDynamicContentInString(selectorString),
                        context, messageEndpoint.getEndpointConfiguration().getTimeout());
            }
        } else {
            log.warn(String.format("Unable to receive selective with consumer implementation: '%s'", consumer.getClass()));
            return receive(context);
        }
    }
750268
def _recv_ack(self, method_frame):
        ''''''
        if self._ack_listener:
            delivery_tag = method_frame.args.read_longlong()
            multiple = method_frame.args.read_bit()
            if multiple:
                while self._last_ack_id < delivery_tag:
                    self._last_ack_id += 1
                    self._ack_listener(self._last_ack_id)
            else:
                self._last_ack_id = delivery_tag
                self._ack_listener(self._last_ack_id)
147572
public void startMessagePreFetchTask() {
		try {
			boolean updateMC = isMessageCenterInForeground();
			fetchAndStoreMessages(updateMC, false, null);
		} catch (final Exception e) {
			ApptentiveLog.w(MESSAGES, e, "Unhandled Exception thrown from fetching new message task");
			logException(e);
		}
	}
793954
def subscribe(
            self,
            subject="",
            queue="",
            cb=None,
            future=None,
            max_msgs=0,
            is_async=False,
            pending_msgs_limit=DEFAULT_SUB_PENDING_MSGS_LIMIT,
            pending_bytes_limit=DEFAULT_SUB_PENDING_BYTES_LIMIT,
    ):
        
        if self.is_closed:
            raise ErrConnectionClosed

        if self.is_draining:
            raise ErrConnectionDraining

        self._ssid += 1
        sid = self._ssid
        sub = Subscription(
            subject=subject,
            queue=queue,
            cb=cb,
            future=future,
            max_msgs=max_msgs,
            is_async=is_async,
            sid=sid,
        )
        self._subs[sid] = sub

        if cb is not None:
            sub.pending_msgs_limit = pending_msgs_limit
            sub.pending_bytes_limit = pending_bytes_limit
            sub.pending_queue = tornado.queues.Queue(
                maxsize=pending_msgs_limit)

            @tornado.gen.coroutine
            def wait_for_msgs():
                while True:
                    sub = wait_for_msgs.sub
                    err_cb = wait_for_msgs.err_cb

                    try:
                        sub = wait_for_msgs.sub
                        if sub.closed:
                            break

                        msg = yield sub.pending_queue.get()
                        if msg is None:
                            break

                        sub.received += 1
                        sub.pending_size -= len(msg.data)
                        if sub.max_msgs > 0 and sub.received >= sub.max_msgs:
                            # If we have hit the max for delivered msgs, remove sub.
                            self._subs.pop(sub.sid, None)
                            self._remove_subscription(sub)

                        # Invoke depending of type of handler.
                        if sub.is_async:
                            # NOTE: Deprecate this usage in a next release,
                            # the handler implementation ought to decide
                            # the concurrency level at which the messages
                            # should be processed.
                            self._loop.spawn_callback(sub.cb, msg)
                        else:
                            yield sub.cb(msg)
                    except Exception as e:
                        # All errors from calling an async subscriber
                        # handler are async errors.
                        if err_cb is not None:
                            yield err_cb(e)

            # Bind the subscription and error cb if present
            wait_for_msgs.sub = sub
            wait_for_msgs.err_cb = self._error_cb
            self._loop.spawn_callback(wait_for_msgs)

        elif future is not None:
            # Used to handle the single response from a request
            # based on auto unsubscribe.
            sub.future = future

        # Send SUB command...
        sub_cmd = b''.join([
            SUB_OP, _SPC_,
            sub.subject.encode(), _SPC_,
            sub.queue.encode(), _SPC_, ("%d" % sid).encode(), _CRLF_
        ])
        yield self.send_command(sub_cmd)
        yield self._flush_pending()
        raise tornado.gen.Return(sid)
355089
@Override
    public Message receive() throws JMSRuntimeException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "receive");

        try {
            return receive(0);
        } finally {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "receive");
        }
    }