Merge branch 'master' into issue-121

CHANGES.rst

@@ -3,14 +3,28 @@ Changelog
 
 0.4.x
 ~~~~~
+Version 0.4.11
+-------------
+
+Released on Mar 17, 2020.
+
+- Support Cython in HTTP and fix TCyBufferedTransport early flush issue, via `2-#129`_.
+- Fix exception handling in TProcessor, via `2-#128`_.
+- Rename socket_timeout to timeout for compatibility, via `2-#115`_.
+
+.. _2-#115: https://github.com/Thriftpy/thriftpy2/pull/115
+.. _2-#128: https://github.com/Thriftpy/thriftpy2/pull/128
+.. _2-#129: https://github.com/Thriftpy/thriftpy2/pull/129
+
+
 Version 0.4.10
 -------------
 
 Released on Jan 1, 2020.
 
 - Add TAsyncCompactProtocol and TAsyncFramedTransport, via `2-#103`_.
 - Add TAsyncProtocolBase and TAsyncTransportBase, via `2-#108`_.
-- Add __str__ on TProtocolException, , via `2-#109`_.
+- Add __str__ on TProtocolException, via `2-#109`_.
 - Support passing socket_family in make_client, via `2-#110`_.
 
 .. _2-#103: https://github.com/Thriftpy/thriftpy2/pull/103

tests/compatible/version_2/tracking.py

@@ -192,7 +192,8 @@ def _do_process(self, iprot, oprot, api, seqid, result, call):
             result.success = call()
         except Exception as e:
             # raise if api don't have throws
-            self.handle_exception(e, result)
+            if not self.handle_exception(e, result):
+                raise
 
         if not result.oneway:
             self.send_result(oprot, api, result, seqid)

tests/test_aio.py

@@ -50,6 +50,9 @@ def add(self, person):
 
     @asyncio.coroutine
     def remove(self, name):
+        if not name:
+            # undeclared exception
+            raise ValueError('name cannot be empty')
         try:
             self.ab.people.pop(name)
             return True
@@ -210,6 +213,13 @@ async def test_exception(self):
             await c.remove("Bob")
         c.close()
 
+    @pytest.mark.asyncio
+    async def test_undeclared_exception(self):
+        c = await self.client()
+        with pytest.raises(TTransportException):
+            await c.remove('')
+        c.close()
+
     @pytest.mark.asyncio
     async def test_client_socket_timeout(self):
         c = await self.client(timeout=500)

tests/test_tornado.py

@@ -13,6 +13,7 @@
 import thriftpy2
 from thriftpy2.tornado import make_client
 from thriftpy2.tornado import make_server
+from thriftpy2.transport import TTransportException
 
 
 logging.basicConfig(level=logging.INFO)
@@ -39,6 +40,9 @@ def get(self, name):
         """
         Person get(1: string name) throws (1: PersonNotExistsError not_exists);
         """
+        if not name:
+            # undeclared exception
+            raise ValueError('name cannot be empty')
         if name not in self.registry:
             raise addressbook.PersonNotExistsError(
                 'Person "{}" does not exist!'.format(name))
@@ -51,6 +55,9 @@ def remove(self, name):
         """
         # delay action for later
         yield gen.Task(self.io_loop.add_callback)
+        if not name:
+            # undeclared exception
+            raise ValueError('name cannot be empty')
         if name not in self.registry:
             raise addressbook.PersonNotExistsError(
                 'Person "{}" does not exist!'.format(name))
@@ -126,6 +133,17 @@ def test_synchronous_exception(self):
 
         assert isinstance(exc, addressbook.PersonNotExistsError)
 
+    @testing.gen_test
+    @pytest.mark.skipif(sys.version_info[:2] == (2, 6), reason="not support")
+    def test_synchronous_undeclared_exception(self):
+        exc = None
+        try:
+            yield self.client.get('')
+        except Exception as e:
+            exc = e
+
+        assert isinstance(exc, TTransportException)
+
     @testing.gen_test
     @pytest.mark.skipif(sys.version_info[:2] == (2, 6), reason="not support")
     def test_asynchronous_result(self):
@@ -143,3 +161,13 @@ def test_asynchronous_exception(self):
         except Exception as e:
             exc = e
         assert isinstance(exc, addressbook.PersonNotExistsError)
+
+    @testing.gen_test
+    @pytest.mark.skipif(sys.version_info[:2] == (2, 6), reason="not support")
+    def test_asynchronous_undeclared_exception(self):
+        exc = None
+        try:
+            yield self.client.remove('')
+        except Exception as e:
+            exc = e
+        assert isinstance(exc, TTransportException)

tests/test_tracking.py

@@ -50,6 +50,13 @@
     TrackerBase as TrackerBaseV2,
 )
 
+try:
+    from pytest_cov.embed import cleanup_on_sigterm
+except ImportError:
+    pass
+else:
+    cleanup_on_sigterm()
+
 addressbook = thriftpy2.load(os.path.join(os.path.dirname(__file__),
                                           "addressbook.thrift"))
 _, db_file = tempfile.mkstemp()
@@ -135,6 +142,9 @@ def add(self, person):
         return True
 
     def get(self, name):
+        if not name:
+            # undeclared exception
+            raise ValueError('name cannot be empty')
         raise addressbook.PersonNotExistsError()
 
 
@@ -161,9 +171,9 @@ def handle(self, client):
             pass
         except Exception:
             raise
-
-        itrans.close()
-        otrans.close()
+        finally:
+            itrans.close()
+            otrans.close()
 
 
 def gen_server(port, tracker=tracker, processor=TTrackedProcessor):
@@ -188,6 +198,7 @@ def server(request):
     def fin():
         if ps.is_alive():
             ps.terminate()
+            ps.join()
 
     request.addfinalizer(fin)
     return ser
@@ -201,6 +212,7 @@ def server1(request):
     def fin():
         if ps.is_alive():
             ps.terminate()
+            ps.join()
 
     request.addfinalizer(fin)
     return ser
@@ -214,6 +226,7 @@ def server2(request):
     def fin():
         if ps.is_alive():
             ps.terminate()
+            ps.join()
 
     request.addfinalizer(fin)
     return ser
@@ -227,6 +240,7 @@ def native_server(request):
     def fin():
         if ps.is_alive():
             ps.terminate()
+            ps.join()
 
     request.addfinalizer(fin)
     return ser
@@ -241,6 +255,7 @@ def tracked_server_v2(request):
     def fin():
         if ps.is_alive():
             ps.terminate()
+            ps.join()
 
     request.addfinalizer(fin)
     return ser
@@ -368,6 +383,12 @@ def test_exception(server, dbm_db, tracker_ctx):
     assert header["status"] is False
 
 
+def test_undeclared_exception(server, dbm_db, tracker_ctx):
+    with pytest.raises(TTransportException):
+        with client() as c:
+            c.get('')
+
+
 def test_request_id_func():
     ctx.__dict__.clear()
 

thriftpy2/__init__.py

@@ -5,7 +5,7 @@
 from .hook import install_import_hook, remove_import_hook
 from .parser import load, load_module, load_fp
 
-__version__ = '0.4.10'
+__version__ = '0.4.11'
 __python__ = sys.version_info
 __all__ = ["install_import_hook", "remove_import_hook", "load", "load_module",
            "load_fp"]

thriftpy2/contrib/aio/processor.py

@@ -54,9 +54,8 @@ def handle_exception(self, e, result):
             _, exc_name, exc_cls, _ = result.thrift_spec[k]
             if isinstance(e, exc_cls):
                 setattr(result, exc_name, e)
-                break
-        else:
-            raise
+                return True
+        return False
 
     @asyncio.coroutine
     def process(self, iprot, oprot):
@@ -69,7 +68,8 @@ def process(self, iprot, oprot):
             result.success = yield from call()
         except Exception as e:
             # raise if api don't have throws
-            self.handle_exception(e, result)
+            if not self.handle_exception(e, result):
+                raise
 
         if not result.oneway:
             yield from self.send_result(oprot, api, result, seqid)

thriftpy2/contrib/tracking/__init__.py

@@ -211,7 +211,8 @@ def _do_process(self, iprot, oprot, api, seqid, result, call):
             result.success = call()
         except Exception as e:
             # raise if api don't have throws
-            self.handle_exception(e, result)
+            if not self.handle_exception(e, result):
+                raise
 
         if not result.oneway:
             if self.check_version(

thriftpy2/http.py

@@ -54,23 +54,10 @@
 
 from thriftpy2.thrift import TProcessor, TClient
 from thriftpy2.server import TServer
-from thriftpy2.transport import (
-    TTransportBase,
-    TMemoryBuffer
-)
-# Explicitly use Python version instead of Cython version for libraries below
-# to address some mystery issues for now.
-#
-# Avoid TypeError: Cannot convert TBufferedTransport to
-# thriftpy2.transport.cybase.CyTransportBase.
-from thriftpy2.protocol.binary import TBinaryProtocolFactory
-# Avoid raised error of too small buffer allocated by TCyBufferedTransport.
-# Also, using TCyBufferedTransportFactory will let THttpClient write a broken
-# string to server, which making server freezed in transport.readall() method.
-from thriftpy2.transport.buffered import (
-    TBufferedTransport,
-    TBufferedTransportFactory,
-)
+from thriftpy2.transport import TTransportBase, TMemoryBuffer
+
+from thriftpy2.protocol import TBinaryProtocolFactory
+from thriftpy2.transport import TBufferedTransportFactory
 
 
 HTTP_URI = '{scheme}://{host}:{port}{path}'
@@ -120,14 +107,16 @@ class THttpServer(TServer):
     def __init__(self,
                  processor,
                  server_address,
+                 itrans_factory,
                  iprot_factory,
                  server_class=http_server.HTTPServer):
         """Set up protocol factories and HTTP server.
         See http.server for server_address.
         See TServer for protocol factories.
         """
         TServer.__init__(self, processor, trans=None,
-                         itrans_factory=None, iprot_factory=iprot_factory,
+                         itrans_factory=itrans_factory,
+                         iprot_factory=iprot_factory,
                          otrans_factory=None, oprot_factory=None)
 
         thttpserver = self
@@ -137,12 +126,18 @@ class RequestHander(http_server.BaseHTTPRequestHandler):
 
             def do_POST(self):
                 # Don't care about the request path.
-                itrans = TFileObjectTransport(self.rfile)
-                otrans = TFileObjectTransport(self.wfile)
-                itrans = TBufferedTransport(
-                    itrans, int(self.headers['Content-Length']))
-                otrans = TMemoryBuffer()
+                # Pre-read all of the data into a BytesIO. Buffered transport
+                # was previously configured to read everything on the first
+                # consumption, but that was a hack relying on the internal
+                # mechanism and prevents other transports from working, so
+                # replicate that properly to prevent timeout issues
+                content_len = int(self.headers['Content-Length'])
+                buf = BytesIO(self.rfile.read(content_len))
+                itrans = TFileObjectTransport(buf)
+                itrans = thttpserver.itrans_factory.get_transport(itrans)
                 iprot = thttpserver.iprot_factory.get_protocol(itrans)
+
+                otrans = TMemoryBuffer()
                 oprot = thttpserver.oprot_factory.get_protocol(otrans)
                 try:
                     thttpserver.processor.process(iprot, oprot)
@@ -222,13 +217,16 @@ def write(self, buf):
         self.__wbuf.write(buf)
 
     def flush(self):
-        if self.isOpen():
-            self.close()
-        self.open()
-
         # Pull data out of buffer
+        # Do this before opening a new connection in case there isn't data
         data = self.__wbuf.getvalue()
         self.__wbuf = BytesIO()
+        if not data:  # No data to flush, ignore
+            return
+
+        if self.isOpen():
+            self.close()
+        self.open()
 
         # HTTP request
         self.__http.putrequest('POST', self.path, skip_host=True)
@@ -323,8 +321,10 @@ def client_context(service, host='localhost', port=9090, path='', scheme='http',
 
 
 def make_server(service, handler, host, port,
-                proto_factory=TBinaryProtocolFactory()):
+                proto_factory=TBinaryProtocolFactory(),
+                trans_factory=TBufferedTransportFactory()):
     processor = TProcessor(service, handler)
     server = THttpServer(processor, (host, port),
+                         itrans_factory=trans_factory,
                          iprot_factory=proto_factory)
     return server

thriftpy2/protocol/cybin/cybin.pyx

@@ -460,7 +460,7 @@ cdef class TCyBinaryProtocol(object):
         write_i32(self.trans, seqid)
 
     def write_message_end(self):
-        self.trans.c_flush()
+        pass
 
     def read_struct(self, obj):
         try:

thriftpy2/tornado.py

@@ -211,7 +211,8 @@ def handle_stream(self, stream, address):
                         result.success = yield gen.maybe_future(call())
                     except Exception as e:
                         # raise if api don't have throws
-                        self._processor.handle_exception(e, result)
+                        if not self._processor.handle_exception(e, result):
+                            raise
 
                     self._processor.send_result(oprot, api, result, seqid)
         except Exception: