codekingpro/portable-devtools

1# Copyright 2015 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"""Internal utilities for gRPC Python."""
15
16import collections
17import logging
18import threading
19import time
20from typing import Callable, Dict, Optional, Sequence
21
22import grpc
23from grpc import _common
24from grpc._typing import DoneCallbackType
25
26_LOGGER = logging.getLogger(__name__)
27
28_DONE_CALLBACK_EXCEPTION_LOG_MESSAGE = (
29    'Exception calling connectivity future "done" callback!'
30)
31
32
33class RpcMethodHandler(
34    collections.namedtuple(
35        "_RpcMethodHandler",
36        (
37            "request_streaming",
38            "response_streaming",
39            "request_deserializer",
40            "response_serializer",
41            "unary_unary",
42            "unary_stream",
43            "stream_unary",
44            "stream_stream",
45        ),
46    ),
47    grpc.RpcMethodHandler,
48):
49    pass
50
51
52class DictionaryGenericHandler(grpc.ServiceRpcHandler):
53    _name: str
54    _method_handlers: Dict[str, grpc.RpcMethodHandler]
55
56    def __init__(
57        self, service: str, method_handlers: Dict[str, grpc.RpcMethodHandler]
58    ):
59        self._name = service
60        self._method_handlers = {
61            _common.fully_qualified_method(service, method): method_handler
62            for method, method_handler in method_handlers.items()
63        }
64
65    def service_name(self) -> str:
66        return self._name
67
68    def service(
69        self, handler_call_details: grpc.HandlerCallDetails
70    ) -> Optional[grpc.RpcMethodHandler]:
71        details_method = handler_call_details.method
72        return self._method_handlers.get(
73            details_method
74        )  # pytype: disable=attribute-error
75
76
77class _ChannelReadyFuture(grpc.Future):
78    _condition: threading.Condition
79    _channel: grpc.Channel
80    _matured: bool
81    _cancelled: bool
82    _done_callbacks: Sequence[Callable]
83
84    def __init__(self, channel: grpc.Channel):
85        self._condition = threading.Condition()
86        self._channel = channel
87
88        self._matured = False
89        self._cancelled = False
90        self._done_callbacks = []
91
92    def _block(self, timeout: Optional[float]) -> None:
93        until = None if timeout is None else time.time() + timeout
94        with self._condition:
95            while True:
96                if self._cancelled:
97                    raise grpc.FutureCancelledError()
98                if self._matured:
99                    return
100                if until is None:
101                    self._condition.wait()
102                else:
103                    remaining = until - time.time()
104                    if remaining < 0:
105                        raise grpc.FutureTimeoutError()
106                    self._condition.wait(timeout=remaining)
107
108    def _update(self, connectivity: Optional[grpc.ChannelConnectivity]) -> None:
109        with self._condition:
110            if (
111                not self._cancelled
112                and connectivity is grpc.ChannelConnectivity.READY
113            ):
114                self._matured = True
115                self._channel.unsubscribe(self._update)
116                self._condition.notify_all()
117                done_callbacks = tuple(self._done_callbacks)
118                self._done_callbacks = None
119            else:
120                return
121
122        for done_callback in done_callbacks:
123            try:
124                done_callback(self)
125            except Exception:  # pylint: disable=broad-except
126                _LOGGER.exception(_DONE_CALLBACK_EXCEPTION_LOG_MESSAGE)
127
128    def cancel(self) -> bool:
129        with self._condition:
130            if not self._matured:
131                self._cancelled = True
132                self._channel.unsubscribe(self._update)
133                self._condition.notify_all()
134                done_callbacks = tuple(self._done_callbacks)
135                self._done_callbacks = None
136            else:
137                return False
138
139        for done_callback in done_callbacks:
140            try:
141                done_callback(self)
142            except Exception:  # pylint: disable=broad-except
143                _LOGGER.exception(_DONE_CALLBACK_EXCEPTION_LOG_MESSAGE)
144
145        return True
146
147    def cancelled(self) -> bool:
148        with self._condition:
149            return self._cancelled
150
151    def running(self) -> bool:
152        with self._condition:
153            return not self._cancelled and not self._matured
154
155    def done(self) -> bool:
156        with self._condition:
157            return self._cancelled or self._matured
158
159    def result(self, timeout: Optional[float] = None) -> None:
160        self._block(timeout)
161
162    def exception(self, timeout: Optional[float] = None) -> None:
163        self._block(timeout)
164
165    def traceback(self, timeout: Optional[float] = None) -> None:
166        self._block(timeout)
167
168    def add_done_callback(self, fn: DoneCallbackType):
169        with self._condition:
170            if not self._cancelled and not self._matured:
171                self._done_callbacks.append(fn)
172                return
173
174        fn(self)
175
176    def start(self):
177        with self._condition:
178            self._channel.subscribe(self._update, try_to_connect=True)
179
180    def __del__(self):
181        with self._condition:
182            if not self._cancelled and not self._matured:
183                self._channel.unsubscribe(self._update)
184
185
186def channel_ready_future(channel: grpc.Channel) -> _ChannelReadyFuture:
187    ready_future = _ChannelReadyFuture(channel)
188    ready_future.start()
189    return ready_future
190
191
192def first_version_is_lower(version1: str, version2: str) -> bool:
193    """
194    Compares two versions in the format '1.60.1' or '1.60.1.dev0'.
195
196    This method will be used in all stubs generated by grpcio-tools to check whether
197    the stub version is compatible with the runtime grpcio.
198
199    Args:
200        version1: The first version string.
201        version2: The second version string.
202
203    Returns:
204        True if version1 is lower, False otherwise.
205    """
206    version1_list = version1.split(".")
207    version2_list = version2.split(".")
208
209    try:
210        for i in range(3):
211            if int(version1_list[i]) < int(version2_list[i]):
212                return True
213            if int(version1_list[i]) > int(version2_list[i]):
214                return False
215    except ValueError:
216        # Return false in case we can't convert version to int.
217        return False
218
219    # The version without dev0 will be considered lower.
220    return len(version1_list) < len(version2_list)
221