The gathering browse will browse messages from the local itemstream as
a priority. If none are found it will move through the other queue points
until it finds a message.
511717
public JsMessage next() throws SIResourceException, SISessionDroppedException 
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "next");
    
    JsMessage msg = null;
    Iterator<BrowseCursor> it = cursors.iterator();
    while(it.hasNext() && msg==null)
      msg = it.next().next();     
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "next");
    
    return msg;
  }
1039357
public boolean seekToBeginning(String consumerGroupId, String topic) {
        KafkaMsgConsumer consumer = getKafkaConsumer(consumerGroupId, false);
        return consumer.seekToBeginning(topic);
    }
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;
    }
721895
def gather(self):
        
        limit = self.op_sequence - self.op_sequence_start
        results = MessageIterator(self.sqs, self.reduce_queue, limit)
        for m in results:
            # sequence_id from above
            msg_id = int(m['MessageAttributes']['sequence_id']['StringValue'])
            if (not msg_id > self.op_sequence_start or not msg_id <= self.op_sequence or
            msg_id not in self.futures):
                raise RuntimeError(
                    "Concurrent queue user from different "
                    "process or previous results")
            f = self.futures[msg_id]
            f.set_result(m)
            results.ack(m)
1315056
def _peek_async_request(self, msg_id, msg_name):
        
        assert get_thread_ident() == self.ioloop_thread_id
        if msg_id is None:
            msg_id = self._msg_id_for_name(msg_name)
        if msg_id in self._async_queue:
            return self._async_queue[msg_id]
        else:
            return None, None, None, None, None
1313228
public void handleIncomingMessage(byte[] data, int offset, int len) {
		synchronized(messageQueue) {
			messageQueue.add(new IncomingMessage(data, offset, len));
		}
		synchronized(lock) {
			lock.notifyAll();
		}
	}
696952
@Override
	public void onFetchFinish(MessageManager manager, List<ApptentiveMessage> messages) {
		if (isPolling()) {
			long pollingInterval = getPollingInterval();
			ApptentiveLog.v(MESSAGES, "Scheduled polling messages in %d sec", pollingInterval / 1000);
			dispatchOnConversationQueueOnce(messagePollingTask, pollingInterval);
		}
	}
935902
public void callASTVisitors(JavacAST ast, long priority) {
		for (VisitorContainer container : visitorHandlers) try {
			if (container.getPriority() == priority) ast.traverse(container.visitor);
		} catch (Throwable t) {
			javacError(String.format("Lombok visitor handler %s failed", container.visitor.getClass()), t);
		}
	}
343795
def gather(self):
        
        if self.state in constants.RECEIVE_STATES:
            raise TypeError("Only new messages can be gathered.")
        if not self._message:
            raise ValueError("Message data already consumed.")
        try:
            raise self._response
        except TypeError:
            pass
        return [self]
31796
@Override
    public ItemStream findFirstMatching(Filter filter) throws MessageStoreException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "findFirstMatching", filter);

        if (_rootMembership != null)
        {
            ItemStream item = _rootMembership.findFirstMatchingItemStream(filter);

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "findFirstMatching", item);
            return item;
        }
        else
        {
            MessageStoreUnavailableException msue;
            if (!_startupExceptions.isEmpty())
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled())
                    SibTr.event(this, tc, "Operation not possible as MessageStore failed to start!");
                msue = new MessageStoreUnavailableException("Operation not possible as MessageStore failed to start!", _startupExceptions.get(0));
            }
            else
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled())
                    SibTr.event(this, tc, "Operation not possible as MessageStore is unavailable!");
                msue = new MessageStoreUnavailableException("Operation not possible as MessageStore is unavailable!");
            }

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "findFirstMatching");
            throw msue;
        }
    }
357890
private void seek(List<TopicPartition> seekList) {
        for (TopicPartition tp : seekList) {
            AtomicLong lastCommittedOffset = m_lastCommittedOffSets.get().get(tp);
            if (lastCommittedOffset != null && lastCommittedOffset.get() > -1L) {
                AtomicLong lastSeeked = m_lastSeekedOffSets.get(tp);

                //eliminate duplicate seek
                if (lastSeeked != null && lastSeeked.get() == lastCommittedOffset.get()) {
                    continue;
                }
                m_consumer.seek(tp, lastCommittedOffset.longValue());
                m_lastSeekedOffSets.put(tp, new AtomicLong(lastCommittedOffset.get()));

                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("Moves offset for group " +  m_config.getGroupId() + " -" + tp + " to " + lastCommittedOffset);
                }
            }
        }
    }
772632
@Override
    public T next(long systemCurrentTimeMillis) {
        // drain all the waiting messages from the source (up to 10k)
        while (delayed.size() < 10000) {
            T event = source.next(systemCurrentTimeMillis);
            if (event == null) {
                break;
            }
            transformAndQueue(event, systemCurrentTimeMillis);
        }

        return delayed.nextReady(systemCurrentTimeMillis);
    }
475664
private void _fetch(BlockingQueue<ConsumerRecord<String, byte[]>> buffer, String topic,
            long waitTime, TimeUnit waitTimeUnit) {
        KafkaConsumer<String, byte[]> consumer = _getConsumer(topic);
        synchronized (consumer) {
            _checkAndSubscribe(consumer, topic);
            Set<String> subscription = consumer.subscription();
            ConsumerRecords<String, byte[]> crList = subscription != null
                    && subscription.contains(topic) ? consumer.poll(waitTimeUnit.toMillis(waitTime))
                            : null;
            if (crList != null) {
                for (ConsumerRecord<String, byte[]> cr : crList) {
                    buffer.offer(cr);
                }
            }
        }
    }
382289
public void updateLowestMessagePriority(short messagePriority)
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "updateLowestMessagePriority", messagePriority);

      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Joined resource:", joinedResource);

      if (joinedResource != null)
      {
         joinedResource.updateLowestMessagePriority(messagePriority);
      }
      else
      {
         super.updateLowestMessagePriority(messagePriority);
      }

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "updateLowestMessagePriority");
   }
422628
public void onMessagesRead(Peer peer, long fromDate) {

        // Filtering obsolete read events
        if (fromDate < getLastReadDate(peer)) {
            return;
        }

        // Removing read messages from pending storage
        getNotifications().clear();
        pendingStorage.setMessagesCount(0);
        pendingStorage.setDialogsCount(0);
        allPendingNotifications.clear();
        saveStorage();
        updateNotification();

        // Setting last read date
        setLastReadDate(peer, fromDate);
    }
536919
def next_message(self):
        
        if self._is_running:
            raise Exception('Event loop already running')
        if self._pending_messages:
            return self._pending_messages.popleft()
        self._async_session.run(self._enqueue_request_and_stop,
                                self._enqueue_notification_and_stop)
        if self._pending_messages:
            return self._pending_messages.popleft()
513119
@Override
	public AbstractMessage nextElement()
	{
		if (endOfQueueReached)
			throw new NoSuchElementException();
			
		try
		{
			checkNotClosed();
			AbstractMessage msg = fetchNext();
			if (msg != null)
			{
				nextMessage = null; // Consume fetched message

				// Make sure the message is fully deserialized and marked as read-only
				msg.ensureDeserializationLevel(MessageSerializationLevel.FULL);
				msg.markAsReadOnly();
				
				return msg;
			}
			
			throw new NoSuchElementException();
		}
		catch (NoSuchElementException e)
		{
			throw e;
		}
		catch (JMSException e)
		{
			throw new IllegalStateException(e.toString());
		}
	}