code: def _make_request_to_broker(self, broker, requestId, request, **kwArgs):
          
          def _timeout_request(broker, requestId):
              """The time we allotted for the request expired, cancel it."""
              try:
                  # FIXME: This should be done by calling .cancel() on the Deferred
                  # returned by the broker client.
                  broker.cancelRequest(requestId, reason=RequestTimedOutError(
                      'Request: {} cancelled due to timeout'.format(requestId)))
              except KeyError:  # pragma: no cover This should never happen...
                  log.exception('ERROR: Failed to find key for timed-out '
                                'request. Broker: %r Req: %d',
                                broker, requestId)
                  raise
              if self._disconnect_on_timeout:
                  broker.disconnect()
  
          def _alert_blocked_reactor(timeout, start):
              """Complain if this timer didn't fire before the timeout elapsed"""
              now = self.reactor.seconds()
              if now >= (start + timeout):
                  log.warning('Reactor was starved for %r seconds', now - start)
  
          def _cancel_timeout(result, dc):
              """Request completed/cancelled, cancel the timeout delayedCall."""
              if dc.active():
                  dc.cancel()
              return result
  
          # Make the request to the specified broker
          log.debug('_mrtb: sending request: %d to broker: %r',
                    requestId, broker)
          d = broker.makeRequest(requestId, request, **kwArgs)
          # Set a delayedCall to fire if we don't get a reply in time
          dc = self.reactor.callLater(
              self.timeout, _timeout_request, broker, requestId)
          # Set a delayedCall to complain if the reactor has been blocked
          rc = self.reactor.callLater(
              (self.timeout * 0.9), _alert_blocked_reactor, self.timeout,
              self.reactor.seconds())
          # Setup a callback on the request deferred to cancel both callLater
          d.addBoth(_cancel_timeout, dc)
          d.addBoth(_cancel_timeout, rc)
          return d
docs: Send a request to the specified broker.