codekingpro/portable-devtools

1# Copyright 2019 The gRPC Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""Server-side implementation of gRPC Asyncio Python."""
15
16from concurrent.futures import Executor
17from typing import Any, Dict, Optional, Sequence
18
19import grpc
20from grpc import _common
21from grpc import _compression
22from grpc._cython import cygrpc
23
24from . import _base_server
25from ._interceptor import ServerInterceptor
26from ._typing import ChannelArgumentType
27
28
29def _augment_channel_arguments(
30    base_options: ChannelArgumentType, compression: Optional[grpc.Compression]
31):
32    compression_option = _compression.create_channel_option(compression)
33    return tuple(base_options) + compression_option
34
35
36class Server(_base_server.Server):
37    """Serves RPCs."""
38
39    def __init__(
40        self,
41        thread_pool: Optional[Executor],
42        generic_handlers: Optional[Sequence[grpc.GenericRpcHandler]],
43        interceptors: Optional[Sequence[Any]],
44        options: ChannelArgumentType,
45        maximum_concurrent_rpcs: Optional[int],
46        compression: Optional[grpc.Compression],
47    ):
48        self._loop = cygrpc.get_working_loop()
49        if interceptors:
50            invalid_interceptors = [
51                interceptor
52                for interceptor in interceptors
53                if not isinstance(interceptor, ServerInterceptor)
54            ]
55            if invalid_interceptors:
56                error_msg = (
57                    "Interceptor must be ServerInterceptor,"
58                    "the following are invalid: {invalid_interceptors}"
59                )
60                # TODO(asheshvidyut): fix the value error below
61                # not caught by ruff.
62                raise ValueError(error_msg)
63        self._server = cygrpc.AioServer(
64            self._loop,
65            thread_pool,
66            generic_handlers,
67            interceptors,
68            _augment_channel_arguments(options, compression),
69            maximum_concurrent_rpcs,
70        )
71
72    def add_generic_rpc_handlers(
73        self, generic_rpc_handlers: Sequence[grpc.GenericRpcHandler]
74    ) -> None:
75        """Registers GenericRpcHandlers with this Server.
76
77        This method is only safe to call before the server is started.
78
79        Args:
80          generic_rpc_handlers: A sequence of GenericRpcHandlers that will be
81          used to service RPCs.
82        """
83        self._server.add_generic_rpc_handlers(generic_rpc_handlers)
84
85    def add_registered_method_handlers(
86        self,
87        service_name: str,
88        method_handlers: Dict[str, grpc.RpcMethodHandler],
89    ) -> None:
90        # TODO(xuanwn): Implement this for AsyncIO.
91        pass
92
93    def add_insecure_port(self, address: str) -> int:
94        """Opens an insecure port for accepting RPCs.
95
96        This method may only be called before starting the server.
97
98        Args:
99          address: The address for which to open a port. If the port is 0,
100            or not specified in the address, then the gRPC runtime will choose a port.
101
102        Returns:
103          An integer port on which the server will accept RPC requests.
104        """
105        return _common.validate_port_binding_result(
106            address, self._server.add_insecure_port(_common.encode(address))
107        )
108
109    def add_secure_port(
110        self, address: str, server_credentials: grpc.ServerCredentials
111    ) -> int:
112        """Opens a secure port for accepting RPCs.
113
114        This method may only be called before starting the server.
115
116        Args:
117          address: The address for which to open a port.
118            if the port is 0, or not specified in the address, then the gRPC
119            runtime will choose a port.
120          server_credentials: A ServerCredentials object.
121
122        Returns:
123          An integer port on which the server will accept RPC requests.
124        """
125        return _common.validate_port_binding_result(
126            address,
127            self._server.add_secure_port(
128                _common.encode(address), server_credentials
129            ),
130        )
131
132    async def start(self) -> None:
133        """Starts this Server.
134
135        This method may only be called once. (i.e. it is not idempotent).
136        """
137        await self._server.start()
138
139    async def stop(self, grace: Optional[float]) -> None:
140        """Stops this Server.
141
142        This method immediately stops the server from servicing new RPCs in
143        all cases.
144
145        If a grace period is specified, this method waits until all active
146        RPCs are finished or until the grace period is reached. RPCs that haven't
147        been terminated within the grace period are aborted.
148        If a grace period is not specified (by passing None for grace), all
149        existing RPCs are aborted immediately and this method blocks until
150        the last RPC handler terminates.
151
152        This method is idempotent and may be called at any time. Passing a
153        smaller grace value in a subsequent call will have the effect of
154        stopping the Server sooner (passing None will have the effect of
155        stopping the server immediately). Passing a larger grace value in a
156        subsequent call will not have the effect of stopping the server later
157        (i.e. the most restrictive grace value is used).
158
159        Args:
160          grace: A duration of time in seconds or None.
161        """
162        await self._server.shutdown(grace)
163
164    async def wait_for_termination(
165        self, timeout: Optional[float] = None
166    ) -> bool:
167        """Block current coroutine until the server stops.
168
169        This is an EXPERIMENTAL API.
170
171        The wait will not consume computational resources during blocking, and
172        it will block until one of the two following conditions are met:
173
174        1) The server is stopped or terminated;
175        2) A timeout occurs if timeout is not `None`.
176
177        The timeout argument works in the same way as `threading.Event.wait()`.
178        https://docs.python.org/3/library/threading.html#threading.Event.wait
179
180        Args:
181          timeout: A floating point number specifying a timeout for the
182            operation in seconds.
183
184        Returns:
185          A bool indicates if the operation times out.
186        """
187        return await self._server.wait_for_termination(timeout)
188
189    def __del__(self):
190        """Schedules a graceful shutdown in current event loop.
191
192        The Cython AioServer doesn't hold a ref-count to this class. It should
193        be safe to slightly extend the underlying Cython object's life span.
194        """
195        if hasattr(self, "_server") and self._server.is_running():
196            cygrpc.schedule_coro_threadsafe(
197                self._server.shutdown(None),
198                self._loop,
199            )
200
201
202def server(
203    migration_thread_pool: Optional[Executor] = None,
204    handlers: Optional[Sequence[grpc.GenericRpcHandler]] = None,
205    interceptors: Optional[Sequence[Any]] = None,
206    options: Optional[ChannelArgumentType] = None,
207    maximum_concurrent_rpcs: Optional[int] = None,
208    compression: Optional[grpc.Compression] = None,
209):
210    """Creates a Server with which RPCs can be serviced.
211
212    Args:
213      migration_thread_pool: A futures.ThreadPoolExecutor to be used by the
214        Server to execute non-AsyncIO RPC handlers for migration purpose.
215      handlers: An optional list of GenericRpcHandlers used for executing RPCs.
216        More handlers may be added by calling add_generic_rpc_handlers any time
217        before the server is started.
218      interceptors: An optional list of ServerInterceptor objects that observe
219        and optionally manipulate the incoming RPCs before handing them over to
220        handlers. The interceptors are given control in the order they are
221        specified. This is an EXPERIMENTAL API.
222      options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC runtime)
223        to configure the channel.
224      maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server
225        will service before returning RESOURCE_EXHAUSTED status, or None to
226        indicate no limit.
227      compression: An element of grpc.Compression, e.g.
228        grpc.Compression.Gzip. This compression algorithm will be used for the
229        lifetime of the server unless overridden by set_compression.
230
231    Returns:
232      A Server object.
233    """
234    return Server(
235        migration_thread_pool,
236        () if handlers is None else handlers,
237        () if interceptors is None else interceptors,
238        () if options is None else options,
239        maximum_concurrent_rpcs,
240        compression,
241    )
242