Method sendQuery

client/conn.go:390–474

(ctx context.Context, queryType types.QueryType, queries []types.Query)

}

func (c *conn) sendQuery(ctx context.Context, queryType types.QueryType, queries []types.Query) (affectedRows int64, lastInsertID int64, rows driver.Rows, err error) {
	var uc *pconn // peer connection used to execute the queries

	uc = c.leader
	// use follower pconn only when the query is readonly
	if queryType == types.ReadQuery && c.follower != nil {
		uc = c.follower
	}
	if uc == nil {
		uc = c.follower
	}

	// allocate sequence
	connID, seqNo := allocateConnAndSeq()
	defer putBackConn(connID)

	defer func() {
		log.WithFields(log.Fields{
			"count":  len(queries),
			"type":   queryType.String(),
			"connID": connID,
			"seqNo":  seqNo,
			"target": uc.pCaller.Target(),
			"source": c.localNodeID,
		}).WithError(err).Debug("send query")
	}()

	// build request
	req := &types.Request{
		Header: types.SignedRequestHeader{
			RequestHeader: types.RequestHeader{
				QueryType:    queryType,
				NodeID:       c.localNodeID,
				DatabaseID:   c.dbID,
				ConnectionID: connID,
				SeqNo:        seqNo,
				Timestamp:    getLocalTime(),
			},
		},
		Payload: types.RequestPayload{
			Queries: queries,
		},
	}

	if err = req.Sign(c.privKey); err != nil {
		return
	}

	// set receipt if key exists in context
	if val := ctx.Value(&ctxReceiptKey); val != nil {
		val.(*atomic.Value).Store(&Receipt{
			RequestHash: req.Header.Hash(),
		})
	}

	var response types.Response
	if err = uc.pCaller.Call(route.DBSQuery.String(), req, &response); err != nil {
		return