Reads and delivers as many messages to the listener as possible.
770119
private void deliver() {
        // We can have reentrancy here when using a direct executor, triggered by calls to
        // request more messages. This is safe as we simply loop until pendingDelivers = 0
        if (inDelivery) {
            return;
        }
        inDelivery = true;
        try {
            // Process the uncompressed bytes.
            while (pendingDeliveries > 0 && hasRequiredBytes()) {
                switch (state) {
                    case HEADER:
                        readHeader();
                        break;
                    case BODY:
                        // Read the body and deliver the message.
                        readBody();

                        // Since we've delivered a message, decrement the number of pending
                        // deliveries remaining.
                        pendingDeliveries--;
                        break;
                    default:
                        throw new IllegalStateException("Invalid state: " + state);
                }
            }

            /*
             * We are stalled when there are no more bytes to process. This allows delivering errors as
             * soon as the buffered input has been consumed, independent of whether the application
             * has requested another message.  At this point in the function, either all frames have been
             * delivered, or unprocessed is empty.  If there is a partial message, it will be inside next
             * frame and not in unprocessed.  If there is extra data but no pending deliveries, it will
             * be in unprocessed.
             */
            if (closeWhenComplete && isStalled()) {
                close();
            }
        } finally {
            inDelivery = false;
        }
    }
1012641
public static void consumeMessage(BigEndianBinaryReader in) throws IOException {
        FieldInfo fieldInfo;
        while ((fieldInfo = readFieldInfo(in)) != null) {
            readFieldValue(in, fieldInfo, null, false);
        }
    }
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);
    }
464119
@Override
    public void processMessage() {
        if (overwrite)
            storage.setArray(key, payload);
        else if (!storage.arrayExists(key))
            storage.setArray(key, payload);
    }
610658
def stream_n_messages(n):
    
    response = get_dict("url", "args", "headers", "origin")
    n = min(n, 100)

    def generate_stream():
        for i in range(n):
            response["id"] = i
            yield json.dumps(response) + "\n"

    return Response(generate_stream(), headers={"Content-Type": "application/json"})
1261740
async def __run_listen(self):
        
        while True:
            message = await ZMQUtils.recv(self.__backend_socket)
            await self.__handle_backend_message(message)
191384
def ProcessMessage(self, message):
    """"""
    self.ProcessResponse(message.source.Basename(), message.payload)
854875
public void deliverTell (ClientObject caller, UserMessage message,
                             Name target, ChatService.TellListener listener)
        throws InvocationException
    {
        // we just forward the message as if it originated on this server
        _chatprov.deliverTell(message, target, listener);
    }
1079177
def write(self, data):
        
        
        # Do the write
        self.backingStream.write(data)
        
        for listener in self.writeListeners:
            # Send out notifications
            listener(len(data))
339398
function() {
      // send data to console anyway...
      if(prevReader) prevReader(receivedData);
      receivedData = "";
      // start the previous reader listening again
      Espruino.Core.Serial.startListening(prevReader);
      // call our callback
      if (callback) callback(hadToBreak);
    }
663061
private void reads() {
    int index = random.nextInt();
    for (;;) {
      Integer key = ints[index++ & MASK];
      cache.get(key);
      calls.increment();
    }
  }
1082468
private void deliver() {
    try {
      deliverUnsafe();
    } catch (Throwable t) {
      // This should never happen. If does, it means we are in an inconsistent state and should
      // close the stream and further processing should be prevented. This is accomplished by
      // purposefully leaving the lock non-zero and notifying the outerResponseObserver of the
      // error. Care must be taken to avoid calling close twice in case the first invocation threw
      // an error.
      innerController.cancel();
      if (!finished) {
        outerResponseObserver.onError(t);
      }
    }
  }
788543
public function receive()
    {
        if ($this->isWritable()) {
            throw new \RuntimeException('Cannot read from writable message');
        }

        $this->readable = true;

        while ($content = fread($this->resource, $this->chunkSize)) {
            $this->content .= $content;
            $this->emit('data', [$content]);
            if (strpos($content, $this->getTerminateString()) !== false) {
                $this->emit('end', [$this]);
            }
        }
    }
1122487
def QueueMessages(self, messages):
    """"""
    # Push all the messages to our input queue
    for message in messages:
      self._in_queue.put(message, block=True)
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 = []
379677
def consume(self, limit=None):
        """"""
        for total_message_count in count():
            if limit and total_message_count >= limit:
                raise StopIteration

            if not self.channel.is_open:
                raise StopIteration

            self.channel.conn.drain_events()
            yield True
271002
async def listen(self):
        ""
        if self.subscribed:
            return self.handle_message(await self.parse_response(block=True))