Received an ACK; executes on the node asking&receiving the Value
180396
@Override public void onAck() {
    if( _val != null ) {        // Set transient fields after deserializing
      assert !_xkey.home() && _val._key == null;
      _val._key = _xkey;
    }
    // Now update the local store, caching the result.

    // We only started down the TGK path because we missed locally, so we only
    // expect to find a NULL in the local store.  If somebody else installed
    // another value (e.g. a racing TGK, or racing local Put) this value must
    // be more recent than our NULL - but is UNORDERED relative to the Value
    // returned from the Home.  We'll take the local Value to preserve ordering
    // and rely on invalidates from Home to force refreshes as needed.

    // Hence we can do a blind putIfMatch here over a null or empty Value
    // If it fails, what is there is also the TGK result.
    Value old = H2O.raw_get(_xkey);
    if( old != null && !old.isEmpty() ) old=null;
    Value res = H2O.putIfMatch(_xkey,_val,old);
    if( res != old ) _val = res;
  }
1308117
@Override
    public void ack() {
        waitClientRunning();
        if (!running) {
            return;
        }

        if (kafkaConsumer != null) {
            kafkaConsumer.commitSync();
        }
        if (kafkaConsumer2 != null) {
            kafkaConsumer2.commitSync();
        }
    }
1081852
def HandleVerack(self):
        """"""
        m = Message('verack')
        self.SendSerializedMessage(m)
        self.leader.NodeCount += 1
        self.identifier = self.leader.NodeCount
        logger.debug(f"{self.prefix} Handshake complete!")
        self.handshake_complete = True
        self.ProtocolReady()
1278584
@Override
    public void onMessage(StreamMessage received)
    {
        if (this.reloadConfig && this.watcher != null)
        {
            Map<String, Object> reloadedConfig = null;

            try
            {
                reloadedConfig = this.watcher.readIfUpdated();
            }
            catch (IOException ex)
            {
                String logFormat = "Config file reload failed. Skip reload config.";
                logger.warn(logFormat, ex);
            }

            if (reloadedConfig != null)
            {
                onUpdate(reloadedConfig);
            }
        }

        this.executingKeyHistory = received.getHeader().getHistory();
        this.responsed = false;

        // Execute message processing.
        // This class not handles exceptions.
        // Because if throws exception, this class not need ack.
        try
        {
            onExecute(received);
        }
        finally
        {
            clearExecuteStatus();
        }

        // If not responsed, auto ack
        if (this.responsed == false)
        {
            super.ack();
        }
    }
803718
def read_ack_from_sock(sock, unpacker)
      begin
        raw_data = sock.instance_of?(Fluent::PluginHelper::Socket::WrappedSocket::TLS) ? sock.readpartial(@read_length) : sock.recv(@read_length)
      rescue Errno::ECONNRESET, EOFError # ECONNRESET for #recv, #EOFError for #readpartial
        raw_data = ""
      end
      info = @sock_ack_waiting_mutex.synchronize{ @sock_ack_waiting.find{|i| i.sock == sock } }

      # When connection is closed by remote host, socket is ready to read and #recv returns an empty string that means EOF.
      # If this happens we assume the data wasn't delivered and retry it.
      if raw_data.empty?
        log.warn "destination node closed the connection. regard it as unavailable.", host: info.node.host, port: info.node.port
        info.node.disable!
        rollback_write(info.chunk_id, update_retry: false)
        return nil
      else
        unpacker.feed(raw_data)
        res = unpacker.read
        log.trace "getting response from destination", host: info.node.host, port: info.node.port, chunk_id: dump_unique_id_hex(info.chunk_id), response: res
        if res['ack'] != info.chunk_id_base64
          # Some errors may have occurred when ack and chunk id is different, so send the chunk again.
          log.warn "ack in response and chunk id in sent data are different", chunk_id: dump_unique_id_hex(info.chunk_id), ack: res['ack']
          rollback_write(info.chunk_id, update_retry: false)
          return nil
        else
          log.trace "got a correct ack response", chunk_id: dump_unique_id_hex(info.chunk_id)
        end
        return info.chunk_id
      end
    rescue => e
      log.error "unexpected error while receiving ack message", error: e
      log.error_backtrace
    ensure
      info.sock.close_write rescue nil
      info.sock.close rescue nil
      @sock_ack_waiting_mutex.synchronize do
        @sock_ack_waiting.delete(info)
      end
    end
425011
function (node) {
        if (funcInfo.shouldCheck) {
          funcInfo.hasReturn = true

          if (!node.argument) {
            context.report({
              node: node,
              message: 'Expected a return value'
            })
          }
        }
      }
638698
function(msg, callback) {
    try {
      var result = JSON.parse(msg.content);

      // the message is ack'ed on success
      // second parameter is optional for logging
      callback(null, result);
    } catch(err) {

      // the message is nack'ed on error
      callback(err);
    }
  }
568575
public function ack()
    {
        $this->checkChannel();

        $outputFrame = BasicAckFrame::create($this->tag);
        $outputFrame->frameChannelId = $this->channel->getId();

        $this->channel->getConnection()->command($outputFrame);
    }
543778
def getCallBack(self, ps, post, action):
        '''
        '''
        node = self.getNode(post)
        if node is None:
            raise NoSuchFunction
        if node.authorize(None, post, action):
            return node.getOperation(ps, action)
        else:
            raise NotAuthorized, "Authorization failed for method %s" % action
26643
def send_ack(self):
        """"""

        if self.last_ack == self.proto.max_id:
            return
        LOGGER.debug("ack (%d)", self.proto.max_id)
        self.last_ack = self.proto.max_id
        self.send_message(f"4{to_json([self.proto.max_id])}")
633264
def ack_answer(opts = {})
      msg = <<-BODY

ACK [next_url] SIP/2.0
Via: SIP/2.0/[transport] #{@adv_ip}:[local_port];branch=[branch]
From: "#{@from_user}" <sip:#{@from_user}@#{@adv_ip}:[local_port]>;tag=[call_number]
To: <sip:#{to_addr}>[peer_tag_param]
Call-ID: [call_id]
CSeq: [cseq] ACK
Contact: <sip:[$local_addr];transport=[transport]>
Max-Forwards: 100
User-Agent: #{USER_AGENT}
Content-Length: 0
[routes]
      BODY
      send msg, opts
      start_media
    end
7217
void handleGet(
    Object value,
    MatchSpaceKey msg,
    EvalCache cache,
    Object contextValue,
    SearchResults result)
    throws MatchingException, BadMessageFormatMatchingException
  {
    if (tc.isEntryEnabled())
      tc.entry(
        this,
        cclass,
        "handleGet",
        "value: "
          + value
          + "msg: "
          + msg
          + ", context: "
          + contextValue
          + ", result: "
          + result);

    if (!(value instanceof String))
      return;
    char[] chars = ((String) value).toCharArray();
    subTree.get(chars, 0, chars.length, false, msg, cache, contextValue, result);
    if (haveEqualityMatches())
      handleEqualityGet(value, msg, cache, contextValue, result);
    if (tc.isEntryEnabled())
      tc.exit(this,cclass, "handleGet");
  }
1296377
def on_recv_rsp(self, rsp_pb):
        """"""
        ret_code, msg, _= SubAccPush.unpack_rsp(rsp_pb)

        if self._notify_obj is not None:
            self._notify_obj.on_async_sub_acc_push(ret_code, msg)

        return ret_code, msg
143998
private void processExchange(final int idx, final HTTPExchange exch) {
        assertUnlocked();

        HTTPResponse resp;
        AbstractBody body;
        int respCode;
        try {
            if (LOG.isLoggable(Level.FINEST))
                LOG.finest(
                    "Thread " + idx + " is sending "
                        + exch.getRequest().getAttribute(Attributes.RID));

            resp = exch.getHTTPResponse();
            body = resp.getBody();
            respCode = resp.getHTTPStatus();

            if (LOG.isLoggable(Level.FINEST)) {
                String respRid = body.getAttribute(Attributes.RID);
                if (respRid == null)
                    respRid = exch.getRequest().getAttribute(Attributes.RID);
                LOG.finest(
                    "Thread " + idx + " received response"
                        + " for RID: " + respRid
                        + " code: " + respCode
                        + " ACK: " + body.getAttribute(Attributes.ACK));
            }
        } catch (BOSHException boshx) {
            LOG.log(Level.FINEST, "Could not obtain response", boshx);
            dispose(boshx);
            return;
        } catch (InterruptedException intx) {
            LOG.log(Level.FINEST, INTERRUPTED, intx);
            dispose(intx);
            return;
        }
        fireResponseReceived(body);

        // Process the message with the current session state
        AbstractBody req = exch.getRequest();
        CMSessionParams params;
        List<HTTPExchange> toResend = null;
        lock.lock();
        try {
            if (!isWorking()) {
                lock.unlock();
                return;
            }
            
            // Check for session creation response info, if needed
            if (cmParams == null) {
                cmParams = CMSessionParams.fromSessionInit(req, body);

                // Adjust number of request processors based on REQ value
                adjustRequestProcessorsPool();

                // The following call handles the lock. It's not an escape.
                fireConnectionEstablished();
            }
            params = cmParams;

            checkForTerminalBindingConditions(body, respCode);
            if (isTermination(body)) {
                // Explicit termination
                lock.unlock();
                dispose(null);
                return;
            }

            // FIXME this may not work with more than 1 RequestProcessor
            if (isRecoverableBindingCondition(body)) {
                // Retransmit outstanding requests
                if (toResend == null) {
                    toResend = new ArrayList<HTTPExchange>(exchanges.size());
                }
                for (HTTPExchange exchange : exchanges) {
                    HTTPExchange resendExch =
                            new HTTPExchange(exchange.getRequest());
                    toResend.add(resendExch);
                }
                for (HTTPExchange exchange : toResend) {
                    exchanges.add(exchange);
                }
            } else {
                // Process message as normal
                processRequestAcknowledgements(req, body);
                processResponseAcknowledgementData(req);
                HTTPExchange resendExch =
                        processResponseAcknowledgementReport(body);
                if (resendExch != null && toResend == null) {
                    toResend = new ArrayList<HTTPExchange>(1);
                    toResend.add(resendExch);
                    exchanges.add(resendExch);
                }
            }
        } catch (BOSHException boshx) {
            LOG.log(Level.FINEST, "Could not process response", boshx);
            lock.unlock();
            dispose(boshx);
            return;
        } finally {
            if (lock.isHeldByCurrentThread()) {
                try {
                    exchanges.remove(exch);
                    if (exchanges.isEmpty()) {
                        scheduleEmptyRequest(processPauseRequest(req));
                    }
                    notFull.signalAll();
                } finally {
                    lock.unlock();
                }
            }
        }

        if (toResend != null) {
            for (HTTPExchange resend : toResend) {
                HTTPResponse response =
                        httpSender.send(params, resend.getRequest());
                resend.setHTTPResponse(response);
                fireRequestSent(resend.getRequest());
            }
        }
    }
584961
def wait_for_answer(opts = {})
      receive_trying opts
      receive_ringing opts
      receive_progress opts
      receive_answer opts
      ack_answer opts
    end
1149878
def on_return(self, node):  # ('value',)
        """"""
        self.retval = self.run(node.value)
        if self.retval is None:
            self.retval = ReturnedNone
        return
232843
def ack(self):
        
        if self.acknowledged:
            raise self.MessageStateError(
                "Message already acknowledged with state: %s" % self._state)
        self.backend.ack(self._frame)
        self._state = "ACK"