// MessageAck acks messages across multiple shards.
341783
func (stc *ScatterConn) MessageAck(ctx context.Context, rss []*srvtopo.ResolvedShard, values [][]*querypb.Value, name string) (int64, error) {
	var mu sync.Mutex
	var totalCount int64
	allErrors := stc.multiGo(ctx, "MessageAck", rss, topodatapb.TabletType_MASTER, func(rs *srvtopo.ResolvedShard, i int) error {
		count, err := rs.QueryService.MessageAck(ctx, rs.Target, name, values[i])
		if err != nil {
			return err
		}
		mu.Lock()
		totalCount += count
		mu.Unlock()
		return nil
	})
	return totalCount, allErrors.AggrError(vterrors.Aggregate)
}
1056073
func NewMessageBatch(maxMessageCount int) MessageBatch {
	now := time.Now().Unix()
	return MessageBatch{
		queue:     [2]messageBuffer{newMessageBuffer(maxMessageCount), newMessageBuffer(maxMessageCount)},
		flushing:  new(tsync.WaitGroup),
		lastFlush: &now,
		activeSet: new(uint32),
		closed:    new(int32),
	}
}
147799
func (mp *MachineProvider) ack(ctx context.Context, hostname, backend string) error {
	return mp.client.Ack(&machine.ComponentsMachineProviderRpcMessagesAckRequest{
		Backend:  backend,
		Hostname: hostname,
	}).Context(ctx).Do()
}
1108644
func encodeMsgPack(msg interface{}) ([]byte, error) {
	var buf bytes.Buffer
	err := codec.NewEncoder(&buf, msgpackHandle).Encode(msg)
	return buf.Bytes(), err
}
619975
func (sc *conn) processAck(m *nats.Msg) {
	pa := &pb.PubAck{}
	err := pa.Unmarshal(m.Data)
	if err != nil {
		panic(fmt.Errorf("error during ack unmarshal: %v", err))
	}

	// Remove
	a := sc.removeAck(pa.Guid)
	if a != nil {
		// Capture error if it exists.
		if pa.Error != "" {
			err = errors.New(pa.Error)
		}
		if a.ah != nil {
			// Perform the ackHandler callback
			a.ah(pa.Guid, err)
		} else if a.ch != nil {
			// Send to channel directly
			a.ch <- err
		}
	}
}
125532
func (s *StanServer) processAckMsg(m *nats.Msg) {
	ack := &pb.Ack{}
	if ack.Unmarshal(m.Data) != nil {
		if s.processCtrlMsg(m) {
			return
		}
	}
	c := s.channels.get(ack.Subject)
	if c == nil {
		s.log.Errorf("Unable to process ack seq=%d, channel %s not found", ack.Sequence, ack.Subject)
		return
	}
	sub := c.ss.LookupByAckInbox(m.Subject)
	if sub == nil {
		return
	}
	s.processAck(c, sub, ack.Sequence, true)
}
689191
func (sub *subState) startAckSub(nc *nats.Conn, cb nats.MsgHandler) error {
	ackSub, err := nc.Subscribe(sub.AckInbox, cb)
	if err != nil {
		return err
	}
	sub.Lock()
	// Should not occur, but if it was already set,
	// unsubscribe old and replace.
	sub.stopAckSub()
	sub.ackSub = ackSub
	sub.ackSub.SetPendingLimits(-1, -1)
	sub.Unlock()
	return nil
}
477329
func messageTag(hash func() hash.Hash, km, msg, shared []byte) []byte {
	if shared == nil {
		shared = make([]byte, 0)
	}
	mac := hmac.New(hash, km)
	mac.Write(msg)
	tag := mac.Sum(nil)
	return tag
}
182208
func (v *Validator) Messages() []string {
	msgs := []string{}
	for _, vr := range v.results {
		msgs = append(msgs, vr.message)
	}
	return msgs
}
1084977
func (m *mux) ackRequest(ok bool, data []byte) error {
	if ok {
		return m.sendMessage(globalRequestSuccessMsg{Data: data})
	}
	return m.sendMessage(globalRequestFailureMsg{Data: data})
}
297352
func (h *handle) messageToCDummy(msg *Message) {
	var cmsg C.rd_kafka_message_t
	h.messageToC(msg, &cmsg)
}
524435
func MsgPackEncode(b *[]byte, in interface{}) error {
	return codec.NewEncoderBytes(b, MsgpackHandle).Encode(in)
}
743464
func (m *ReceivedMessageImpl) Ack(err error) {
	ackMsg := &proto.GossipMessage{
		Nonce: m.GetGossipMessage().Nonce,
		Content: &proto.GossipMessage_Ack{
			Ack: &proto.Acknowledgement{},
		},
	}
	if err != nil {
		ackMsg.GetAck().Error = err.Error()
	}
	m.Respond(ackMsg)
}
194075
func (vtg *VTGate) MessageAck(ctx context.Context, request *vtgatepb.MessageAckRequest) (response *querypb.MessageAckResponse, err error) {
	defer vtg.server.HandlePanic(&err)
	ctx = withCallerIDContext(ctx, request.CallerId)
	count, vtgErr := vtg.server.MessageAck(ctx, request.Keyspace, request.Name, request.Ids)
	if vtgErr != nil {
		return nil, vterrors.ToGRPC(vtgErr)
	}
	return &querypb.MessageAckResponse{
		Result: &querypb.QueryResult{
			RowsAffected: uint64(count),
		},
	}, nil
}
251679
func (ss *SQLSubStore) ackSeq(subid, seqno uint64) (bool, error) {
	ap := ss.getOrCreateAcksPending(subid, seqno)
	// If still in cache and not persisted into a row,
	// then simply remove from map and do not persist the ack.
	if _, exists := ap.msgs[seqno]; exists {
		delete(ap.msgs, seqno)
	} else if row := ap.msgToRow[seqno]; row != nil {
		ap.acks[seqno] = struct{}{}
		// This is an ack for a pending msg that was persisted
		// in a row. Update the row's msgRef count.
		delete(ap.msgToRow, seqno)
		row.msgsRefs--
		// If all pending messages in that row have been ack'ed
		if row.msgsRefs == 0 {
			// and if all acks on that row are no longer needed
			// (or there was none)
			if row.acksRefs == 0 {
				// then this row can be deleted.
				if err := ss.deleteSubPendingRow(subid, row.ID); err != nil {
					return false, err
				}
				// If there is no error, we don't even need
				// to persist this ack.
				delete(ap.acks, seqno)
			}
			// Since there is no pending message left in this
			// row, let's find all the corresponding acks' rows
			// for these sequences and update their acksRefs
			for seq := range row.msgs {
				delete(row.msgs, seq)
				ackRow := ap.ackToRow[seq]
				if ackRow != nil {
					// We found the row for the ack of this sequence,
					// remove from map and update reference count.
					// delete(ap.ackToRow, seq)
					ackRow.acksRefs--
					// If all acks for that row are no longer needed and
					// that row has also no pending messages, then ok to
					// delete.
					if ackRow.acksRefs == 0 && ackRow.msgsRefs == 0 {
						if err := ss.deleteSubPendingRow(subid, ackRow.ID); err != nil {
							return false, err
						}
					}
				} else {
					// That means the ack is in current cache so we won't
					// need to persist it.
					delete(ap.acks, seq)
				}
			}
			sqlSeqMapPool.Put(row.msgs)
			row.msgs = nil
		}
	}
	return len(ap.acks) >= sqlMaxPendingAcks, nil
}
818603
func ReplyOk(message ...string) *Reply {
	if len(message) == 0 {
		message = []string{"Ok"}
	}
	return &Reply{250, message, nil}
}
1188303
func (s *Socket) ackSyn(syn syn) (c *Conn, ok bool) {
	c = s.newConn(s.strNetAddr(syn.addr))
	c.send_id = syn.conn_id
	c.recv_id = c.send_id + 1
	c.seq_nr = uint16(rand.Int())
	c.lastAck = c.seq_nr - 1
	c.ack_nr = syn.seq_nr
	c.synAcked = true
	c.updateCanWrite()
	if !s.registerConn(c.recv_id, resolvedAddrStr(syn.addr), c) {
		// SYN that triggered this accept duplicates existing connection.
		// Ack again in case the SYN was a resend.
		c = s.conns[connKey{resolvedAddrStr(syn.addr), c.recv_id}]
		if c.send_id != syn.conn_id {
			panic(":|")
		}
		c.sendState()
		return
	}
	c.sendState()
	ok = true
	return
}