// MessageAck is part of queryservice.QueryService
193880
func (itc *internalTabletConn) MessageAck(ctx context.Context, target *querypb.Target, name string, ids []*querypb.Value) (int64, error) {
	count, err := itc.tablet.qsc.QueryService().MessageAck(ctx, target, name, ids)
	return count, tabletconn.ErrorFromGRPC(vterrors.ToGRPC(err))
}
919923
func (tsv *TabletServer) MessageAck(ctx context.Context, target *querypb.Target, name string, ids []*querypb.Value) (count int64, err error) {
	sids := make([]string, 0, len(ids))
	for _, val := range ids {
		sids = append(sids, sqltypes.ProtoToValue(val).ToString())
	}
	count, err = tsv.execDML(ctx, target, func() (string, map[string]*querypb.BindVariable, error) {
		return tsv.messager.GenerateAckQuery(name, sids)
	})
	if err != nil {
		return 0, err
	}
	messager.MessageStats.Add([]string{name, "Acked"}, count)
	return count, nil
}
1055425
func (res *Resolver) MessageAckKeyspaceIds(ctx context.Context, keyspace, name string, idKeyspaceIDs []*vtgatepb.IdKeyspaceId) (int64, error) {
	ids := make([]*querypb.Value, len(idKeyspaceIDs))
	ksids := make([]key.Destination, len(idKeyspaceIDs))
	for i, iki := range idKeyspaceIDs {
		ids[i] = iki.Id
		ksids[i] = key.DestinationKeyspaceID(iki.KeyspaceId)
	}

	rss, values, err := res.resolver.ResolveDestinations(ctx, keyspace, topodatapb.TabletType_MASTER, ids, ksids)
	if err != nil {
		return 0, err
	}

	return res.scatterConn.MessageAck(ctx, rss, values, name)
}
1074704
func Message(out proto.Message) interface {
	cfgclient.Resolver
	cfgclient.FormattingResolver
} {
	if out == nil {
		panic("cannot pass a nil proto.Message instance")
	}
	return &resolver{
		messageName: proto.MessageName(out),
		single:      out,
		singleType:  reflect.TypeOf(out),
	}
}
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)
}
130842
func (s *SubscribePacket) Unpack(src []byte) {
	s.MessageID = loadUint16(src)
	if len(src) < 2 {
		return // FIXME: error
	}
	src = src[2:]
	for len(src) > 2 {
		topic, end := loadString(src)
		src = src[end:]
		if len(src) < 1 {
			break // FIXME: error
		}
		s.Topics = append(s.Topics, topic)
		qos := loadByte(src)
		src = src[1:]
		s.Qoss = append(s.Qoss, qos)
	}
}
1010286
func (e *Error) EncodeMsgpack(enc *msgpack.Encoder) error {
	var err error
	switch v := e.err.(type) {
	case *Error:
		err = enc.Encode(ErrorLocal)
		if err != nil {
			return err
		}
		err = enc.Encode(v)
		if err != nil {
			return err
		}
	case error:
		err = enc.Encode(ErrorGo)
		if err != nil {
			return err
		}
		err = enc.Encode(GoError(v.Error()))
		if err != nil {
			return err
		}
	default:
		panic("type not supported")
	}
	err = enc.Encode(e.args)
	if err != nil {
		return err
	}
	err = enc.Encode(e.pkg)
	if err != nil {
		return err
	}
	err = enc.Encode(e.file)
	if err != nil {
		return err
	}
	err = enc.Encode(e.line)
	if err != nil {
		return err
	}
	err = enc.Encode(e.debugInfo)
	if err != nil {
		return err
	}
	if e.next == nil {
		err = enc.Encode(NextIsNill)
		if err != nil {
			return err
		}
	} else {
		err = enc.Encode(Next)
		if err != nil {
			return err
		}
		err = enc.Encode(e.next)
		if err != nil {
			return err
		}
	}
	return nil
}
538537
func (client *Client) ReplyMessage(replyToken string, messages ...SendingMessage) *ReplyMessageCall {
	return &ReplyMessageCall{
		c:          client,
		replyToken: replyToken,
		messages:   messages,
	}
}
524435
func MsgPackEncode(b *[]byte, in interface{}) error {
	return codec.NewEncoderBytes(b, MsgpackHandle).Encode(in)
}
989486
func (client *QueryClient) MessageAck(name string, ids []string) (int64, error) {
	bids := make([]*querypb.Value, 0, len(ids))
	for _, id := range ids {
		bids = append(bids, &querypb.Value{
			Type:  sqltypes.VarChar,
			Value: []byte(id),
		})
	}
	return client.server.MessageAck(client.ctx, &client.target, name, bids)
}
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),
	}
}
299565
func (tsv *TabletServer) MessageStream(ctx context.Context, target *querypb.Target, name string, callback func(*sqltypes.Result) error) (err error) {
	return tsv.execRequest(
		ctx, 0,
		"MessageStream", "stream", nil,
		target, nil, false /* isBegin */, false, /* allowOnShutdown */
		func(ctx context.Context, logStats *tabletenv.LogStats) error {
			plan, err := tsv.qe.GetMessageStreamPlan(name)
			if err != nil {
				return err
			}
			qre := &QueryExecutor{
				query:    "stream from msg",
				plan:     plan,
				ctx:      ctx,
				logStats: logStats,
				tsv:      tsv,
			}
			return qre.MessageStream(callback)
		},
	)
}
426080
func (b *boxedMessage) Load(from []byte) error {
	var ssver, spub, snonce, sbox string
	var err error

	allMatches := messageParser.FindAllStringSubmatch(string(from), -1) // -> [][][]byte
	if len(allMatches) != 1 {
		return fmt.Errorf("invalid message format")
	}
	matches := allMatches[0]
	if len(matches) != 5 {
		return fmt.Errorf("invalid message format")
	}

	ssver = matches[1]
	spub = matches[2]
	snonce = matches[3]
	sbox = matches[4]

	b.SchemaVersion, err = strconv.Atoi(ssver)
	if err != nil {
		return err
	}

	pub, err := base64.StdEncoding.DecodeString(spub)
	if err != nil {
		return err
	}
	pubBytes := []byte(pub)
	if len(pubBytes) != 32 {
		return fmt.Errorf("public key invalid")
	}
	var public [32]byte
	copy(public[:], pubBytes[0:32])
	b.EncrypterPublic = public

	nnc, err := base64.StdEncoding.DecodeString(snonce)
	if err != nil {
		return err
	}
	nonceBytes := []byte(nnc)
	if len(nonceBytes) != 24 {
		return fmt.Errorf("nonce invalid")
	}
	var nonce [24]byte
	copy(nonce[:], nonceBytes[0:24])
	b.Nonce = nonce

	box, err := base64.StdEncoding.DecodeString(sbox)
	if err != nil {
		return err
	}
	b.Box = []byte(box)

	return nil
}
1372468
func SexpToMsgpack(exp Sexp) ([]byte, interface{}) {

	json := []byte(SexpToJson(exp))
	iface, err := JsonToGo(json)
	panicOn(err)
	by, err := GoToMsgpack(iface)
	panicOn(err)
	return by, iface
}
1311681
func (sc *conn) removeAck(guid string) *ack {
	var t *time.Timer
	sc.Lock()
	a := sc.pubAckMap[guid]
	if a != nil {
		t = a.t
		delete(sc.pubAckMap, guid)
	}
	pac := sc.pubAckChan
	sc.Unlock()

	// Cancel timer if needed.
	if t != nil {
		t.Stop()
	}

	// Remove from channel to unblock PublishAsync
	if a != nil && len(pac) > 0 {
		<-pac
	}
	return a
}
776424
func NewMissionAck(TARGET_SYSTEM uint8, TARGET_COMPONENT uint8, TYPE uint8) *MissionAck {
	m := MissionAck{}
	m.TARGET_SYSTEM = TARGET_SYSTEM
	m.TARGET_COMPONENT = TARGET_COMPONENT
	m.TYPE = TYPE
	return &m
}
1188521
func (mp *LogMergePolicy) message(message string, w *IndexWriter) {
	if mp.verbose(w) {
		w.infoStream.Message("LMP", message)
	}
}