mirror of https://github.com/grpc/grpc.git
Merge branch 'master' into revert-20610-revert-20277-grpc-19871/unary_unary_client_aio-implement-timeout
Lidi Zheng
5 years ago
committed by
GitHub
commit
84fd0a1ad0
20 changed files with 840 additions and 19 deletions
@ -0,0 +1,44 @@ |
||||
# Copyright 2019 The 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. |
||||
|
||||
cdef class _HandlerCallDetails: |
||||
cdef readonly str method |
||||
cdef readonly tuple invocation_metadata |
||||
|
||||
|
||||
cdef class RPCState: |
||||
cdef grpc_call* call, |
||||
cdef grpc_call_details details |
||||
cdef grpc_metadata_array request_metadata |
||||
|
||||
cdef bytes method(self) |
||||
|
||||
|
||||
cdef enum AioServerStatus: |
||||
AIO_SERVER_STATUS_UNKNOWN |
||||
AIO_SERVER_STATUS_READY |
||||
AIO_SERVER_STATUS_RUNNING |
||||
AIO_SERVER_STATUS_STOPPED |
||||
|
||||
|
||||
cdef class _CallbackCompletionQueue: |
||||
cdef grpc_completion_queue *_cq |
||||
cdef grpc_completion_queue* c_ptr(self) |
||||
|
||||
|
||||
cdef class AioServer: |
||||
cdef Server _server |
||||
cdef _CallbackCompletionQueue _cq |
||||
cdef list _generic_handlers |
||||
cdef AioServerStatus _status |
||||
@ -0,0 +1,275 @@ |
||||
# Copyright 2019 The 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. |
||||
|
||||
cdef class _HandlerCallDetails: |
||||
def __cinit__(self, str method, tuple invocation_metadata): |
||||
self.method = method |
||||
self.invocation_metadata = invocation_metadata |
||||
|
||||
|
||||
class _ServicerContextPlaceHolder(object): pass |
||||
|
||||
|
||||
# TODO(https://github.com/grpc/grpc/issues/20669) |
||||
# Apply this to the client-side |
||||
cdef class CallbackWrapper: |
||||
cdef CallbackContext context |
||||
cdef object _reference |
||||
|
||||
def __cinit__(self, object future): |
||||
self.context.functor.functor_run = self.functor_run |
||||
self.context.waiter = <cpython.PyObject*>(future) |
||||
self._reference = future |
||||
|
||||
@staticmethod |
||||
cdef void functor_run( |
||||
grpc_experimental_completion_queue_functor* functor, |
||||
int succeed): |
||||
cdef CallbackContext *context = <CallbackContext *>functor |
||||
if succeed == 0: |
||||
(<object>context.waiter).set_exception(RuntimeError()) |
||||
else: |
||||
(<object>context.waiter).set_result(None) |
||||
|
||||
cdef grpc_experimental_completion_queue_functor *c_functor(self): |
||||
return &self.context.functor |
||||
|
||||
|
||||
cdef class RPCState: |
||||
|
||||
def __cinit__(self): |
||||
grpc_metadata_array_init(&self.request_metadata) |
||||
grpc_call_details_init(&self.details) |
||||
|
||||
cdef bytes method(self): |
||||
return _slice_bytes(self.details.method) |
||||
|
||||
def __dealloc__(self): |
||||
"""Cleans the Core objects.""" |
||||
grpc_call_details_destroy(&self.details) |
||||
grpc_metadata_array_destroy(&self.request_metadata) |
||||
if self.call: |
||||
grpc_call_unref(self.call) |
||||
|
||||
|
||||
cdef _find_method_handler(str method, list generic_handlers): |
||||
# TODO(lidiz) connects Metadata to call details |
||||
cdef _HandlerCallDetails handler_call_details = _HandlerCallDetails( |
||||
method, |
||||
tuple() |
||||
) |
||||
|
||||
for generic_handler in generic_handlers: |
||||
method_handler = generic_handler.service(handler_call_details) |
||||
if method_handler is not None: |
||||
return method_handler |
||||
return None |
||||
|
||||
|
||||
async def callback_start_batch(RPCState rpc_state, |
||||
tuple operations, |
||||
object loop): |
||||
"""The callback version of start batch operations.""" |
||||
cdef _BatchOperationTag batch_operation_tag = _BatchOperationTag(None, operations, None) |
||||
batch_operation_tag.prepare() |
||||
|
||||
cdef object future = loop.create_future() |
||||
cdef CallbackWrapper wrapper = CallbackWrapper(future) |
||||
# NOTE(lidiz) Without Py_INCREF, the wrapper object will be destructed |
||||
# when calling "await". This is an over-optimization by Cython. |
||||
cpython.Py_INCREF(wrapper) |
||||
cdef grpc_call_error error = grpc_call_start_batch( |
||||
rpc_state.call, |
||||
batch_operation_tag.c_ops, |
||||
batch_operation_tag.c_nops, |
||||
wrapper.c_functor(), NULL) |
||||
|
||||
if error != GRPC_CALL_OK: |
||||
raise RuntimeError("Error with callback_start_batch {}".format(error)) |
||||
|
||||
await future |
||||
cpython.Py_DECREF(wrapper) |
||||
cdef grpc_event c_event |
||||
# Tag.event must be called, otherwise messages won't be parsed from C |
||||
batch_operation_tag.event(c_event) |
||||
|
||||
|
||||
async def _handle_unary_unary_rpc(object method_handler, |
||||
RPCState rpc_state, |
||||
object loop): |
||||
# Receives request message |
||||
cdef tuple receive_ops = ( |
||||
ReceiveMessageOperation(_EMPTY_FLAGS), |
||||
) |
||||
await callback_start_batch(rpc_state, receive_ops, loop) |
||||
|
||||
# Deserializes the request message |
||||
cdef bytes request_raw = receive_ops[0].message() |
||||
cdef object request_message |
||||
if method_handler.request_deserializer: |
||||
request_message = method_handler.request_deserializer(request_raw) |
||||
else: |
||||
request_message = request_raw |
||||
|
||||
# Executes application logic |
||||
cdef object response_message = await method_handler.unary_unary(request_message, _ServicerContextPlaceHolder()) |
||||
|
||||
# Serializes the response message |
||||
cdef bytes response_raw |
||||
if method_handler.response_serializer: |
||||
response_raw = method_handler.response_serializer(response_message) |
||||
else: |
||||
response_raw = response_message |
||||
|
||||
# Sends response message |
||||
cdef tuple send_ops = ( |
||||
SendStatusFromServerOperation( |
||||
tuple(), StatusCode.ok, b'', _EMPTY_FLAGS), |
||||
SendInitialMetadataOperation(tuple(), _EMPTY_FLAGS), |
||||
SendMessageOperation(response_raw, _EMPTY_FLAGS), |
||||
) |
||||
await callback_start_batch(rpc_state, send_ops, loop) |
||||
|
||||
|
||||
async def _handle_rpc(list generic_handlers, RPCState rpc_state, object loop): |
||||
# Finds the method handler (application logic) |
||||
cdef object method_handler = _find_method_handler( |
||||
rpc_state.method().decode(), |
||||
generic_handlers |
||||
) |
||||
if method_handler is None: |
||||
# TODO(lidiz) return unimplemented error to client side |
||||
raise NotImplementedError() |
||||
# TODO(lidiz) extend to all 4 types of RPC |
||||
if method_handler.request_streaming or method_handler.response_streaming: |
||||
raise NotImplementedError() |
||||
else: |
||||
await _handle_unary_unary_rpc( |
||||
method_handler, |
||||
rpc_state, |
||||
loop |
||||
) |
||||
|
||||
|
||||
async def _server_call_request_call(Server server, |
||||
_CallbackCompletionQueue cq, |
||||
object loop): |
||||
cdef grpc_call_error error |
||||
cdef RPCState rpc_state = RPCState() |
||||
cdef object future = loop.create_future() |
||||
cdef CallbackWrapper wrapper = CallbackWrapper(future) |
||||
# NOTE(lidiz) Without Py_INCREF, the wrapper object will be destructed |
||||
# when calling "await". This is an over-optimization by Cython. |
||||
cpython.Py_INCREF(wrapper) |
||||
error = grpc_server_request_call( |
||||
server.c_server, &rpc_state.call, &rpc_state.details, |
||||
&rpc_state.request_metadata, |
||||
cq.c_ptr(), cq.c_ptr(), |
||||
wrapper.c_functor() |
||||
) |
||||
if error != GRPC_CALL_OK: |
||||
raise RuntimeError("Error in _server_call_request_call: %s" % error) |
||||
|
||||
await future |
||||
cpython.Py_DECREF(wrapper) |
||||
return rpc_state |
||||
|
||||
|
||||
async def _server_main_loop(Server server, |
||||
_CallbackCompletionQueue cq, |
||||
list generic_handlers): |
||||
cdef object loop = asyncio.get_event_loop() |
||||
cdef RPCState rpc_state |
||||
|
||||
while True: |
||||
rpc_state = await _server_call_request_call( |
||||
server, |
||||
cq, |
||||
loop) |
||||
|
||||
loop.create_task(_handle_rpc(generic_handlers, rpc_state, loop)) |
||||
|
||||
|
||||
async def _server_start(Server server, |
||||
_CallbackCompletionQueue cq, |
||||
list generic_handlers): |
||||
server.start() |
||||
await _server_main_loop(server, cq, generic_handlers) |
||||
|
||||
|
||||
cdef class _CallbackCompletionQueue: |
||||
|
||||
def __cinit__(self): |
||||
self._cq = grpc_completion_queue_create_for_callback( |
||||
NULL, |
||||
NULL |
||||
) |
||||
|
||||
cdef grpc_completion_queue* c_ptr(self): |
||||
return self._cq |
||||
|
||||
|
||||
cdef class AioServer: |
||||
|
||||
def __init__(self, thread_pool, generic_handlers, interceptors, options, |
||||
maximum_concurrent_rpcs, compression): |
||||
self._server = Server(options) |
||||
self._cq = _CallbackCompletionQueue() |
||||
self._status = AIO_SERVER_STATUS_READY |
||||
self._generic_handlers = [] |
||||
grpc_server_register_completion_queue( |
||||
self._server.c_server, |
||||
self._cq.c_ptr(), |
||||
NULL |
||||
) |
||||
self.add_generic_rpc_handlers(generic_handlers) |
||||
|
||||
if interceptors: |
||||
raise NotImplementedError() |
||||
if maximum_concurrent_rpcs: |
||||
raise NotImplementedError() |
||||
if compression: |
||||
raise NotImplementedError() |
||||
if thread_pool: |
||||
raise NotImplementedError() |
||||
|
||||
def add_generic_rpc_handlers(self, generic_rpc_handlers): |
||||
for h in generic_rpc_handlers: |
||||
self._generic_handlers.append(h) |
||||
|
||||
def add_insecure_port(self, address): |
||||
return self._server.add_http2_port(address) |
||||
|
||||
def add_secure_port(self, address, server_credentials): |
||||
return self._server.add_http2_port(address, |
||||
server_credentials._credentials) |
||||
|
||||
async def start(self): |
||||
if self._status == AIO_SERVER_STATUS_RUNNING: |
||||
return |
||||
elif self._status != AIO_SERVER_STATUS_READY: |
||||
raise RuntimeError('Server not in ready state') |
||||
|
||||
self._status = AIO_SERVER_STATUS_RUNNING |
||||
loop = asyncio.get_event_loop() |
||||
loop.create_task(_server_start( |
||||
self._server, |
||||
self._cq, |
||||
self._generic_handlers, |
||||
)) |
||||
|
||||
# TODO(https://github.com/grpc/grpc/issues/20668) |
||||
# Implement Destruction Methods for AsyncIO Server |
||||
def stop(self, unused_grace): |
||||
pass |
||||
@ -0,0 +1,179 @@ |
||||
# Copyright 2019 The 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. |
||||
"""Server-side implementation of gRPC Asyncio Python.""" |
||||
|
||||
from typing import Text, Optional |
||||
import asyncio |
||||
import grpc |
||||
from grpc import _common |
||||
from grpc._cython import cygrpc |
||||
|
||||
|
||||
class Server: |
||||
"""Serves RPCs.""" |
||||
|
||||
def __init__(self, thread_pool, generic_handlers, interceptors, options, |
||||
maximum_concurrent_rpcs, compression): |
||||
self._server = cygrpc.AioServer(thread_pool, generic_handlers, |
||||
interceptors, options, |
||||
maximum_concurrent_rpcs, compression) |
||||
|
||||
def add_generic_rpc_handlers( |
||||
self, |
||||
generic_rpc_handlers, |
||||
# generic_rpc_handlers: Iterable[grpc.GenericRpcHandlers] |
||||
) -> None: |
||||
"""Registers GenericRpcHandlers with this Server. |
||||
|
||||
This method is only safe to call before the server is started. |
||||
|
||||
Args: |
||||
generic_rpc_handlers: An iterable of GenericRpcHandlers that will be |
||||
used to service RPCs. |
||||
""" |
||||
self._server.add_generic_rpc_handlers(generic_rpc_handlers) |
||||
|
||||
def add_insecure_port(self, address: Text) -> int: |
||||
"""Opens an insecure port for accepting RPCs. |
||||
|
||||
This method may only be called before starting the server. |
||||
|
||||
Args: |
||||
address: The address for which to open a port. If the port is 0, |
||||
or not specified in the address, then the gRPC runtime will choose a port. |
||||
|
||||
Returns: |
||||
An integer port on which the server will accept RPC requests. |
||||
""" |
||||
return self._server.add_insecure_port(_common.encode(address)) |
||||
|
||||
def add_secure_port(self, address: Text, |
||||
server_credentials: grpc.ServerCredentials) -> int: |
||||
"""Opens a secure port for accepting RPCs. |
||||
|
||||
This method may only be called before starting the server. |
||||
|
||||
Args: |
||||
address: The address for which to open a port. |
||||
if the port is 0, or not specified in the address, then the gRPC |
||||
runtime will choose a port. |
||||
server_credentials: A ServerCredentials object. |
||||
|
||||
Returns: |
||||
An integer port on which the server will accept RPC requests. |
||||
""" |
||||
return self._server.add_secure_port( |
||||
_common.encode(address), server_credentials) |
||||
|
||||
async def start(self) -> None: |
||||
"""Starts this Server. |
||||
|
||||
This method may only be called once. (i.e. it is not idempotent). |
||||
""" |
||||
await self._server.start() |
||||
|
||||
def stop(self, grace: Optional[float]) -> asyncio.Event: |
||||
"""Stops this Server. |
||||
|
||||
"This method immediately stops the server from servicing new RPCs in |
||||
all cases. |
||||
|
||||
If a grace period is specified, this method returns immediately |
||||
and all RPCs active at the end of the grace period are aborted. |
||||
If a grace period is not specified (by passing None for `grace`), |
||||
all existing RPCs are aborted immediately and this method |
||||
blocks until the last RPC handler terminates. |
||||
|
||||
This method is idempotent and may be called at any time. |
||||
Passing a smaller grace value in a subsequent call will have |
||||
the effect of stopping the Server sooner (passing None will |
||||
have the effect of stopping the server immediately). Passing |
||||
a larger grace value in a subsequent call *will not* have the |
||||
effect of stopping the server later (i.e. the most restrictive |
||||
grace value is used). |
||||
|
||||
Args: |
||||
grace: A duration of time in seconds or None. |
||||
|
||||
Returns: |
||||
A threading.Event that will be set when this Server has completely |
||||
stopped, i.e. when running RPCs either complete or are aborted and |
||||
all handlers have terminated. |
||||
""" |
||||
raise NotImplementedError() |
||||
|
||||
async def wait_for_termination(self, |
||||
timeout: Optional[float] = None) -> bool: |
||||
"""Block current coroutine until the server stops. |
||||
|
||||
This is an EXPERIMENTAL API. |
||||
|
||||
The wait will not consume computational resources during blocking, and |
||||
it will block until one of the two following conditions are met: |
||||
|
||||
1) The server is stopped or terminated; |
||||
2) A timeout occurs if timeout is not `None`. |
||||
|
||||
The timeout argument works in the same way as `threading.Event.wait()`. |
||||
https://docs.python.org/3/library/threading.html#threading.Event.wait |
||||
|
||||
Args: |
||||
timeout: A floating point number specifying a timeout for the |
||||
operation in seconds. |
||||
|
||||
Returns: |
||||
A bool indicates if the operation times out. |
||||
""" |
||||
if timeout: |
||||
raise NotImplementedError() |
||||
# TODO(lidiz) replace this wait forever logic |
||||
future = asyncio.get_event_loop().create_future() |
||||
await future |
||||
|
||||
|
||||
def server(migration_thread_pool=None, |
||||
handlers=None, |
||||
interceptors=None, |
||||
options=None, |
||||
maximum_concurrent_rpcs=None, |
||||
compression=None): |
||||
"""Creates a Server with which RPCs can be serviced. |
||||
|
||||
Args: |
||||
migration_thread_pool: A futures.ThreadPoolExecutor to be used by the |
||||
Server to execute non-AsyncIO RPC handlers for migration purpose. |
||||
handlers: An optional list of GenericRpcHandlers used for executing RPCs. |
||||
More handlers may be added by calling add_generic_rpc_handlers any time |
||||
before the server is started. |
||||
interceptors: An optional list of ServerInterceptor objects that observe |
||||
and optionally manipulate the incoming RPCs before handing them over to |
||||
handlers. The interceptors are given control in the order they are |
||||
specified. This is an EXPERIMENTAL API. |
||||
options: An optional list of key-value pairs (channel args in gRPC runtime) |
||||
to configure the channel. |
||||
maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server |
||||
will service before returning RESOURCE_EXHAUSTED status, or None to |
||||
indicate no limit. |
||||
compression: An element of grpc.compression, e.g. |
||||
grpc.compression.Gzip. This compression algorithm will be used for the |
||||
lifetime of the server unless overridden. This is an EXPERIMENTAL option. |
||||
|
||||
Returns: |
||||
A Server object. |
||||
""" |
||||
return Server(migration_thread_pool, () |
||||
if handlers is None else handlers, () |
||||
if interceptors is None else interceptors, () |
||||
if options is None else options, maximum_concurrent_rpcs, |
||||
compression) |
||||
@ -0,0 +1,52 @@ |
||||
# Copyright 2019 The 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. |
||||
|
||||
import asyncio |
||||
import logging |
||||
import unittest |
||||
|
||||
from grpc.experimental import aio |
||||
from src.proto.grpc.testing import messages_pb2 |
||||
from src.proto.grpc.testing import benchmark_service_pb2_grpc |
||||
|
||||
|
||||
class BenchmarkServer(benchmark_service_pb2_grpc.BenchmarkServiceServicer): |
||||
|
||||
async def UnaryCall(self, request, context): |
||||
payload = messages_pb2.Payload(body=b'\0' * request.response_size) |
||||
return messages_pb2.SimpleResponse(payload=payload) |
||||
|
||||
|
||||
async def _start_async_server(): |
||||
server = aio.server() |
||||
|
||||
port = server.add_insecure_port(('localhost:%s' % 50051).encode('ASCII')) |
||||
servicer = BenchmarkServer() |
||||
benchmark_service_pb2_grpc.add_BenchmarkServiceServicer_to_server( |
||||
servicer, server) |
||||
|
||||
await server.start() |
||||
await server.wait_for_termination() |
||||
|
||||
|
||||
def main(): |
||||
aio.init_grpc_aio() |
||||
loop = asyncio.get_event_loop() |
||||
loop.create_task(_start_async_server()) |
||||
loop.run_forever() |
||||
|
||||
|
||||
if __name__ == '__main__': |
||||
logging.basicConfig() |
||||
main() |
||||
@ -0,0 +1,43 @@ |
||||
# Copyright 2019 The 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. |
||||
|
||||
package( |
||||
default_testonly = 1, |
||||
default_visibility = ["//visibility:public"], |
||||
) |
||||
|
||||
GRPC_ASYNC_TESTS = [ |
||||
"server_test.py", |
||||
] |
||||
|
||||
[ |
||||
py_test( |
||||
name=test_file_name[:-3], |
||||
size="small", |
||||
srcs=[test_file_name], |
||||
main=test_file_name, |
||||
python_version="PY3", |
||||
deps=[ |
||||
"//src/python/grpcio/grpc:grpcio", |
||||
"//src/proto/grpc/testing:py_messages_proto", |
||||
"//src/proto/grpc/testing:benchmark_service_py_pb2_grpc", |
||||
"//src/proto/grpc/testing:benchmark_service_py_pb2", |
||||
"//external:six" |
||||
], |
||||
imports=["../../",], |
||||
data=[ |
||||
"//src/python/grpcio_tests/tests/unit/credentials", |
||||
], |
||||
) for test_file_name in GRPC_ASYNC_TESTS |
||||
] |
||||
@ -0,0 +1,62 @@ |
||||
# Copyright 2019 The 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. |
||||
|
||||
import asyncio |
||||
import logging |
||||
import unittest |
||||
|
||||
import grpc |
||||
from grpc.experimental import aio |
||||
from src.proto.grpc.testing import messages_pb2 |
||||
from src.proto.grpc.testing import benchmark_service_pb2_grpc |
||||
|
||||
_TEST_METHOD_PATH = '' |
||||
|
||||
_REQUEST = b'\x00\x00\x00' |
||||
_RESPONSE = b'\x01\x01\x01' |
||||
|
||||
|
||||
async def unary_unary(unused_request, unused_context): |
||||
return _RESPONSE |
||||
|
||||
|
||||
class GenericHandler(grpc.GenericRpcHandler): |
||||
|
||||
def service(self, unused_handler_details): |
||||
return grpc.unary_unary_rpc_method_handler(unary_unary) |
||||
|
||||
|
||||
class TestServer(unittest.TestCase): |
||||
|
||||
def test_unary_unary(self): |
||||
loop = asyncio.get_event_loop() |
||||
|
||||
async def test_unary_unary_body(): |
||||
server = aio.server() |
||||
port = server.add_insecure_port('[::]:0') |
||||
server.add_generic_rpc_handlers((GenericHandler(),)) |
||||
await server.start() |
||||
|
||||
async with aio.insecure_channel('localhost:%d' % port) as channel: |
||||
unary_call = channel.unary_unary(_TEST_METHOD_PATH) |
||||
response = await unary_call(_REQUEST) |
||||
self.assertEqual(response, _RESPONSE) |
||||
|
||||
loop.run_until_complete(test_unary_unary_body()) |
||||
|
||||
|
||||
if __name__ == '__main__': |
||||
aio.init_grpc_aio() |
||||
logging.basicConfig() |
||||
unittest.main(verbosity=2) |
||||
Loading…
Reference in new issue