Add support for rd transport header.

Closes #324.

tchannel/request.py

@@ -83,6 +83,7 @@ class TransportHeaders(object):
         'claim_at_finish',
         'failure_domain',
         'retry_flags',
+        'routing_delegate',
         'scheme',
         'speculative_exe',
         'shard_key',
@@ -97,6 +98,7 @@ def __init__(self,
                  scheme=None,
                  speculative_exe=None,
                  shard_key=None,
+                 routing_delegate=None,
                  **kwargs):
 
         if scheme is None:
@@ -107,6 +109,7 @@ def __init__(self,
         self.claim_at_finish = claim_at_finish
         self.failure_domain = failure_domain
         self.retry_flags = retry_flags
+        self.routing_delegate = routing_delegate
         self.scheme = scheme
         self.speculative_exe = speculative_exe
         self.shard_key = shard_key

tchannel/schemes/json.py

@@ -50,6 +50,7 @@ def __call__(
         hostport=None,
         shard_key=None,
         trace=None,
+        routing_delegate=None,
     ):
         """Make JSON TChannel Request.
 
@@ -86,6 +87,9 @@ def __call__(
         :param string hostport:
             A 'host:port' value to use when making a request directly to a
             TChannel service, bypassing Hyperbahn.
+        :param routing_delegate:
+            Name of a service to which the request router should forward the
+            request instead of the service specified in the call req.
 
         :rtype: Response
         """
@@ -107,6 +111,7 @@ def __call__(
             hostport=hostport,
             shard_key=shard_key,
             trace=trace,
+            routing_delegate=routing_delegate,
         )
 
         # deserialize

tchannel/schemes/raw.py

@@ -45,6 +45,7 @@ def __call__(
         hostport=None,
         shard_key=None,
         trace=None,
+        routing_delegate=None,
     ):
         """Make a raw TChannel request.
 
@@ -84,6 +85,9 @@ def __call__(
         :param string hostport:
             A 'host:port' value to use when making a request directly to a
             TChannel service, bypassing Hyperbahn.
+        :param routing_delegate:
+            Name of a service to which the request router should forward the
+            request instead of the service specified in the call req.
 
         :rtype: Response
         """
@@ -100,6 +104,7 @@ def __call__(
             hostport=hostport,
             shard_key=shard_key,
             trace=trace,
+            routing_delegate=routing_delegate,
         )
 
     def register(self, endpoint, **kwargs):

tchannel/schemes/thrift.py

@@ -78,6 +78,7 @@ def __call__(
         shard_key=None,
         trace=None,
         hostport=None,
+        routing_delegate=None,
     ):
         """Make a Thrift TChannel request.
 
@@ -103,6 +104,10 @@ def __call__(
             ``tchannel.retry``.
         :param string retry_limit:
             How many times to retry before
+        :param routing_delegate:
+            Name of a service to which the request router should forward the
+            request instead of the service specified in the call req.
+
         :rtype: Response
         """
         if not headers:
@@ -134,6 +139,7 @@ def __call__(
             hostport=hostport or request.hostport,
             shard_key=shard_key,
             trace=trace,
+            routing_delegate=routing_delegate,
         )
 
         response.headers = serializer.deserialize_header(

tchannel/tchannel.py

@@ -137,6 +137,7 @@ def call(
         timeout=None,
         retry_on=None,
         retry_limit=None,
+        routing_delegate=None,
         hostport=None,
         shard_key=None,
         trace=None,
@@ -184,6 +185,8 @@ def call(
         }
         if shard_key:
             transport_headers[transport.SHARD_KEY] = shard_key
+        if routing_delegate:
+            transport_headers[transport.ROUTING_DELEGATE] = routing_delegate
 
         # If we got some parent tracing info we always want to propagate it
         # along. Otherwise use the ``trace`` parameter that was passed in. If

tchannel/transport.py

@@ -27,6 +27,7 @@
 CLAIM_AT_FINISH = "caf"
 FAILURE_DOMAIN = "fd"
 RETRY_FLAGS = "re"
+ROUTING_DELEGATE = "rd"
 SCHEME = "as"
 SHARD_KEY = "sk"
 SPECULATIVE_EXE = "se"
@@ -40,6 +41,7 @@ def to_kwargs(data):
     args['claim_at_finish'] = data.get(CLAIM_AT_FINISH)
     args['failure_domain'] = data.get(FAILURE_DOMAIN)
     args['retry_flags'] = data.get(RETRY_FLAGS)
+    args['routing_delegate'] = data.get(ROUTING_DELEGATE)
     args['scheme'] = data.get(SCHEME)
     args['shard_key'] = data.get(SHARD_KEY)
     args['speculative_exe'] = data.get(SPECULATIVE_EXE)

tests/test_dispatch.py

@@ -22,6 +22,7 @@
 
 import pytest
 
+from tchannel import TChannel
 from tchannel.errors import FatalProtocolError
 from tchannel.messages import CallResponseMessage
 from tchannel.serializer.raw import RawSerializer
@@ -50,3 +51,50 @@ def test_dispatch_unexpected_message():
     dispatcher = RequestDispatcher()
     with pytest.raises(FatalProtocolError):
         dispatcher.handle(CallResponseMessage(), None)
+
+
+@pytest.mark.gen_test
+def test_routing_delegate_is_propagated_raw():
+    server = TChannel('server')
+    server.listen()
+
+    @server.raw.register('foo')
+    def handler(request):
+        assert request.transport.routing_delegate == 'delegate'
+        return b'success'
+
+    client = TChannel('client', known_peers=[server.hostport])
+    res = yield client.raw('service', 'foo', b'', routing_delegate='delegate')
+    assert res.body == b'success'
+
+
+@pytest.mark.gen_test
+def test_routing_delegate_is_propagated_json():
+    server = TChannel('server')
+    server.listen()
+
+    @server.json.register('foo')
+    def handler(request):
+        assert request.transport.routing_delegate == 'delegate'
+        return {'success': True}
+
+    client = TChannel('client', known_peers=[server.hostport])
+    res = yield client.json('service', 'foo', {}, routing_delegate='delegate')
+    assert res.body == {'success': True}
+
+
+@pytest.mark.gen_test
+def test_routing_delegate_is_propagated_thrift(thrift_module):
+    server = TChannel('server')
+    server.listen()
+
+    @server.thrift.register(thrift_module.Service)
+    def healthy(request):
+        assert request.transport.routing_delegate == 'delegate'
+        return True
+
+    client = TChannel('client', known_peers=[server.hostport])
+    res = yield client.thrift(
+        thrift_module.Service.healthy(), routing_delegate='delegate'
+    )
+    assert res.body is True