Method sendAndReceive

broker.go:1173–1196

(req protocolBody, res protocolBody)

}

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

	promise, err := b.send(req, res)
	if err != nil {
		b.maybeCloseLocked(err)
		return err
	}

	if promise == nil {
		return nil
	}

	err = handleResponsePromise(req, res, promise, b.metricRegistry)
	if err != nil {
		b.maybeCloseLocked(err)
		return err
	}
	if res != nil {
		b.handleThrottledResponse(res)
	}
	return nil
}

// negotiateApiVersion clamps pb's version to the broker's advertised maximum
// for pb's API (treating pb's current version as the client max). When the