commit 9bba7b7af53c17ed22283b871c2708ca6b201af3
author Lidi Zheng <lidiz@google.com> Wed Jan 08 10:20:26 2020 -0800
committer Lidi Zheng <lidiz@google.com> Tue Jan 14 12:12:07 2020 -0800
tree c733eabe0bda67a2bd8283455fb9c3887bf53501
parent 26367cecc239ca7b62ba1800a58df88da96a2549

Improve readability

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi
index 4659492..ed1a61e 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi
@@ -141,15 +141,25 @@
               object code,
               str details='',
               tuple trailing_metadata=_EMPTY_METADATA):
-        await _perform_abort(
-            self._rpc_state,
-            code.value[0],
-            details,
-            trailing_metadata,
-            self._loop
-        )
+        if self._rpc_state.abort_exception is not None:
+            raise RuntimeError('Abort already called!')
+        else:
+            # Keeps track of the exception object. After abort happen, the RPC
+            # should stop execution. However, if users decided to suppress it, it
+            # could lead to undefined behavior.
+            self._rpc_state.abort_exception = AbortError('Locally aborted.')
 
-        raise self._rpc_state.abort_exception
+            self._rpc_state.status_sent = True
+            await _send_error_status_from_server(
+                self._rpc_state,
+                code.value[0],
+                details,
+                trailing_metadata,
+                self._rpc_state.metadata_sent,
+                self._loop
+            )
+
+            raise self._rpc_state.abort_exception
 
 
 cdef _find_method_handler(str method, list generic_handlers):
@@ -207,8 +217,8 @@
         SendInitialMetadataOperation(None, _EMPTY_FLAGS),
         SendMessageOperation(response_raw, _EMPTY_FLAGS),
     )
-    await execute_batch(rpc_state, send_ops, loop)
     rpc_state.status_sent = True
+    await execute_batch(rpc_state, send_ops, loop)
 
 
 async def _handle_unary_stream_rpc(object method_handler,
@@ -270,8 +280,8 @@
     )
 
     cdef tuple ops = (op,)
-    await execute_batch(rpc_state, ops, loop)
     rpc_state.status_sent = True
+    await execute_batch(rpc_state, ops, loop)
 
 
 async def _handle_exceptions(RPCState rpc_state, object rpc_coro, object loop):
@@ -292,11 +302,12 @@
     except Exception as e:
         _LOGGER.exception(e)
         if not rpc_state.status_sent and rpc_state.server._status != AIO_SERVER_STATUS_STOPPED:
-            await _perform_abort(
+            await _send_error_status_from_server(
                 rpc_state,
                 StatusCode.unknown,
                 '%s: %s' % (type(e), e),
                 _EMPTY_METADATA,
+                rpc_state.metadata_sent,
                 loop
             )
 
@@ -333,11 +344,13 @@
         generic_handlers,
     )
     if method_handler is None:
-        await _perform_abort(
+        rpc_state.status_sent = True
+        await _send_error_status_from_server(
             rpc_state,
             StatusCode.unimplemented,
             b'Method not found!',
             _EMPTY_METADATA,
+            rpc_state.metadata_sent,
             loop
         )
         return