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, 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(base_options: ChannelArgumentType,
30                               compression: Optional[grpc.Compression]):
31    compression_option = _compression.create_channel_option(compression)
32    return tuple(base_options) + compression_option
33
34
35class Server(_base_server.Server):
36    """Serves RPCs."""
37
38    def __init__(self, thread_pool: Optional[Executor],
39                 generic_handlers: Optional[Sequence[grpc.GenericRpcHandler]],
40                 interceptors: Optional[Sequence[Any]],
41                 options: ChannelArgumentType,
42                 maximum_concurrent_rpcs: Optional[int],
43                 compression: Optional[grpc.Compression]):
44        self._loop = cygrpc.get_working_loop()
45        if interceptors:
46            invalid_interceptors = [
47                interceptor for interceptor in interceptors
48                if not isinstance(interceptor, ServerInterceptor)
49            ]
50            if invalid_interceptors:
51                raise ValueError(
52                    'Interceptor must be ServerInterceptor, the '
53                    f'following are invalid: {invalid_interceptors}')
54        self._server = cygrpc.AioServer(
55            self._loop, thread_pool, generic_handlers, interceptors,
56            _augment_channel_arguments(options, compression),
57            maximum_concurrent_rpcs)
58
59    def add_generic_rpc_handlers(
60            self,
61            generic_rpc_handlers: Sequence[grpc.GenericRpcHandler]) -> None:
62        """Registers GenericRpcHandlers with this Server.
63
64        This method is only safe to call before the server is started.
65
66        Args:
67          generic_rpc_handlers: A sequence of GenericRpcHandlers that will be
68          used to service RPCs.
69        """
70        self._server.add_generic_rpc_handlers(generic_rpc_handlers)
71
72    def add_insecure_port(self, address: str) -> int:
73        """Opens an insecure port for accepting RPCs.
74
75        This method may only be called before starting the server.
76
77        Args:
78          address: The address for which to open a port. If the port is 0,
79            or not specified in the address, then the gRPC runtime will choose a port.
80
81        Returns:
82          An integer port on which the server will accept RPC requests.
83        """
84        return _common.validate_port_binding_result(
85            address, self._server.add_insecure_port(_common.encode(address)))
86
87    def add_secure_port(self, address: str,
88                        server_credentials: grpc.ServerCredentials) -> int:
89        """Opens a secure port for accepting RPCs.
90
91        This method may only be called before starting the server.
92
93        Args:
94          address: The address for which to open a port.
95            if the port is 0, or not specified in the address, then the gRPC
96            runtime will choose a port.
97          server_credentials: A ServerCredentials object.
98
99        Returns:
100          An integer port on which the server will accept RPC requests.
101        """
102        return _common.validate_port_binding_result(
103            address,
104            self._server.add_secure_port(_common.encode(address),
105                                         server_credentials))
106
107    async def start(self) -> None:
108        """Starts this Server.
109
110        This method may only be called once. (i.e. it is not idempotent).
111        """
112        await self._server.start()
113
114    async def stop(self, grace: Optional[float]) -> None:
115        """Stops this Server.
116
117        This method immediately stops the server from servicing new RPCs in
118        all cases.
119
120        If a grace period is specified, this method returns immediately and all
121        RPCs active at the end of the grace period are aborted. If a grace
122        period is not specified (by passing None for grace), all existing RPCs
123        are aborted immediately and this method blocks until the last RPC
124        handler terminates.
125
126        This method is idempotent and may be called at any time. Passing a
127        smaller grace value in a subsequent call will have the effect of
128        stopping the Server sooner (passing None will have the effect of
129        stopping the server immediately). Passing a larger grace value in a
130        subsequent call will not have the effect of stopping the server later
131        (i.e. the most restrictive grace value is used).
132
133        Args:
134          grace: A duration of time in seconds or None.
135        """
136        await self._server.shutdown(grace)
137
138    async def wait_for_termination(self,
139                                   timeout: Optional[float] = None) -> bool:
140        """Block current coroutine until the server stops.
141
142        This is an EXPERIMENTAL API.
143
144        The wait will not consume computational resources during blocking, and
145        it will block until one of the two following conditions are met:
146
147        1) The server is stopped or terminated;
148        2) A timeout occurs if timeout is not `None`.
149
150        The timeout argument works in the same way as `threading.Event.wait()`.
151        https://docs.python.org/3/library/threading.html#threading.Event.wait
152
153        Args:
154          timeout: A floating point number specifying a timeout for the
155            operation in seconds.
156
157        Returns:
158          A bool indicates if the operation times out.
159        """
160        return await self._server.wait_for_termination(timeout)
161
162    def __del__(self):
163        """Schedules a graceful shutdown in current event loop.
164
165        The Cython AioServer doesn't hold a ref-count to this class. It should
166        be safe to slightly extend the underlying Cython object's life span.
167        """
168        if hasattr(self, '_server'):
169            if self._server.is_running():
170                cygrpc.schedule_coro_threadsafe(
171                    self._server.shutdown(None),
172                    self._loop,
173                )
174
175
176def server(migration_thread_pool: Optional[Executor] = None,
177           handlers: Optional[Sequence[grpc.GenericRpcHandler]] = None,
178           interceptors: Optional[Sequence[Any]] = None,
179           options: Optional[ChannelArgumentType] = None,
180           maximum_concurrent_rpcs: Optional[int] = None,
181           compression: Optional[grpc.Compression] = None):
182    """Creates a Server with which RPCs can be serviced.
183
184    Args:
185      migration_thread_pool: A futures.ThreadPoolExecutor to be used by the
186        Server to execute non-AsyncIO RPC handlers for migration purpose.
187      handlers: An optional list of GenericRpcHandlers used for executing RPCs.
188        More handlers may be added by calling add_generic_rpc_handlers any time
189        before the server is started.
190      interceptors: An optional list of ServerInterceptor objects that observe
191        and optionally manipulate the incoming RPCs before handing them over to
192        handlers. The interceptors are given control in the order they are
193        specified. This is an EXPERIMENTAL API.
194      options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC runtime)
195        to configure the channel.
196      maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server
197        will service before returning RESOURCE_EXHAUSTED status, or None to
198        indicate no limit.
199      compression: An element of grpc.compression, e.g.
200        grpc.compression.Gzip. This compression algorithm will be used for the
201        lifetime of the server unless overridden by set_compression. This is an
202        EXPERIMENTAL option.
203
204    Returns:
205      A Server object.
206    """
207    return Server(migration_thread_pool, () if handlers is None else handlers,
208                  () if interceptors is None else interceptors,
209                  () if options is None else options, maximum_concurrent_rpcs,
210                  compression)
211