Method sendInternal

broker.go:1124–1171

b.lock must be held by caller

(rb protocolBody, promise *responsePromise)

// b.lock must be held by caller
func (b *Broker) sendInternal(rb protocolBody, promise *responsePromise) error {
	// try restricting API version to ranges advertised by the broker
	if err := restrictApiVersion(rb, b.brokerAPIVersions); err != nil {
		return err
	}

	// response versions must always match their corresponding request's
	if promise != nil && promise.response != nil {
		promise.response.setVersion(rb.version())
	}

	if !b.conf.Version.IsAtLeast(rb.requiredVersion()) {
		return ErrUnsupportedVersion
	}

	req := &request{correlationID: b.correlationID, clientID: b.conf.ClientID, body: rb}
	buf, err := encode(req, b.metricRegistry)
	if err != nil {
		return err
	}

	// check and wait if throttled
	b.waitIfThrottled()

	requestTime := time.Now()
	// Will be decremented in responseReceiver (except error or request with NoResponse)
	b.addRequestInFlightMetrics(1)
	bytes, err := b.write(buf)
	b.updateOutgoingCommunicationMetrics(bytes)
	b.updateProtocolMetrics(rb)
	if err != nil {
		b.addRequestInFlightMetrics(-1)
		return err
	}
	b.correlationID++

	if promise == nil {
		// Record request latency without the response
		b.updateRequestLatencyAndInFlightMetrics(time.Since(requestTime))
		return nil
	}

	promise.requestTime = requestTime
	promise.correlationID = req.correlationID
	b.responses <- promise

	return nil
}

func (b *Broker) sendAndReceive(req protocolBody, res protocolBody) error {
	b.lock.Lock()