1 files changed, 166 insertions, 0 deletions
diff --git a/src/python/grpcio/grpc/framework/crust/_service.py b/src/python/grpcio/grpc/framework/crust/_service.py
new file mode 100644
index 0000000000..2455a58f59
--- /dev/null
+++ b/src/python/grpcio/grpc/framework/crust/_service.py
@@ -0,0 +1,166 @@
+# 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(
+            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