Reform cygrpc.Operation

It is now a family of classes conforming to an interface rather than a
single class no single instance of which makes use of all behavior
scoped to the class.

It also now only uses gRPC Core memory for the time of a single batch
rather than for the entire lifetime of the instance.

6 files changed, 136 insertions, 99 deletions

diff --git a/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py b/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py
index a8a7175cc7..458b4fe542 100644
--- a/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py
+++ b/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py
@@ -65,10 +65,10 @@ class _Handler(object):
 
         with self._lock:
             self._call.start_server_batch(
-                (cygrpc.operation_receive_close_on_server(_EMPTY_FLAGS),),
+                (cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS),),
                 _RECEIVE_CLOSE_ON_SERVER_TAG)
             self._call.start_server_batch(
-                (cygrpc.operation_receive_message(_EMPTY_FLAGS),),
+                (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),),
                 _RECEIVE_MESSAGE_TAG)
         first_event = self._completion_queue.poll()
         if _is_cancellation_event(first_event):
@@ -76,10 +76,10 @@ class _Handler(object):
         else:
             with self._lock:
                 operations = (
-                    cygrpc.operation_send_initial_metadata(_EMPTY_METADATA,
-                                                           _EMPTY_FLAGS),
-                    cygrpc.operation_send_message(b'\x79\x57', _EMPTY_FLAGS),
-                    cygrpc.operation_send_status_from_server(
+                    cygrpc.SendInitialMetadataOperation(_EMPTY_METADATA,
+                                                        _EMPTY_FLAGS),
+                    cygrpc.SendMessageOperation(b'\x79\x57', _EMPTY_FLAGS),
+                    cygrpc.SendStatusFromServerOperation(
                         _EMPTY_METADATA, cygrpc.StatusCode.ok, b'test details!',
                         _EMPTY_FLAGS),)
                 self._call.start_server_batch(operations,
@@ -169,13 +169,13 @@ class CancelManyCallsTest(unittest.TestCase):
                     None, _EMPTY_FLAGS, client_completion_queue, b'/twinkies',
                     None, _INFINITE_FUTURE)
                 operations = (
-                    cygrpc.operation_send_initial_metadata(_EMPTY_METADATA,
-                                                           _EMPTY_FLAGS),
-                    cygrpc.operation_send_message(b'\x45\x56', _EMPTY_FLAGS),
-                    cygrpc.operation_send_close_from_client(_EMPTY_FLAGS),
-                    cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),
-                    cygrpc.operation_receive_message(_EMPTY_FLAGS),
-                    cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS),)
+                    cygrpc.SendInitialMetadataOperation(_EMPTY_METADATA,
+                                                        _EMPTY_FLAGS),
+                    cygrpc.SendMessageOperation(b'\x45\x56', _EMPTY_FLAGS),
+                    cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
+                    cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),
+                    cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
+                    cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),)
                 tag = 'client_complete_call_{0:04d}_tag'.format(index)
                 client_call.start_client_batch(operations, tag)
                 client_due.add(tag)
diff --git a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py
index d08003af44..41291cc88f 100644
--- a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py
+++ b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py
@@ -49,18 +49,19 @@ class Test(_common.RpcTest, unittest.TestCase):
         with self.client_condition:
             client_receive_initial_metadata_start_batch_result = (
                 client_call.start_client_batch([
-                    cygrpc.operation_receive_initial_metadata(
-                        _common.EMPTY_FLAGS),
+                    cygrpc.ReceiveInitialMetadataOperation(_common.EMPTY_FLAGS),
                 ], client_receive_initial_metadata_tag))
+            self.assertEqual(cygrpc.CallError.ok,
+                             client_receive_initial_metadata_start_batch_result)
             client_complete_rpc_start_batch_result = client_call.start_client_batch(
                 [
-                    cygrpc.operation_send_initial_metadata(
+                    cygrpc.SendInitialMetadataOperation(
                         _common.INVOCATION_METADATA, _common.EMPTY_FLAGS),
-                    cygrpc.operation_send_close_from_client(
-                        _common.EMPTY_FLAGS),
-                    cygrpc.operation_receive_status_on_client(
-                        _common.EMPTY_FLAGS),
+                    cygrpc.SendCloseFromClientOperation(_common.EMPTY_FLAGS),
+                    cygrpc.ReceiveStatusOnClientOperation(_common.EMPTY_FLAGS),
                 ], client_complete_rpc_tag)
+            self.assertEqual(cygrpc.CallError.ok,
+                             client_complete_rpc_start_batch_result)
             self.client_driver.add_due({
                 client_receive_initial_metadata_tag,
                 client_complete_rpc_tag,
@@ -72,7 +73,7 @@ class Test(_common.RpcTest, unittest.TestCase):
         with server_call_condition:
             server_send_initial_metadata_start_batch_result = (
                 server_request_call_event.operation_call.start_server_batch([
-                    cygrpc.operation_send_initial_metadata(
+                    cygrpc.SendInitialMetadataOperation(
                         _common.INITIAL_METADATA, _common.EMPTY_FLAGS),
                 ], server_send_initial_metadata_tag))
             server_call_driver.add_due({
@@ -84,9 +85,8 @@ class Test(_common.RpcTest, unittest.TestCase):
         with server_call_condition:
             server_complete_rpc_start_batch_result = (
                 server_request_call_event.operation_call.start_server_batch([
-                    cygrpc.operation_receive_close_on_server(
-                        _common.EMPTY_FLAGS),
-                    cygrpc.operation_send_status_from_server(
+                    cygrpc.ReceiveCloseOnServerOperation(_common.EMPTY_FLAGS),
+                    cygrpc.SendStatusFromServerOperation(
                         _common.TRAILING_METADATA, cygrpc.StatusCode.ok,
                         b'test details', _common.EMPTY_FLAGS),
                 ], server_complete_rpc_tag))
diff --git a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py
index d0166a2b29..b429a20ed7 100644
--- a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py
+++ b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py
@@ -44,17 +44,14 @@ class Test(_common.RpcTest, unittest.TestCase):
         with self.client_condition:
             client_receive_initial_metadata_start_batch_result = (
                 client_call.start_client_batch([
-                    cygrpc.operation_receive_initial_metadata(
-                        _common.EMPTY_FLAGS),
+                    cygrpc.ReceiveInitialMetadataOperation(_common.EMPTY_FLAGS),
                 ], client_receive_initial_metadata_tag))
             client_complete_rpc_start_batch_result = client_call.start_client_batch(
                 [
-                    cygrpc.operation_send_initial_metadata(
+                    cygrpc.SendInitialMetadataOperation(
                         _common.INVOCATION_METADATA, _common.EMPTY_FLAGS),
-                    cygrpc.operation_send_close_from_client(
-                        _common.EMPTY_FLAGS),
-                    cygrpc.operation_receive_status_on_client(
-                        _common.EMPTY_FLAGS),
+                    cygrpc.SendCloseFromClientOperation(_common.EMPTY_FLAGS),
+                    cygrpc.ReceiveStatusOnClientOperation(_common.EMPTY_FLAGS),
                 ], client_complete_rpc_tag)
             self.client_driver.add_due({
                 client_receive_initial_metadata_tag,
@@ -67,7 +64,7 @@ class Test(_common.RpcTest, unittest.TestCase):
         with self.server_condition:
             server_send_initial_metadata_start_batch_result = (
                 server_request_call_event.operation_call.start_server_batch([
-                    cygrpc.operation_send_initial_metadata(
+                    cygrpc.SendInitialMetadataOperation(
                         _common.INITIAL_METADATA, _common.EMPTY_FLAGS),
                 ], server_send_initial_metadata_tag))
             self.server_driver.add_due({
@@ -79,11 +76,10 @@ class Test(_common.RpcTest, unittest.TestCase):
         with self.server_condition:
             server_complete_rpc_start_batch_result = (
                 server_request_call_event.operation_call.start_server_batch([
-                    cygrpc.operation_receive_close_on_server(
-                        _common.EMPTY_FLAGS),
-                    cygrpc.operation_send_status_from_server(
+                    cygrpc.ReceiveCloseOnServerOperation(_common.EMPTY_FLAGS),
+                    cygrpc.SendStatusFromServerOperation(
                         _common.TRAILING_METADATA, cygrpc.StatusCode.ok,
-                        b'test details', _common.EMPTY_FLAGS),
+                        'test details', _common.EMPTY_FLAGS),
                 ], server_complete_rpc_tag))
             self.server_driver.add_due({
                 server_complete_rpc_tag,
diff --git a/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py b/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py
index 1deb15ba03..a6d92f2203 100644
--- a/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py
+++ b/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py
@@ -158,15 +158,15 @@ class ReadSomeButNotAllResponsesTest(unittest.TestCase):
         with client_condition:
             client_receive_initial_metadata_start_batch_result = (
                 client_call.start_client_batch([
-                    cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),
+                    cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),
                 ], client_receive_initial_metadata_tag))
             client_due.add(client_receive_initial_metadata_tag)
             client_complete_rpc_start_batch_result = (
                 client_call.start_client_batch([
-                    cygrpc.operation_send_initial_metadata(_EMPTY_METADATA,
-                                                           _EMPTY_FLAGS),
-                    cygrpc.operation_send_close_from_client(_EMPTY_FLAGS),
-                    cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS),
+                    cygrpc.SendInitialMetadataOperation(_EMPTY_METADATA,
+                                                        _EMPTY_FLAGS),
+                    cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
+                    cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),
                 ], client_complete_rpc_tag))
             client_due.add(client_complete_rpc_tag)
 
@@ -175,12 +175,12 @@ class ReadSomeButNotAllResponsesTest(unittest.TestCase):
         with server_call_condition:
             server_send_initial_metadata_start_batch_result = (
                 server_rpc_event.operation_call.start_server_batch([
-                    cygrpc.operation_send_initial_metadata(_EMPTY_METADATA,
-                                                           _EMPTY_FLAGS),
+                    cygrpc.SendInitialMetadataOperation(_EMPTY_METADATA,
+                                                        _EMPTY_FLAGS),
                 ], server_send_initial_metadata_tag))
             server_send_first_message_start_batch_result = (
                 server_rpc_event.operation_call.start_server_batch([
-                    cygrpc.operation_send_message(b'\x07', _EMPTY_FLAGS),
+                    cygrpc.SendMessageOperation(b'\x07', _EMPTY_FLAGS),
                 ], server_send_first_message_tag))
         server_send_initial_metadata_event = server_call_driver.event_with_tag(
             server_send_initial_metadata_tag)
@@ -189,12 +189,12 @@ class ReadSomeButNotAllResponsesTest(unittest.TestCase):
         with server_call_condition:
             server_send_second_message_start_batch_result = (
                 server_rpc_event.operation_call.start_server_batch([
-                    cygrpc.operation_send_message(b'\x07', _EMPTY_FLAGS),
+                    cygrpc.SendMessageOperation(b'\x07', _EMPTY_FLAGS),
                 ], server_send_second_message_tag))
             server_complete_rpc_start_batch_result = (
                 server_rpc_event.operation_call.start_server_batch([
-                    cygrpc.operation_receive_close_on_server(_EMPTY_FLAGS),
-                    cygrpc.operation_send_status_from_server(
+                    cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS),
+                    cygrpc.SendStatusFromServerOperation(
                         (), cygrpc.StatusCode.ok, b'test details',
                         _EMPTY_FLAGS),
                 ], server_complete_rpc_tag))
@@ -208,7 +208,7 @@ class ReadSomeButNotAllResponsesTest(unittest.TestCase):
             client_receive_first_message_tag = 'client_receive_first_message_tag'
             client_receive_first_message_start_batch_result = (
                 client_call.start_client_batch([
-                    cygrpc.operation_receive_message(_EMPTY_FLAGS),
+                    cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
                 ], client_receive_first_message_tag))
             client_due.add(client_receive_first_message_tag)
         client_receive_first_message_event = client_driver.event_with_tag(
diff --git a/src/python/grpcio_tests/tests/unit/_cython/_server_test.py b/src/python/grpcio_tests/tests/unit/_cython/_server_test.py
new file mode 100644
index 0000000000..12bf40be6b
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/_cython/_server_test.py
@@ -0,0 +1,49 @@
+# Copyright 2017 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Test servers at the level of the Cython API."""
+
+import threading
+import time
+import unittest
+
+from grpc._cython import cygrpc
+
+
+class Test(unittest.TestCase):
+
+    def test_lonely_server(self):
+        server_call_completion_queue = cygrpc.CompletionQueue()
+        server_shutdown_completion_queue = cygrpc.CompletionQueue()
+        server = cygrpc.Server(cygrpc.ChannelArgs([]))
+        server.register_completion_queue(server_call_completion_queue)
+        server.register_completion_queue(server_shutdown_completion_queue)
+        port = server.add_http2_port(b'[::]:0')
+        server.start()
+
+        server_request_call_tag = 'server_request_call_tag'
+        server_request_call_start_batch_result = server.request_call(
+            server_call_completion_queue, server_call_completion_queue,
+            server_request_call_tag)
+
+        time.sleep(4)
+
+        server_shutdown_tag = 'server_shutdown_tag'
+        server_shutdown_result = server.shutdown(
+            server_shutdown_completion_queue, server_shutdown_tag)
+        server_request_call_event = server_call_completion_queue.poll()
+        server_shutdown_event = server_shutdown_completion_queue.poll()
+
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)
diff --git a/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py b/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py
index 4eda685486..002fb9b7f4 100644
--- a/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py
+++ b/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py
@@ -35,11 +35,6 @@ def _metadata_plugin(context, callback):
 
 class TypeSmokeTest(unittest.TestCase):
 
-    def testOperationFlags(self):
-        operation = cygrpc.operation_send_message(b'asdf',
-                                                  cygrpc.WriteFlag.no_compress)
-        self.assertEqual(cygrpc.WriteFlag.no_compress, operation.flags)
-
     def testTimespec(self):
         now = time.time()
         now_timespec_a = cygrpc.Timespec(now)
@@ -170,7 +165,7 @@ class ServerClientMixin(object):
         SERVER_TRAILING_METADATA_KEY = 'california_is_in_a_drought'
         SERVER_TRAILING_METADATA_VALUE = 'zomg it is'
         SERVER_STATUS_CODE = cygrpc.StatusCode.ok
-        SERVER_STATUS_DETAILS = b'our work is never over'
+        SERVER_STATUS_DETAILS = 'our work is never over'
         REQUEST = b'in death a member of project mayhem has a name'
         RESPONSE = b'his name is robert paulson'
         METHOD = b'twinkies'
@@ -192,13 +187,13 @@ class ServerClientMixin(object):
             (CLIENT_METADATA_ASCII_KEY, CLIENT_METADATA_ASCII_VALUE,),
             (CLIENT_METADATA_BIN_KEY, CLIENT_METADATA_BIN_VALUE,),)
         client_start_batch_result = client_call.start_client_batch([
-            cygrpc.operation_send_initial_metadata(client_initial_metadata,
-                                                   _EMPTY_FLAGS),
-            cygrpc.operation_send_message(REQUEST, _EMPTY_FLAGS),
-            cygrpc.operation_send_close_from_client(_EMPTY_FLAGS),
-            cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),
-            cygrpc.operation_receive_message(_EMPTY_FLAGS),
-            cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS)
+            cygrpc.SendInitialMetadataOperation(client_initial_metadata,
+                                                _EMPTY_FLAGS),
+            cygrpc.SendMessageOperation(REQUEST, _EMPTY_FLAGS),
+            cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
+            cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),
+            cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
+            cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),
         ], client_call_tag)
         self.assertEqual(cygrpc.CallError.ok, client_start_batch_result)
         client_event_future = test_utilities.CompletionQueuePollFuture(
@@ -227,12 +222,12 @@ class ServerClientMixin(object):
         server_trailing_metadata = (
             (SERVER_TRAILING_METADATA_KEY, SERVER_TRAILING_METADATA_VALUE,),)
         server_start_batch_result = server_call.start_server_batch([
-            cygrpc.operation_send_initial_metadata(
+            cygrpc.SendInitialMetadataOperation(
                 server_initial_metadata,
-                _EMPTY_FLAGS), cygrpc.operation_receive_message(_EMPTY_FLAGS),
-            cygrpc.operation_send_message(RESPONSE, _EMPTY_FLAGS),
-            cygrpc.operation_receive_close_on_server(_EMPTY_FLAGS),
-            cygrpc.operation_send_status_from_server(
+                _EMPTY_FLAGS), cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
+            cygrpc.SendMessageOperation(RESPONSE, _EMPTY_FLAGS),
+            cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS),
+            cygrpc.SendStatusFromServerOperation(
                 server_trailing_metadata, SERVER_STATUS_CODE,
                 SERVER_STATUS_DETAILS, _EMPTY_FLAGS)
         ], server_call_tag)
@@ -245,25 +240,24 @@ class ServerClientMixin(object):
         found_client_op_types = set()
         for client_result in client_event.batch_operations:
             # we expect each op type to be unique
-            self.assertNotIn(client_result.type, found_client_op_types)
-            found_client_op_types.add(client_result.type)
-            if client_result.type == cygrpc.OperationType.receive_initial_metadata:
+            self.assertNotIn(client_result.type(), found_client_op_types)
+            found_client_op_types.add(client_result.type())
+            if client_result.type(
+            ) == cygrpc.OperationType.receive_initial_metadata:
                 self.assertTrue(
                     test_common.metadata_transmitted(
                         server_initial_metadata,
-                        client_result.received_metadata))
-            elif client_result.type == cygrpc.OperationType.receive_message:
-                self.assertEqual(RESPONSE,
-                                 client_result.received_message.bytes())
-            elif client_result.type == cygrpc.OperationType.receive_status_on_client:
+                        client_result.initial_metadata()))
+            elif client_result.type() == cygrpc.OperationType.receive_message:
+                self.assertEqual(RESPONSE, client_result.message())
+            elif client_result.type(
+            ) == cygrpc.OperationType.receive_status_on_client:
                 self.assertTrue(
                     test_common.metadata_transmitted(
                         server_trailing_metadata,
-                        client_result.received_metadata))
-                self.assertEqual(SERVER_STATUS_DETAILS,
-                                 client_result.received_status_details)
-                self.assertEqual(SERVER_STATUS_CODE,
-                                 client_result.received_status_code)
+                        client_result.trailing_metadata()))
+                self.assertEqual(SERVER_STATUS_DETAILS, client_result.details())
+                self.assertEqual(SERVER_STATUS_CODE, client_result.code())
         self.assertEqual(
             set([
                 cygrpc.OperationType.send_initial_metadata,
@@ -277,13 +271,13 @@ class ServerClientMixin(object):
         self.assertEqual(5, len(server_event.batch_operations))
         found_server_op_types = set()
         for server_result in server_event.batch_operations:
-            self.assertNotIn(client_result.type, found_server_op_types)
-            found_server_op_types.add(server_result.type)
-            if server_result.type == cygrpc.OperationType.receive_message:
-                self.assertEqual(REQUEST,
-                                 server_result.received_message.bytes())
-            elif server_result.type == cygrpc.OperationType.receive_close_on_server:
-                self.assertFalse(server_result.received_cancelled)
+            self.assertNotIn(client_result.type(), found_server_op_types)
+            found_server_op_types.add(server_result.type())
+            if server_result.type() == cygrpc.OperationType.receive_message:
+                self.assertEqual(REQUEST, server_result.message())
+            elif server_result.type(
+            ) == cygrpc.OperationType.receive_close_on_server:
+                self.assertFalse(server_result.cancelled())
         self.assertEqual(
             set([
                 cygrpc.OperationType.send_initial_metadata,
@@ -319,9 +313,8 @@ class ServerClientMixin(object):
                                             cygrpc_deadline, description)
 
         client_event_future = perform_client_operations([
-            cygrpc.operation_send_initial_metadata(empty_metadata,
-                                                   _EMPTY_FLAGS),
-            cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),
+            cygrpc.SendInitialMetadataOperation(empty_metadata, _EMPTY_FLAGS),
+            cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),
         ], "Client prologue")
 
         request_event = self.server_completion_queue.poll(cygrpc_deadline)
@@ -333,8 +326,7 @@ class ServerClientMixin(object):
                                             cygrpc_deadline, description)
 
         server_event_future = perform_server_operations([
-            cygrpc.operation_send_initial_metadata(empty_metadata,
-                                                   _EMPTY_FLAGS),
+            cygrpc.SendInitialMetadataOperation(empty_metadata, _EMPTY_FLAGS),
         ], "Server prologue")
 
         client_event_future.result()  # force completion
@@ -343,12 +335,12 @@ class ServerClientMixin(object):
         # Messaging
         for _ in range(10):
             client_event_future = perform_client_operations([
-                cygrpc.operation_send_message(b'', _EMPTY_FLAGS),
-                cygrpc.operation_receive_message(_EMPTY_FLAGS),
+                cygrpc.SendMessageOperation(b'', _EMPTY_FLAGS),
+                cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
             ], "Client message")
             server_event_future = perform_server_operations([
-                cygrpc.operation_send_message(b'', _EMPTY_FLAGS),
-                cygrpc.operation_receive_message(_EMPTY_FLAGS),
+                cygrpc.SendMessageOperation(b'', _EMPTY_FLAGS),
+                cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),
             ], "Server receive")
 
             client_event_future.result()  # force completion
@@ -356,13 +348,13 @@ class ServerClientMixin(object):
 
         # Epilogue
         client_event_future = perform_client_operations([
-            cygrpc.operation_send_close_from_client(_EMPTY_FLAGS),
-            cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS)
+            cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS),
+            cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS)
         ], "Client epilogue")
 
         server_event_future = perform_server_operations([
-            cygrpc.operation_receive_close_on_server(_EMPTY_FLAGS),
-            cygrpc.operation_send_status_from_server(
+            cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS),
+            cygrpc.SendStatusFromServerOperation(
                 empty_metadata, cygrpc.StatusCode.ok, b'', _EMPTY_FLAGS)
         ], "Server epilogue")