Method ProcessRequest

internal/cli/processor.go:263–303

(request *proto.OperationalMessage)

}

func (c *Processor) ProcessRequest(request *proto.OperationalMessage) (resp *proto.OperationalMessage, err error) {
	timeStart := time.Now()

	ch, err := c.send(request)
	if err == nil {
		if r, ok := <-ch; ok {
			resp = r.GetResponse()
			err = r.GetError()
		} else {
			resp = nil
			err = fmt.Errorf("response channel closed by request processor")
		}
	}

	if err != nil && !specialError(err) {
		err = &IOError{err}
	}
	if cal.IsEnabled() {
		var txnType string
		if c.server.SSLEnabled {
			txnType = kCalSslTxnType
		} else {
			txnType = kCalTxnType
		}
		rht := time.Since(timeStart)

		if err == nil {
			status := resp.GetOpStatus()
			b := logging.NewKVBuffer()
			b.AddOpRequestResponseInfo(request, resp)
			cal.AtomicTransaction(txnType, request.GetOpCode().String(), logging.CalStatus(status).CalStatus(), rht, b.Bytes())
		} else {
			tail := fmt.Sprintf("raddr=%s&res_timeout=%dms&ns=%s&%s", c.server.Addr,
				c.responseTimeout.Nanoseconds()/int64(1e6), request.GetNamespace(), err.Error())
			cal.AtomicTransaction(txnType, request.GetOpCode().String(),
				cal.StatusError, rht, []byte(tail)) ///TODO to change: data to cal
		}
	}

	return
}

func (c *Processor) ProcessBatchRequests(requests []*proto.OperationalMessage) (responses []*proto.OperationalMessage, err error) {
	numRequests := len(requests)