src/python/grpcio/grpc/framework/crust/_service.py

# 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.

"""Behaviors for servicing RPCs."""

from grpc.framework.crust import _control
from grpc.framework.foundation import abandonment
from grpc.framework.interfaces.base import utilities
from grpc.framework.interfaces.face import face


class _ServicerContext(face.ServicerContext):

  def __init__(self, rendezvous):
    self._rendezvous = rendezvous

  def is_active(self):
    return self._rendezvous.is_active()

  def time_remaining(self):
    return self._rendezvous.time_remaining()

  def add_abortion_callback(self, abortion_callback):
    return self._rendezvous.add_abortion_callback(abortion_callback)

  def cancel(self):
    self._rendezvous.cancel()

  def invocation_metadata(self):
    return self._rendezvous.initial_metadata()

  def initial_metadata(self, initial_metadata):
    self._rendezvous.set_initial_metadata(initial_metadata)

  def terminal_metadata(self, terminal_metadata):
    self._rendezvous.set_terminal_metadata(terminal_metadata)

  def code(self, code):
    self._rendezvous.set_code(code)

  def details(self, details):
    self._rendezvous.set_details(details)


def _adaptation(pool, in_pool):
  def adaptation(operator, operation_context):
    rendezvous = _control.Rendezvous(operator, operation_context)
    outcome = operation_context.add_termination_callback(rendezvous.set_outcome)
    if outcome is None:
      pool.submit(_control.pool_wrap(in_pool, operation_context), rendezvous)
      return utilities.full_subscription(rendezvous)
    else:
      raise abandonment.Abandoned()
  return adaptation


def adapt_inline_unary_unary(method, pool):
  def in_pool(rendezvous):
    request = next(rendezvous)
    response = method(request, _ServicerContext(rendezvous))
    rendezvous.consume_and_terminate(response)
  return _adaptation(pool, in_pool)


def adapt_inline_unary_stream(method, pool):
  def in_pool(rendezvous):
    request = next(rendezvous)
    response_iterator = method(request, _ServicerContext(rendezvous))
    for response in response_iterator:
      rendezvous.consume(response)
    rendezvous.terminate()
  return _adaptation(pool, in_pool)


def adapt_inline_stream_unary(method, pool):
  def in_pool(rendezvous):
    response = method(rendezvous, _ServicerContext(rendezvous))
    rendezvous.consume_and_terminate(response)
  return _adaptation(pool, in_pool)


def adapt_inline_stream_stream(method, pool):
  def in_pool(rendezvous):
    response_iterator = method(rendezvous, _ServicerContext(rendezvous))
    for response in response_iterator:
      rendezvous.consume(response)
    rendezvous.terminate()
  return _adaptation(pool, in_pool)


def adapt_event_unary_unary(method, pool):
  def in_pool(rendezvous):
    request = next(rendezvous)
    method(
        request, rendezvous.consume_and_terminate, _ServicerContext(rendezvous))
  return _adaptation(pool, in_pool)


def adapt_event_unary_stream(method, pool):
  def in_pool(rendezvous):
    request = next(rendezvous)
    method(request, rendezvous, _ServicerContext(rendezvous))
  return _adaptation(pool, in_pool)


def adapt_event_stream_unary(method, pool):
  def in_pool(rendezvous):
    request_consumer = method(
        rendezvous.consume_and_terminate, _ServicerContext(rendezvous))
    for request in rendezvous:
      request_consumer.consume(request)
    request_consumer.terminate()
  return _adaptation(pool, in_pool)


def adapt_event_stream_stream(method, pool):
  def in_pool(rendezvous):
    request_consumer = method(rendezvous, _ServicerContext(rendezvous))
    for request in rendezvous:
      request_consumer.consume(request)
    request_consumer.terminate()
  return _adaptation(pool, in_pool)


def adapt_multi_method(multi_method, pool):
  def adaptation(group, method, operator, operation_context):
    rendezvous = _control.Rendezvous(operator, operation_context)
    outcome = operation_context.add_termination_callback(rendezvous.set_outcome)
    if outcome is None:
      def in_pool():
        request_consumer = multi_method.service(
            group, method, rendezvous, _ServicerContext(rendezvous))
        for request in rendezvous:
          request_consumer.consume(request)
        request_consumer.terminate()
      pool.submit(_control.pool_wrap(in_pool, operation_context), rendezvous)
      return utilities.full_subscription(rendezvous)
    else:
      raise abandonment.Abandoned()
  return adaptation