add tchannel streaming functionality

Author: Junchao Wu

examples/handler.py

@@ -21,34 +21,54 @@
 from __future__ import absolute_import
 
 import random
-import time
 
 import tornado.gen
 
 from tchannel.tornado.dispatch import TornadoDispatcher
+from tchannel.tornado.stream import InMemStream
+from tchannel.tornado.util import print_arg, get_arg
 
 
+@tornado.gen.coroutine
 def say_hi(request, response, opts):
-    response.write(arg3="hi")
+    arg2 = yield get_arg(request, 1)
+    arg3 = yield get_arg(request, 2)
+    response.argstreams = [
+        InMemStream(request.endpoint),
+        InMemStream(arg2),
+        InMemStream(arg3)
+    ]
 
 
+@tornado.gen.coroutine
 def say_ok(request, response, opts):
-    response.write(arg3="ok")
+    yield print_arg(request, 1)
+    yield print_arg(request, 2)
 
+    response.argstreams = [
+        InMemStream(),
+        InMemStream(),
+        InMemStream("world")]
 
+
+@tornado.gen.coroutine
 def echo(request, response, opts):
-    response.write(arg3=request.message.args[2])
+    # stream args right back to request side
+    print "streaming"
+    response.argstreams = [
+        InMemStream(request.endpoint),
+        request.argstreams[1],
+        request.argstreams[2]
+    ]
 
 
 @tornado.gen.coroutine
 def slow(request, response, opts):
     yield tornado.gen.sleep(random.random())
-    response.write(arg3="done")
-
-
-def blocking(request, response, opts):
-    time.sleep(random.random())
-    response.write(arg3="yawn")
+    response.argstreams = [
+        InMemStream(),
+        InMemStream(),
+        InMemStream("done")]
 
 
 def get_example_handler():
@@ -58,10 +78,15 @@ def get_example_handler():
     dispatcher.register("ok", say_ok)
     dispatcher.register("echo", echo)
     dispatcher.register("slow", slow)
-    dispatcher.register("blocking", blocking)
 
     @dispatcher.route("bye")
     def say_bye(request, response, opts):
-        response.write("bye bye!")
+        yield print_arg(request, 1)
+        yield print_arg(request, 2)
+
+        response.argstreams = [
+            InMemStream(),
+            InMemStream(),
+            InMemStream("world")]
 
     return dispatcher

examples/options.py

@@ -31,4 +31,9 @@ def get_args():
         "--host",
         dest="host", default="localhost"
     )
+    parser.add_argument(
+        "--file",
+        dest="filename"
+    )
+
     return parser.parse_args()

examples/stream_client.py

@@ -0,0 +1,44 @@
+import os
+import sys
+from options import get_args
+import tornado
+import tornado.ioloop
+from tchannel.tornado import TChannel
+from tchannel.tornado.stream import InMemStream, PipeStream
+from tchannel.tornado.util import print_arg
+
+
+@tornado.gen.coroutine
+def send_stream(arg1, arg2, arg3, host):
+    tchannel = TChannel()
+    response = yield tchannel.request(host).send(
+        arg1,
+        arg2,
+        arg3)
+
+    yield print_arg(response, 0)
+    yield print_arg(response, 1)
+    yield print_arg(response, 2)
+
+
+def main():
+    args = get_args()
+
+    arg1 = InMemStream("echo")
+    arg2 = InMemStream()
+    arg3 = InMemStream()
+
+    ioloop = tornado.ioloop.IOLoop.current()
+    if args.filename == "stdin":
+        arg3 = PipeStream(sys.stdin.fileno())
+        send_stream(arg1, arg2, arg3, args.host)
+        ioloop.start()
+    elif args.filename:
+        f = os.open(args.filename, os.O_RDONLY)
+        arg3 = PipeStream(f)
+        ioloop.run_sync(lambda: send_stream(arg1, arg2, arg3, args.host))
+    else:
+        raise NotImplementedError()
+
+if __name__ == '__main__':  # pragma: no cover
+    main()

examples/tchannel_server.py

@@ -23,7 +23,6 @@
 import tornado.ioloop
 
 from tchannel.tornado import TChannel
-
 from options import get_args
 from handler import get_example_handler
 

examples/tornado_client.py

@@ -28,14 +28,14 @@
 import tornado.ioloop
 
 from options import get_args
-from tchannel.tornado.connection import TornadoConnection
+from tchannel.tornado.connection import StreamConnection
 
 
 @tornado.gen.coroutine
 def main():
 
     args = get_args()
-    conn = yield TornadoConnection.outgoing('%s:%d' % (args.host, args.port))
+    conn = yield StreamConnection.outgoing('%s:%d' % (args.host, args.port))
 
     N = 10000
     before = time.time()

tchannel/dispatch.py

@@ -22,10 +22,3 @@
 #
 # Message ID 0xffffffff is reserved
 MAX_MESSAGE_ID = 0xfffffffe
-
-# Largest payload supported by the system.
-#
-# 64KB Max frame size
-# 16B (size:2 | type:1 | reserved:1 | id:4 | reserved:8)
-# 1 2 Bytes can represent 0~2**16-1
-MAX_PAYLOAD_SIZE = 0xFFEF   # 64*1024 - 16 - 1

tchannel/glossary.py

@@ -56,7 +56,8 @@ class BaseRequestHandler(RequestHandler):
 
     _HANDLER_NAMES = {
         Types.PING_REQ: 'ping',
-        Types.CALL_REQ: 'call'
+        Types.CALL_REQ: 'pre_call',
+        Types.CALL_REQ_CONTINUE: 'pre_call'
     }
 
     def __init__(self):
@@ -76,14 +77,33 @@ def handle(self, context, connection):
         handler_name = "handle_" + self._HANDLER_NAMES[message.message_type]
         return getattr(self, handler_name)(message_id, message, connection)
 
+    def handle_pre_call(self, message_id, message, connection):
+        """Handle incoming request message including CallRequestMessage and
+        CallRequestContinueMessage
+
+        This method will build the User friendly request object based on the
+        incoming messages.
+
+        It passes all the messages into the message_factory to build the init
+        request object. Only when it get a CallRequestMessage and a completed
+        arg_1=argstream[0], the message_factory will return a request object.
+        Then it will trigger the async call_handle call.
+
+        :param message_id: message id
+        :param message: CallRequestMessage or CallRequestContinueMessage
+        :param connection: tornado connection
+        """
+        req = connection.request_message_factory.build(message_id, message)
+        # call handler only for the call request message not continue message
+        if req:
+            self.handle_call(req, connection)
+
     def handle_ping(self, message_id, ping, connection):
         return connection.write(PingResponseMessage(), message_id)
 
-    def handle_call(self, message_id, call, connection):
+    def handle_call(self, call, connection):
         """Handle an incoming call.
 
-        :param message_id:
-            Message ID of the request
         :param call:
             CallRequestMessage containing information about the call
         :param connection:

tchannel/handler.py

@@ -20,6 +20,7 @@
 
 from __future__ import absolute_import
 
+
 try:
     from cStringIO import StringIO as BytesIO
 except ImportError:  # pragma: no cover