When it is called it sends all previously queued messages.
513023
public function flush(): void
    {
        while (false == $this->events->isEmpty()) {
            list($topic, $message) = $this->events->dequeue();

            $this->realProducer->sendEvent($topic, $message);
        }

        while (false == $this->commands->isEmpty()) {
            list($command, $message) = $this->commands->dequeue();

            $this->realProducer->sendCommand($command, $message);
        }
    }
1042510
def queue_purge(self, queue, **kwargs):
        """"""
        return self.channel.queue_purge(queue=queue).message_count
206813
def send_messages(self, messages):
        """"""
        if not messages:
            return
        self._lock.acquire()
        try:
            # The try-except is nested to allow for
            # Python 2.4 support (Refs #12147)
            try:
                stream_created = self.open()
                for message in messages:
                    self.stream.write(render_message(message))
                    self.stream.write('\n')
                    self.stream.write('-'*79)
                    self.stream.write('\n')
                    self.stream.flush()  # flush after each message
                if stream_created:
                    self.close()
            except:
                if not self.fail_silently:
                    raise
        finally:
            self._lock.release()
        return len(messages)
540736
private void ack() {
    // Send a message to the other side of the connection indicating the
    // last message that we received in order. This will allow it to
    // purge messages we've already received from its queue.
    if (open && connected) {
      if (log.isDebugEnabled()) {
        log.debug(String.format("%s - Acking messages up to: %d", this, lastReceived));
      }
      eventBus.send(outAddress, new JsonObject().putString("action", "ack").putNumber("id", lastReceived));
      lastFeedbackTime = System.currentTimeMillis();
    }
  }
128040
function(id, msg, result) {
        var q = queues[id], sc;
        if (q && q.onEnd) {
            sc = q.context || q;
            q.onEnd.call(sc, _returnData(q, msg, result));
        }
    }
394242
protected void flushQueuedMessages(IoSession session, AttachedSessionManager attachedSessionManager) {
        Queue<Object> messageQueue = getMessageQueue(session);
        if (messageQueue != null) {
            flushQueuedMessages(messageQueue, session, attachedSessionManager);
            // Note: leave messageQueue intact on session to avoid race condition
            // will be cleared by next received message anyway
        }
    }
916803
def flush_all
      # It can take a long time to flush all of the messages
      # on a server, so we'll set the read timeout to something
      # much higher than usual.
      connection.with_additional_read_timeout(60) do
        resp = send_recv( KJess::Request::FlushAll.new )
        return KJess::Response::End === resp
      end
    end
1074133
def previous(self):
        
        msg = cr.Message()
        msg.type = cr.PREVIOUS
        self.send_message(msg)
1007826
async def _handle_state_forgotten(self, message):
        
        self._send_queue.append(RequestState(MsgsStateInfo(
            req_msg_id=message.msg_id, info=chr(1) * len(message.obj.msg_ids)),
            loop=self._loop))
1236010
def requeue(self, time_ms=0, backoff=True):
        
        if self._has_responded:
            raise NSQException('already responded')
        self._has_responded = True
        self.on_requeue.send(self, timeout=time_ms, backoff=backoff)
358807
def _sendQueued(self):
        """"""
        for tReq in list(self.requests.values()):  # must copy, may del
            if not tReq.sent:
                self._sendRequest(tReq)
592254
def flush_queue(self):
        """"""
        records = []

        while not self.queue.empty() and len(records) < self.batch_size:
            records.append(self.queue.get())

        if records:
            self.send_records(records)
            self.last_flush = time.time()
1092599
public function sendNextMessage()
    {
        if (count($this->outQueue) > 0) {
            $msgnode = array_shift($this->outQueue);
            $msgnode->refreshTimes();
            $this->lastId = $msgnode->getAttribute('id');
            $this->sendNode($msgnode);
        } else {
            $this->lastId = false;
        }
    }
766502
@GuardedBy("lock") protected void sendBundledMessages() {
        for(Map.Entry<Address,List<Message>> entry: msgs.entrySet()) {
            List<Message> list=entry.getValue();
            if(list.isEmpty())
                continue;

            output.position(0);
            if(list.size() == 1)
                sendSingleMessage(list.get(0));
            else {
                Address dst=entry.getKey();
                sendMessageList(dst, list.get(0).getSrc(), list);
                if(transport.statsEnabled())
                    transport.incrBatchesSent(1);
            }
        }
        clearMessages();
        count=0;
    }
757384
void notifyBackupComplete() {
        int newBackupAcksCompleted = BACKUP_ACKS_RECEIVED.incrementAndGet(this);

        Object pendingResponse = this.pendingResponse;
        if (pendingResponse == VOID) {
            // no pendingResponse has been set, so we are done since the invocation on the primary needs to complete first
            return;
        }

        // if a pendingResponse is set, then the backupsAcksExpected has been set (so we can now safely read backupsAcksExpected)
        int backupAcksExpected = this.backupsAcksExpected;
        if (backupAcksExpected < newBackupAcksCompleted) {
            // the backups have not yet completed, so we are done
            return;
        }

        if (backupAcksExpected != newBackupAcksCompleted) {
            // we managed to complete one backup, but we were not the one completing the last backup, so we are done
            return;
        }

        // we are the lucky one since we just managed to complete the last backup for this invocation and since the
        // pendingResponse is set, we can set it on the future
        complete(pendingResponse);
    }
1074128
def flush(self):
        
        self._flush_notification.set()
        if self.sender:
            self.sender.start()
974707
def sendall(self, data, flags=0):
        ''''''
        count = len(data)
        while count:
            sent = self.send(data, flags)
            # This could probably be a buffer object
            data = data[sent:]
            count -= sent