| # Copyright 2015, Google Inc. |
| # All rights reserved. |
| # |
| # Redistribution and use in source and binary forms, with or without |
| # modification, are permitted provided that the following conditions are |
| # met: |
| # |
| # * Redistributions of source code must retain the above copyright |
| # notice, this list of conditions and the following disclaimer. |
| # * Redistributions in binary form must reproduce the above |
| # copyright notice, this list of conditions and the following disclaimer |
| # in the documentation and/or other materials provided with the |
| # distribution. |
| # * Neither the name of Google Inc. nor the names of its |
| # contributors may be used to endorse or promote products derived from |
| # this software without specific prior written permission. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
| # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
| # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
| # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
| # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
| # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
| # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
| # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
| # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
| # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
| # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| |
| """Utility functions for invoking RPCs.""" |
| |
| from grpc.framework.crust import _control |
| from grpc.framework.interfaces.base import utilities |
| from grpc.framework.interfaces.face import face |
| |
| _ITERATOR_EXCEPTION_LOG_MESSAGE = 'Exception iterating over requests!' |
| |
| _EMPTY_COMPLETION = utilities.completion(None, None, None) |
| |
| |
| def _invoke(end, group, method, timeout, initial_metadata, payload, complete): |
| rendezvous = _control.Rendezvous(None, None) |
| operation_context, operator = end.operate( |
| group, method, utilities.full_subscription(rendezvous), timeout, |
| initial_metadata=initial_metadata, payload=payload, |
| completion=_EMPTY_COMPLETION if complete else None) |
| rendezvous.set_operator_and_context(operator, operation_context) |
| outcome = operation_context.add_termination_callback(rendezvous.set_outcome) |
| if outcome is not None: |
| rendezvous.set_outcome(outcome) |
| return rendezvous, operation_context, outcome |
| |
| |
| def _event_return_unary( |
| receiver, abortion_callback, rendezvous, operation_context, outcome, pool): |
| if outcome is None: |
| def in_pool(): |
| abortion = rendezvous.add_abortion_callback(abortion_callback) |
| if abortion is None: |
| try: |
| receiver.initial_metadata(rendezvous.initial_metadata()) |
| receiver.response(next(rendezvous)) |
| receiver.complete( |
| rendezvous.terminal_metadata(), rendezvous.code(), |
| rendezvous.details()) |
| except face.AbortionError: |
| pass |
| else: |
| abortion_callback(abortion) |
| pool.submit(_control.pool_wrap(in_pool, operation_context)) |
| return rendezvous |
| |
| |
| def _event_return_stream( |
| receiver, abortion_callback, rendezvous, operation_context, outcome, pool): |
| if outcome is None: |
| def in_pool(): |
| abortion = rendezvous.add_abortion_callback(abortion_callback) |
| if abortion is None: |
| try: |
| receiver.initial_metadata(rendezvous.initial_metadata()) |
| for response in rendezvous: |
| receiver.response(response) |
| receiver.complete( |
| rendezvous.terminal_metadata(), rendezvous.code(), |
| rendezvous.details()) |
| except face.AbortionError: |
| pass |
| else: |
| abortion_callback(abortion) |
| pool.submit(_control.pool_wrap(in_pool, operation_context)) |
| return rendezvous |
| |
| |
| def blocking_unary_unary( |
| end, group, method, timeout, with_call, initial_metadata, payload): |
| """Services in a blocking fashion a unary-unary servicer method.""" |
| rendezvous, unused_operation_context, unused_outcome = _invoke( |
| end, group, method, timeout, initial_metadata, payload, True) |
| if with_call: |
| return next(rendezvous), rendezvous |
| else: |
| return next(rendezvous) |
| |
| |
| def future_unary_unary(end, group, method, timeout, initial_metadata, payload): |
| """Services a value-in value-out servicer method by returning a Future.""" |
| rendezvous, unused_operation_context, unused_outcome = _invoke( |
| end, group, method, timeout, initial_metadata, payload, True) |
| return rendezvous |
| |
| |
| def inline_unary_stream(end, group, method, timeout, initial_metadata, payload): |
| """Services a value-in stream-out servicer method.""" |
| rendezvous, unused_operation_context, unused_outcome = _invoke( |
| end, group, method, timeout, initial_metadata, payload, True) |
| return rendezvous |
| |
| |
| def blocking_stream_unary( |
| end, group, method, timeout, with_call, initial_metadata, payload_iterator, |
| pool): |
| """Services in a blocking fashion a stream-in value-out servicer method.""" |
| rendezvous, operation_context, outcome = _invoke( |
| end, group, method, timeout, initial_metadata, None, False) |
| if outcome is None: |
| def in_pool(): |
| for payload in payload_iterator: |
| rendezvous.consume(payload) |
| rendezvous.terminate() |
| pool.submit(_control.pool_wrap(in_pool, operation_context)) |
| if with_call: |
| return next(rendezvous), rendezvous |
| else: |
| return next(rendezvous) |
| else: |
| if with_call: |
| return next(rendezvous), rendezvous |
| else: |
| return next(rendezvous) |
| |
| |
| def future_stream_unary( |
| end, group, method, timeout, initial_metadata, payload_iterator, pool): |
| """Services a stream-in value-out servicer method by returning a Future.""" |
| rendezvous, operation_context, outcome = _invoke( |
| end, group, method, timeout, initial_metadata, None, False) |
| if outcome is None: |
| def in_pool(): |
| for payload in payload_iterator: |
| rendezvous.consume(payload) |
| rendezvous.terminate() |
| pool.submit(_control.pool_wrap(in_pool, operation_context)) |
| return rendezvous |
| |
| |
| def inline_stream_stream( |
| end, group, method, timeout, initial_metadata, payload_iterator, pool): |
| """Services a stream-in stream-out servicer method.""" |
| rendezvous, operation_context, outcome = _invoke( |
| end, group, method, timeout, initial_metadata, None, False) |
| if outcome is None: |
| def in_pool(): |
| for payload in payload_iterator: |
| rendezvous.consume(payload) |
| rendezvous.terminate() |
| pool.submit(_control.pool_wrap(in_pool, operation_context)) |
| return rendezvous |
| |
| |
| def event_unary_unary( |
| end, group, method, timeout, initial_metadata, payload, receiver, |
| abortion_callback, pool): |
| rendezvous, operation_context, outcome = _invoke( |
| end, group, method, timeout, initial_metadata, payload, True) |
| return _event_return_unary( |
| receiver, abortion_callback, rendezvous, operation_context, outcome, pool) |
| |
| |
| def event_unary_stream( |
| end, group, method, timeout, initial_metadata, payload, |
| receiver, abortion_callback, pool): |
| rendezvous, operation_context, outcome = _invoke( |
| end, group, method, timeout, initial_metadata, payload, True) |
| return _event_return_stream( |
| receiver, abortion_callback, rendezvous, operation_context, outcome, pool) |
| |
| |
| def event_stream_unary( |
| end, group, method, timeout, initial_metadata, receiver, abortion_callback, |
| pool): |
| rendezvous, operation_context, outcome = _invoke( |
| end, group, method, timeout, initial_metadata, None, False) |
| return _event_return_unary( |
| receiver, abortion_callback, rendezvous, operation_context, outcome, pool) |
| |
| |
| def event_stream_stream( |
| end, group, method, timeout, initial_metadata, receiver, abortion_callback, |
| pool): |
| rendezvous, operation_context, outcome = _invoke( |
| end, group, method, timeout, initial_metadata, None, False) |
| return _event_return_stream( |
| receiver, abortion_callback, rendezvous, operation_context, outcome, pool) |