codekingpro/portable-devtools
114k
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"""Translates gRPC's server-side API into gRPC's server-side Beta API."""
15
16import collections
17import threading
18
19import grpc
20from grpc import _common
21from grpc.beta import _metadata
22from grpc.beta import interfaces
23from grpc.framework.common import cardinality
24from grpc.framework.common import style
25from grpc.framework.foundation import abandonment
26from grpc.framework.foundation import logging_pool
27from grpc.framework.foundation import stream
28from grpc.framework.interfaces.face import face
29
30# pylint: disable=too-many-return-statements
31
32_DEFAULT_POOL_SIZE = 8
33
34
35class _ServerProtocolContext(interfaces.GRPCServicerContext):
36 def __init__(self, servicer_context):
37 self._servicer_context = servicer_context
38
39 def peer(self):
40 return self._servicer_context.peer()
41
42 def disable_next_response_compression(self):
43 pass # TODO(https://github.com/grpc/grpc/issues/4078): design, implement.
44
45
46class _FaceServicerContext(face.ServicerContext):
47 def __init__(self, servicer_context):
48 self._servicer_context = servicer_context
49
50 def is_active(self):
51 return self._servicer_context.is_active()
52
53 def time_remaining(self):
54 return self._servicer_context.time_remaining()
55
56 def add_abortion_callback(self, abortion_callback):
57 error_msg = "add_abortion_callback no longer supported server-side!"
58 raise NotImplementedError(error_msg)
59
60 def cancel(self):
61 self._servicer_context.cancel()
62
63 def protocol_context(self):
64 return _ServerProtocolContext(self._servicer_context)
65
66 def invocation_metadata(self):
67 return _metadata.beta(self._servicer_context.invocation_metadata())
68
69 def initial_metadata(self, initial_metadata):
70 self._servicer_context.send_initial_metadata(
71 _metadata.unbeta(initial_metadata)
72 )
73
74 def terminal_metadata(self, terminal_metadata):
75 self._servicer_context.set_terminal_metadata(
76 _metadata.unbeta(terminal_metadata)
77 )
78
79 def code(self, code):
80 self._servicer_context.set_code(code)
81
82 def details(self, details):
83 self._servicer_context.set_details(details)
84
85
86def _adapt_unary_request_inline(unary_request_inline):
87 def adaptation(request, servicer_context):
88 return unary_request_inline(
89 request, _FaceServicerContext(servicer_context)
90 )
91
92 return adaptation
93
94
95def _adapt_stream_request_inline(stream_request_inline):
96 def adaptation(request_iterator, servicer_context):
97 return stream_request_inline(
98 request_iterator, _FaceServicerContext(servicer_context)
99 )
100
101 return adaptation
102
103
104class _Callback(stream.Consumer):
105 def __init__(self):
106 self._condition = threading.Condition()
107 self._values = []
108 self._terminated = False
109 self._cancelled = False
110
111 def consume(self, value):
112 with self._condition:
113 self._values.append(value)
114 self._condition.notify_all()
115
116 def terminate(self):
117 with self._condition:
118 self._terminated = True
119 self._condition.notify_all()
120
121 def consume_and_terminate(self, value):
122 with self._condition:
123 self._values.append(value)
124 self._terminated = True
125 self._condition.notify_all()
126
127 def cancel(self):
128 with self._condition:
129 self._cancelled = True
130 self._condition.notify_all()
131
132 def draw_one_value(self):
133 with self._condition:
134 while True:
135 if self._cancelled:
136 raise abandonment.Abandoned()
137 if self._values:
138 return self._values.pop(0)
139 if self._terminated:
140 return None
141 self._condition.wait()
142
143 def draw_all_values(self):
144 with self._condition:
145 while True:
146 if self._cancelled:
147 raise abandonment.Abandoned()
148 if self._terminated:
149 all_values = tuple(self._values)
150 self._values = None
151 return all_values
152 self._condition.wait()
153
154
155def _run_request_pipe_thread(
156 request_iterator, request_consumer, servicer_context
157):
158 thread_joined = threading.Event()
159
160 def pipe_requests():
161 for request in request_iterator:
162 if not servicer_context.is_active() or thread_joined.is_set():
163 return
164 request_consumer.consume(request)
165 if not servicer_context.is_active() or thread_joined.is_set():
166 return
167 request_consumer.terminate()
168
169 request_pipe_thread = threading.Thread(target=pipe_requests)
170 request_pipe_thread.daemon = True
171 request_pipe_thread.start()
172
173
174def _adapt_unary_unary_event(unary_unary_event):
175 def adaptation(request, servicer_context):
176 callback = _Callback()
177 if not servicer_context.add_callback(callback.cancel):
178 raise abandonment.Abandoned()
179 unary_unary_event(
180 request,
181 callback.consume_and_terminate,
182 _FaceServicerContext(servicer_context),
183 )
184 return callback.draw_all_values()[0]
185
186 return adaptation
187
188
189def _adapt_unary_stream_event(unary_stream_event):
190 def adaptation(request, servicer_context):
191 callback = _Callback()
192 if not servicer_context.add_callback(callback.cancel):
193 raise abandonment.Abandoned()
194 unary_stream_event(
195 request, callback, _FaceServicerContext(servicer_context)
196 )
197 while True:
198 response = callback.draw_one_value()
199 if response is None:
200 return
201 else:
202 yield response
203
204 return adaptation
205
206
207def _adapt_stream_unary_event(stream_unary_event):
208 def adaptation(request_iterator, servicer_context):
209 callback = _Callback()
210 if not servicer_context.add_callback(callback.cancel):
211 raise abandonment.Abandoned()
212 request_consumer = stream_unary_event(
213 callback.consume_and_terminate,
214 _FaceServicerContext(servicer_context),
215 )
216 _run_request_pipe_thread(
217 request_iterator, request_consumer, servicer_context
218 )
219 return callback.draw_all_values()[0]
220
221 return adaptation
222
223
224def _adapt_stream_stream_event(stream_stream_event):
225 def adaptation(request_iterator, servicer_context):
226 callback = _Callback()
227 if not servicer_context.add_callback(callback.cancel):
228 raise abandonment.Abandoned()
229 request_consumer = stream_stream_event(
230 callback, _FaceServicerContext(servicer_context)
231 )
232 _run_request_pipe_thread(
233 request_iterator, request_consumer, servicer_context
234 )
235 while True:
236 response = callback.draw_one_value()
237 if response is None:
238 return
239 else:
240 yield response
241
242 return adaptation
243
244
245class _SimpleMethodHandler(
246 collections.namedtuple(
247 "_MethodHandler",
248 (
249 "request_streaming",
250 "response_streaming",
251 "request_deserializer",
252 "response_serializer",
253 "unary_unary",
254 "unary_stream",
255 "stream_unary",
256 "stream_stream",
257 ),
258 ),
259 grpc.RpcMethodHandler,
260):
261 pass
262
263
264def _simple_method_handler( # noqa: PLR0911
265 implementation, request_deserializer, response_serializer
266):
267 if implementation.style is style.Service.INLINE:
268 if implementation.cardinality is cardinality.Cardinality.UNARY_UNARY:
269 return _SimpleMethodHandler(
270 False,
271 False,
272 request_deserializer,
273 response_serializer,
274 _adapt_unary_request_inline(implementation.unary_unary_inline),
275 None,
276 None,
277 None,
278 )
279 if implementation.cardinality is cardinality.Cardinality.UNARY_STREAM:
280 return _SimpleMethodHandler(
281 False,
282 True,
283 request_deserializer,
284 response_serializer,
285 None,
286 _adapt_unary_request_inline(implementation.unary_stream_inline),
287 None,
288 None,
289 )
290 if implementation.cardinality is cardinality.Cardinality.STREAM_UNARY:
291 return _SimpleMethodHandler(
292 True,
293 False,
294 request_deserializer,
295 response_serializer,
296 None,
297 None,
298 _adapt_stream_request_inline(
299 implementation.stream_unary_inline
300 ),
301 None,
302 )
303 if implementation.cardinality is cardinality.Cardinality.STREAM_STREAM:
304 return _SimpleMethodHandler(
305 True,
306 True,
307 request_deserializer,
308 response_serializer,
309 None,
310 None,
311 None,
312 _adapt_stream_request_inline(
313 implementation.stream_stream_inline
314 ),
315 )
316 elif implementation.style is style.Service.EVENT:
317 if implementation.cardinality is cardinality.Cardinality.UNARY_UNARY:
318 return _SimpleMethodHandler(
319 False,
320 False,
321 request_deserializer,
322 response_serializer,
323 _adapt_unary_unary_event(implementation.unary_unary_event),
324 None,
325 None,
326 None,
327 )
328 if implementation.cardinality is cardinality.Cardinality.UNARY_STREAM:
329 return _SimpleMethodHandler(
330 False,
331 True,
332 request_deserializer,
333 response_serializer,
334 None,
335 _adapt_unary_stream_event(implementation.unary_stream_event),
336 None,
337 None,
338 )
339 if implementation.cardinality is cardinality.Cardinality.STREAM_UNARY:
340 return _SimpleMethodHandler(
341 True,
342 False,
343 request_deserializer,
344 response_serializer,
345 None,
346 None,
347 _adapt_stream_unary_event(implementation.stream_unary_event),
348 None,
349 )
350 if implementation.cardinality is cardinality.Cardinality.STREAM_STREAM:
351 return _SimpleMethodHandler(
352 True,
353 True,
354 request_deserializer,
355 response_serializer,
356 None,
357 None,
358 None,
359 _adapt_stream_stream_event(implementation.stream_stream_event),
360 )
361 raise ValueError()
362
363
364def _flatten_method_pair_map(method_pair_map):
365 method_pair_map = method_pair_map or {}
366 flat_map = {}
367 for method_pair in method_pair_map:
368 method = _common.fully_qualified_method(method_pair[0], method_pair[1])
369 flat_map[method] = method_pair_map[method_pair]
370 return flat_map
371
372
373class _GenericRpcHandler(grpc.GenericRpcHandler):
374 def __init__(
375 self,
376 method_implementations,
377 multi_method_implementation,
378 request_deserializers,
379 response_serializers,
380 ):
381 self._method_implementations = _flatten_method_pair_map(
382 method_implementations
383 )
384 self._request_deserializers = _flatten_method_pair_map(
385 request_deserializers
386 )
387 self._response_serializers = _flatten_method_pair_map(
388 response_serializers
389 )
390 self._multi_method_implementation = multi_method_implementation
391
392 def service(self, handler_call_details):
393 method_implementation = self._method_implementations.get(
394 handler_call_details.method
395 )
396 if method_implementation is not None:
397 return _simple_method_handler(
398 method_implementation,
399 self._request_deserializers.get(handler_call_details.method),
400 self._response_serializers.get(handler_call_details.method),
401 )
402 if self._multi_method_implementation is None:
403 return None
404 try:
405 return None # TODO(nathaniel): call the multimethod.
406 except face.NoSuchMethodError:
407 return None
408
409
410class _Server(interfaces.Server):
411 def __init__(self, grpc_server):
412 self._grpc_server = grpc_server
413
414 def add_insecure_port(self, address):
415 return self._grpc_server.add_insecure_port(address)
416
417 def add_secure_port(self, address, server_credentials):
418 return self._grpc_server.add_secure_port(address, server_credentials)
419
420 def start(self):
421 self._grpc_server.start()
422
423 def stop(self, grace):
424 return self._grpc_server.stop(grace)
425
426 def __enter__(self):
427 self._grpc_server.start()
428 return self
429
430 def __exit__(self, exc_type, exc_val, exc_tb):
431 self._grpc_server.stop(None)
432 return False
433
434
435def server(
436 service_implementations,
437 multi_method_implementation,
438 request_deserializers,
439 response_serializers,
440 thread_pool,
441 thread_pool_size,
442):
443 generic_rpc_handler = _GenericRpcHandler(
444 service_implementations,
445 multi_method_implementation,
446 request_deserializers,
447 response_serializers,
448 )
449 if thread_pool is None:
450 effective_thread_pool = logging_pool.pool(
451 _DEFAULT_POOL_SIZE if thread_pool_size is None else thread_pool_size
452 )
453 else:
454 effective_thread_pool = thread_pool
455 return _Server(
456 grpc.server(effective_thread_pool, handlers=(generic_rpc_handler,))
457 )
458