Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/grpc/_common.py: 45%

1# Copyright 2016 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"""Shared implementation.""" 

15 

16import logging 

17import time 

18from typing import Any, AnyStr, Callable, Optional, Union 

19 

20import grpc 

21from grpc._cython import cygrpc 

22from grpc._typing import DeserializingFunction 

23from grpc._typing import SerializingFunction 

24 

25_LOGGER = logging.getLogger(__name__) 

26 

27CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY = { 

28 cygrpc.ConnectivityState.idle: 

29 grpc.ChannelConnectivity.IDLE, 

30 cygrpc.ConnectivityState.connecting: 

31 grpc.ChannelConnectivity.CONNECTING, 

32 cygrpc.ConnectivityState.ready: 

33 grpc.ChannelConnectivity.READY, 

34 cygrpc.ConnectivityState.transient_failure: 

35 grpc.ChannelConnectivity.TRANSIENT_FAILURE, 

36 cygrpc.ConnectivityState.shutdown: 

37 grpc.ChannelConnectivity.SHUTDOWN, 

38} 

39 

40CYGRPC_STATUS_CODE_TO_STATUS_CODE = { 

41 cygrpc.StatusCode.ok: grpc.StatusCode.OK, 

42 cygrpc.StatusCode.cancelled: grpc.StatusCode.CANCELLED, 

43 cygrpc.StatusCode.unknown: grpc.StatusCode.UNKNOWN, 

44 cygrpc.StatusCode.invalid_argument: grpc.StatusCode.INVALID_ARGUMENT, 

45 cygrpc.StatusCode.deadline_exceeded: grpc.StatusCode.DEADLINE_EXCEEDED, 

46 cygrpc.StatusCode.not_found: grpc.StatusCode.NOT_FOUND, 

47 cygrpc.StatusCode.already_exists: grpc.StatusCode.ALREADY_EXISTS, 

48 cygrpc.StatusCode.permission_denied: grpc.StatusCode.PERMISSION_DENIED, 

49 cygrpc.StatusCode.unauthenticated: grpc.StatusCode.UNAUTHENTICATED, 

50 cygrpc.StatusCode.resource_exhausted: grpc.StatusCode.RESOURCE_EXHAUSTED, 

51 cygrpc.StatusCode.failed_precondition: grpc.StatusCode.FAILED_PRECONDITION, 

52 cygrpc.StatusCode.aborted: grpc.StatusCode.ABORTED, 

53 cygrpc.StatusCode.out_of_range: grpc.StatusCode.OUT_OF_RANGE, 

54 cygrpc.StatusCode.unimplemented: grpc.StatusCode.UNIMPLEMENTED, 

55 cygrpc.StatusCode.internal: grpc.StatusCode.INTERNAL, 

56 cygrpc.StatusCode.unavailable: grpc.StatusCode.UNAVAILABLE, 

57 cygrpc.StatusCode.data_loss: grpc.StatusCode.DATA_LOSS, 

58} 

59STATUS_CODE_TO_CYGRPC_STATUS_CODE = { 

60 grpc_code: cygrpc_code 

61 for cygrpc_code, grpc_code in CYGRPC_STATUS_CODE_TO_STATUS_CODE.items() 

62} 

63 

64MAXIMUM_WAIT_TIMEOUT = 0.1 

65 

66_ERROR_MESSAGE_PORT_BINDING_FAILED = 'Failed to bind to address %s; set ' \ 

67 'GRPC_VERBOSITY=debug environment variable to see detailed error message.' 

68 

69 

70def encode(s: AnyStr) -> bytes: 

71 if isinstance(s, bytes): 

72 return s 

73 else: 

74 return s.encode('utf8') 

75 

76 

77def decode(b: AnyStr) -> str: 

78 if isinstance(b, bytes): 

79 return b.decode('utf-8', 'replace') 

80 return b 

81 

82 

83def _transform(message: Any, transformer: Union[SerializingFunction, 

84 DeserializingFunction, None], 

85 exception_message: str) -> Any: 

86 if transformer is None: 

87 return message 

88 else: 

89 try: 

90 return transformer(message) 

91 except Exception: # pylint: disable=broad-except 

92 _LOGGER.exception(exception_message) 

93 return None 

94 

95 

96def serialize(message: Any, serializer: Optional[SerializingFunction]) -> bytes: 

97 return _transform(message, serializer, 'Exception serializing message!') 

98 

99 

100def deserialize(serialized_message: bytes, 

101 deserializer: Optional[DeserializingFunction]) -> Any: 

102 return _transform(serialized_message, deserializer, 

103 'Exception deserializing message!') 

104 

105 

106def fully_qualified_method(group: str, method: str) -> str: 

107 return '/{}/{}'.format(group, method) 

108 

109 

110def _wait_once(wait_fn: Callable[..., bool], timeout: float, 

111 spin_cb: Optional[Callable[[], None]]): 

112 wait_fn(timeout=timeout) 

113 if spin_cb is not None: 

114 spin_cb() 

115 

116 

117def wait(wait_fn: Callable[..., bool], 

118 wait_complete_fn: Callable[[], bool], 

119 timeout: Optional[float] = None, 

120 spin_cb: Optional[Callable[[], None]] = None) -> bool: 

121 """Blocks waiting for an event without blocking the thread indefinitely. 

122 

123 See https://github.com/grpc/grpc/issues/19464 for full context. CPython's 

124 `threading.Event.wait` and `threading.Condition.wait` methods, if invoked 

125 without a timeout kwarg, may block the calling thread indefinitely. If the 

126 call is made from the main thread, this means that signal handlers may not 

127 run for an arbitrarily long period of time. 

128 

129 This wrapper calls the supplied wait function with an arbitrary short 

130 timeout to ensure that no signal handler has to wait longer than 

131 MAXIMUM_WAIT_TIMEOUT before executing. 

132 

133 Args: 

134 wait_fn: A callable acceptable a single float-valued kwarg named 

135 `timeout`. This function is expected to be one of `threading.Event.wait` 

136 or `threading.Condition.wait`. 

137 wait_complete_fn: A callable taking no arguments and returning a bool. 

138 When this function returns true, it indicates that waiting should cease. 

139 timeout: An optional float-valued number of seconds after which the wait 

140 should cease. 

141 spin_cb: An optional Callable taking no arguments and returning nothing. 

142 This callback will be called on each iteration of the spin. This may be 

143 used for, e.g. work related to forking. 

144 

145 Returns: 

146 True if a timeout was supplied and it was reached. False otherwise. 

147 """ 

148 if timeout is None: 

149 while not wait_complete_fn(): 

150 _wait_once(wait_fn, MAXIMUM_WAIT_TIMEOUT, spin_cb) 

151 else: 

152 end = time.time() + timeout 

153 while not wait_complete_fn(): 

154 remaining = min(end - time.time(), MAXIMUM_WAIT_TIMEOUT) 

155 if remaining < 0: 

156 return True 

157 _wait_once(wait_fn, remaining, spin_cb) 

158 return False 

159 

160 

161def validate_port_binding_result(address: str, port: int) -> int: 

162 """Validates if the port binding succeed. 

163 

164 If the port returned by Core is 0, the binding is failed. However, in that 

165 case, the Core API doesn't return a detailed failing reason. The best we 

166 can do is raising an exception to prevent further confusion. 

167 

168 Args: 

169 address: The address string to be bound. 

170 port: An int returned by core 

171 """ 

172 if port == 0: 

173 # The Core API doesn't return a failure message. The best we can do 

174 # is raising an exception to prevent further confusion. 

175 raise RuntimeError(_ERROR_MESSAGE_PORT_BINDING_FAILED % address) 

176 else: 

177 return port