Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #979 from nathanielmanistaatgoogle/framework-cleanup

Browse Source 
Refactoring within Python RPC Framework
pull/992/head
Masood Malekghassemi 10 years ago
parent 1722032534 5c87a30c41
commit 874e53ad2d
20 changed files with 664 additions and 1054 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 30
      src/python/src/grpc/_adapter/_face_test_case.py
  2. 41
      src/python/src/grpc/early_adopter/_face_utilities.py
  3. 47
      src/python/src/grpc/early_adopter/_reexport.py
  4. 11
      src/python/src/grpc/early_adopter/implementations.py
  5. 111
      src/python/src/grpc/framework/assembly/implementations.py
  6. 28
      src/python/src/grpc/framework/assembly/implementations_test.py
  7. 56
      src/python/src/grpc/framework/assembly/interfaces.py
  8. 179
      src/python/src/grpc/framework/assembly/utilities.py
  9. 18
      src/python/src/grpc/framework/face/_service.py
  10. 30
      src/python/src/grpc/framework/face/_test_case.py
  11. 268
      src/python/src/grpc/framework/face/implementations.py
  12. 462
      src/python/src/grpc/framework/face/interfaces.py
  13. 16
      src/python/src/grpc/framework/face/testing/blocking_invocation_inline_service_test_case.py
  14. 122
      src/python/src/grpc/framework/face/testing/digest.py
  15. 9
      src/python/src/grpc/framework/face/testing/event_invocation_synchronous_event_service_test_case.py
  16. 17
      src/python/src/grpc/framework/face/testing/future_invocation_asynchronous_event_service_test_case.py
  17. 38
      src/python/src/grpc/framework/face/testing/service.py
  18. 8
      src/python/src/grpc/framework/face/testing/stock_service.py
  19. 53
      src/python/src/grpc/framework/face/testing/test_case.py
  20. 174
      src/python/src/grpc/framework/face/utilities.py

30
src/python/src/grpc/_adapter/_face_test_case.py
Unescape View File

@ -50,31 +50,12 @@ class FaceTestCase(test_case.FaceTestCase, coverage.BlockingCoverage):
"""Provides abstract Face-layer tests a GRPC-backed implementation.""" """Provides abstract Face-layer tests a GRPC-backed implementation."""
def set_up_implementation( def set_up_implementation(
self, self, name, methods, method_implementations,
name, multi_method_implementation):
methods,
inline_value_in_value_out_methods,
inline_value_in_stream_out_methods,
inline_stream_in_value_out_methods,
inline_stream_in_stream_out_methods,
event_value_in_value_out_methods,
event_value_in_stream_out_methods,
event_stream_in_value_out_methods,
event_stream_in_stream_out_methods,
multi_method):
pool = logging_pool.pool(_MAXIMUM_POOL_SIZE) pool = logging_pool.pool(_MAXIMUM_POOL_SIZE)
servicer = face_implementations.servicer( servicer = face_implementations.servicer(
pool, pool, method_implementations, multi_method_implementation)
inline_value_in_value_out_methods=inline_value_in_value_out_methods,
inline_value_in_stream_out_methods=inline_value_in_stream_out_methods,
inline_stream_in_value_out_methods=inline_stream_in_value_out_methods,
inline_stream_in_stream_out_methods=inline_stream_in_stream_out_methods,
event_value_in_value_out_methods=event_value_in_value_out_methods,
event_value_in_stream_out_methods=event_value_in_stream_out_methods,
event_stream_in_value_out_methods=event_stream_in_value_out_methods,
event_stream_in_stream_out_methods=event_stream_in_stream_out_methods,
multi_method=multi_method)
serialization = serial.serialization(methods) serialization = serial.serialization(methods)
@ -96,9 +77,8 @@ class FaceTestCase(test_case.FaceTestCase, coverage.BlockingCoverage):
rear_link.join_fore_link(front) rear_link.join_fore_link(front)
front.join_rear_link(rear_link) front.join_rear_link(rear_link)
server = face_implementations.server() stub = face_implementations.generic_stub(front, pool)
stub = face_implementations.stub(front, pool) return stub, (rear_link, fore_link, front, back)
return server, stub, (rear_link, fore_link, front, back)
def tear_down_implementation(self, memo): def tear_down_implementation(self, memo):
rear_link, fore_link, front, back = memo rear_link, fore_link, front, back = memo

41
src/python/src/grpc/early_adopter/_assembly_utilities.py → src/python/src/grpc/early_adopter/_face_utilities.py
Unescape View File

@ -30,23 +30,20 @@
import abc import abc
import collections import collections
# assembly_interfaces is referenced from specification in this module. # face_interfaces is referenced from specification in this module.
from grpc.framework.assembly import interfaces as assembly_interfaces # pylint: disable=unused-import from grpc.framework.common import cardinality
from grpc.framework.assembly import utilities as assembly_utilities from grpc.framework.face import interfaces as face_interfaces # pylint: disable=unused-import
from grpc.framework.face import utilities as face_utilities
from grpc.early_adopter import _reexport from grpc.early_adopter import _reexport
from grpc.early_adopter import interfaces from grpc.early_adopter import interfaces
# TODO(issue 726): Kill the "implementations" attribute of this in favor
# of the same-information-less-bogusly-represented "cardinalities".
class InvocationBreakdown(object): class InvocationBreakdown(object):
"""An intermediate representation of invocation-side views of RPC methods. """An intermediate representation of invocation-side views of RPC methods.
Attributes: Attributes:
cardinalities: A dictionary from RPC method name to interfaces.Cardinality cardinalities: A dictionary from RPC method name to interfaces.Cardinality
value. value.
implementations: A dictionary from RPC method name to
assembly_interfaces.MethodImplementation describing the method.
request_serializers: A dictionary from RPC method name to callable request_serializers: A dictionary from RPC method name to callable
behavior to be used serializing request values for the RPC. behavior to be used serializing request values for the RPC.
response_deserializers: A dictionary from RPC method name to callable response_deserializers: A dictionary from RPC method name to callable
@ -59,8 +56,7 @@ class _EasyInvocationBreakdown(
InvocationBreakdown, InvocationBreakdown,
collections.namedtuple( collections.namedtuple(
'_EasyInvocationBreakdown', '_EasyInvocationBreakdown',
('cardinalities', 'implementations', 'request_serializers', ('cardinalities', 'request_serializers', 'response_deserializers'))):
'response_deserializers'))):
pass pass
@ -68,8 +64,8 @@ class ServiceBreakdown(object):
"""An intermediate representation of service-side views of RPC methods. """An intermediate representation of service-side views of RPC methods.
Attributes: Attributes:
implementations: A dictionary from RPC method name implementations: A dictionary from RPC method name to
assembly_interfaces.MethodImplementation implementing the RPC method. face_interfaces.MethodImplementation implementing the RPC method.
request_deserializers: A dictionary from RPC method name to callable request_deserializers: A dictionary from RPC method name to callable
behavior to be used deserializing request values for the RPC. behavior to be used deserializing request values for the RPC.
response_serializers: A dictionary from RPC method name to callable response_serializers: A dictionary from RPC method name to callable
@ -97,25 +93,14 @@ def break_down_invocation(method_descriptions):
An InvocationBreakdown corresponding to the given method descriptions. An InvocationBreakdown corresponding to the given method descriptions.
""" """
cardinalities = {} cardinalities = {}
implementations = {}
request_serializers = {} request_serializers = {}
response_deserializers = {} response_deserializers = {}
for name, method_description in method_descriptions.iteritems(): for name, method_description in method_descriptions.iteritems():
cardinality = method_description.cardinality() cardinalities[name] = method_description.cardinality()
cardinalities[name] = cardinality
if cardinality is interfaces.Cardinality.UNARY_UNARY:
implementations[name] = assembly_utilities.unary_unary_inline(None)
elif cardinality is interfaces.Cardinality.UNARY_STREAM:
implementations[name] = assembly_utilities.unary_stream_inline(None)
elif cardinality is interfaces.Cardinality.STREAM_UNARY:
implementations[name] = assembly_utilities.stream_unary_inline(None)
elif cardinality is interfaces.Cardinality.STREAM_STREAM:
implementations[name] = assembly_utilities.stream_stream_inline(None)
request_serializers[name] = method_description.serialize_request request_serializers[name] = method_description.serialize_request
response_deserializers[name] = method_description.deserialize_response response_deserializers[name] = method_description.deserialize_response
return _EasyInvocationBreakdown( return _EasyInvocationBreakdown(
cardinalities, implementations, request_serializers, cardinalities, request_serializers, response_deserializers)
response_deserializers)
def break_down_service(method_descriptions): def break_down_service(method_descriptions):
@ -139,28 +124,28 @@ def break_down_service(method_descriptions):
service_behavior=method_description.service_unary_unary): service_behavior=method_description.service_unary_unary):
return service_behavior( return service_behavior(
request, _reexport.rpc_context(face_rpc_context)) request, _reexport.rpc_context(face_rpc_context))
implementations[name] = assembly_utilities.unary_unary_inline(service) implementations[name] = face_utilities.unary_unary_inline(service)
elif cardinality is interfaces.Cardinality.UNARY_STREAM: elif cardinality is interfaces.Cardinality.UNARY_STREAM:
def service( def service(
request, face_rpc_context, request, face_rpc_context,
service_behavior=method_description.service_unary_stream): service_behavior=method_description.service_unary_stream):
return service_behavior( return service_behavior(
request, _reexport.rpc_context(face_rpc_context)) request, _reexport.rpc_context(face_rpc_context))
implementations[name] = assembly_utilities.unary_stream_inline(service) implementations[name] = face_utilities.unary_stream_inline(service)
elif cardinality is interfaces.Cardinality.STREAM_UNARY: elif cardinality is interfaces.Cardinality.STREAM_UNARY:
def service( def service(
request_iterator, face_rpc_context, request_iterator, face_rpc_context,
service_behavior=method_description.service_stream_unary): service_behavior=method_description.service_stream_unary):
return service_behavior( return service_behavior(
request_iterator, _reexport.rpc_context(face_rpc_context)) request_iterator, _reexport.rpc_context(face_rpc_context))
implementations[name] = assembly_utilities.stream_unary_inline(service) implementations[name] = face_utilities.stream_unary_inline(service)
elif cardinality is interfaces.Cardinality.STREAM_STREAM: elif cardinality is interfaces.Cardinality.STREAM_STREAM:
def service( def service(
request_iterator, face_rpc_context, request_iterator, face_rpc_context,
service_behavior=method_description.service_stream_stream): service_behavior=method_description.service_stream_stream):
return service_behavior( return service_behavior(
request_iterator, _reexport.rpc_context(face_rpc_context)) request_iterator, _reexport.rpc_context(face_rpc_context))
implementations[name] = assembly_utilities.stream_stream_inline(service) implementations[name] = face_utilities.stream_stream_inline(service)
request_deserializers[name] = method_description.deserialize_request request_deserializers[name] = method_description.deserialize_request
response_serializers[name] = method_description.serialize_response response_serializers[name] = method_description.serialize_response

47
src/python/src/grpc/early_adopter/_reexport.py
Unescape View File

@ -27,12 +27,20 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
from grpc.framework.common import cardinality
from grpc.framework.face import exceptions as face_exceptions from grpc.framework.face import exceptions as face_exceptions
from grpc.framework.face import interfaces as face_interfaces from grpc.framework.face import interfaces as face_interfaces
from grpc.framework.foundation import future from grpc.framework.foundation import future
from grpc.early_adopter import exceptions from grpc.early_adopter import exceptions
from grpc.early_adopter import interfaces from grpc.early_adopter import interfaces
_EARLY_ADOPTER_CARDINALITY_TO_COMMON_CARDINALITY = {
interfaces.Cardinality.UNARY_UNARY: cardinality.Cardinality.UNARY_UNARY,
interfaces.Cardinality.UNARY_STREAM: cardinality.Cardinality.UNARY_STREAM,
interfaces.Cardinality.STREAM_UNARY: cardinality.Cardinality.STREAM_UNARY,
interfaces.Cardinality.STREAM_STREAM: cardinality.Cardinality.STREAM_STREAM,
}
_ABORTION_REEXPORT = { _ABORTION_REEXPORT = {
face_interfaces.Abortion.CANCELLED: interfaces.Abortion.CANCELLED, face_interfaces.Abortion.CANCELLED: interfaces.Abortion.CANCELLED,
face_interfaces.Abortion.EXPIRED: interfaces.Abortion.EXPIRED, face_interfaces.Abortion.EXPIRED: interfaces.Abortion.EXPIRED,
@ -142,28 +150,28 @@ class _RpcContext(interfaces.RpcContext):
class _UnaryUnarySyncAsync(interfaces.UnaryUnarySyncAsync): class _UnaryUnarySyncAsync(interfaces.UnaryUnarySyncAsync):
def __init__(self, face_unary_unary_sync_async): def __init__(self, face_unary_unary_multi_callable):
self._underlying = face_unary_unary_sync_async self._underlying = face_unary_unary_multi_callable
def __call__(self, request, timeout): def __call__(self, request, timeout):
return _call_reexporting_errors( return _call_reexporting_errors(
self._underlying, request, timeout) self._underlying, request, timeout)
def async(self, request, timeout): def async(self, request, timeout):
return _ReexportedFuture(self._underlying.async(request, timeout)) return _ReexportedFuture(self._underlying.future(request, timeout))
class _StreamUnarySyncAsync(interfaces.StreamUnarySyncAsync): class _StreamUnarySyncAsync(interfaces.StreamUnarySyncAsync):
def __init__(self, face_stream_unary_sync_async): def __init__(self, face_stream_unary_multi_callable):
self._underlying = face_stream_unary_sync_async self._underlying = face_stream_unary_multi_callable
def __call__(self, request_iterator, timeout): def __call__(self, request_iterator, timeout):
return _call_reexporting_errors( return _call_reexporting_errors(
self._underlying, request_iterator, timeout) self._underlying, request_iterator, timeout)
def async(self, request_iterator, timeout): def async(self, request_iterator, timeout):
return _ReexportedFuture(self._underlying.async(request_iterator, timeout)) return _ReexportedFuture(self._underlying.future(request_iterator, timeout))
class _Stub(interfaces.Stub): class _Stub(interfaces.Stub):
@ -182,31 +190,40 @@ class _Stub(interfaces.Stub):
def __getattr__(self, attr): def __getattr__(self, attr):
underlying_attr = self._assembly_stub.__getattr__(attr) underlying_attr = self._assembly_stub.__getattr__(attr)
cardinality = self._cardinalities.get(attr) method_cardinality = self._cardinalities.get(attr)
# TODO(nathaniel): unify this trick with its other occurrence in the code. # TODO(nathaniel): unify this trick with its other occurrence in the code.
if cardinality is None: if method_cardinality is None:
for name, cardinality in self._cardinalities.iteritems(): for name, method_cardinality in self._cardinalities.iteritems():
last_slash_index = name.rfind('/') last_slash_index = name.rfind('/')
if 0 <= last_slash_index and name[last_slash_index + 1:] == attr: if 0 <= last_slash_index and name[last_slash_index + 1:] == attr:
break break
else: else:
raise AttributeError(attr) raise AttributeError(attr)
if cardinality is interfaces.Cardinality.UNARY_UNARY: if method_cardinality is interfaces.Cardinality.UNARY_UNARY:
return _UnaryUnarySyncAsync(underlying_attr) return _UnaryUnarySyncAsync(underlying_attr)
elif cardinality is interfaces.Cardinality.UNARY_STREAM: elif method_cardinality is interfaces.Cardinality.UNARY_STREAM:
return lambda request, timeout: _CancellableIterator( return lambda request, timeout: _CancellableIterator(
underlying_attr(request, timeout)) underlying_attr(request, timeout))
elif cardinality is interfaces.Cardinality.STREAM_UNARY: elif method_cardinality is interfaces.Cardinality.STREAM_UNARY:
return _StreamUnarySyncAsync(underlying_attr) return _StreamUnarySyncAsync(underlying_attr)
elif cardinality is interfaces.Cardinality.STREAM_STREAM: elif method_cardinality is interfaces.Cardinality.STREAM_STREAM:
return lambda request_iterator, timeout: _CancellableIterator( return lambda request_iterator, timeout: _CancellableIterator(
underlying_attr(request_iterator, timeout)) underlying_attr(request_iterator, timeout))
else: else:
raise AttributeError(attr) raise AttributeError(attr)
def common_cardinalities(early_adopter_cardinalities):
common_cardinalities = {}
for name, early_adopter_cardinality in early_adopter_cardinalities.iteritems():
common_cardinalities[name] = _EARLY_ADOPTER_CARDINALITY_TO_COMMON_CARDINALITY[
early_adopter_cardinality]
return common_cardinalities
def rpc_context(face_rpc_context): def rpc_context(face_rpc_context):
return _RpcContext(face_rpc_context) return _RpcContext(face_rpc_context)
def stub(assembly_stub, cardinalities): def stub(face_stub, cardinalities):
return _Stub(assembly_stub, cardinalities) return _Stub(face_stub, cardinalities)

11
src/python/src/grpc/early_adopter/implementations.py
Unescape View File

@ -33,7 +33,7 @@ import threading
from grpc._adapter import fore as _fore from grpc._adapter import fore as _fore
from grpc._adapter import rear as _rear from grpc._adapter import rear as _rear
from grpc.early_adopter import _assembly_utilities from grpc.early_adopter import _face_utilities
from grpc.early_adopter import _reexport from grpc.early_adopter import _reexport
from grpc.early_adopter import interfaces from grpc.early_adopter import interfaces
from grpc.framework.assembly import implementations as _assembly_implementations from grpc.framework.assembly import implementations as _assembly_implementations
@ -95,12 +95,13 @@ class _Server(interfaces.Server):
def _build_stub(breakdown, activated_rear_link): def _build_stub(breakdown, activated_rear_link):
assembly_stub = _assembly_implementations.assemble_dynamic_inline_stub( assembly_stub = _assembly_implementations.assemble_dynamic_inline_stub(
breakdown.implementations, activated_rear_link) _reexport.common_cardinalities(breakdown.cardinalities),
activated_rear_link)
return _reexport.stub(assembly_stub, breakdown.cardinalities) return _reexport.stub(assembly_stub, breakdown.cardinalities)
def _build_server(methods, port, private_key, certificate_chain): def _build_server(methods, port, private_key, certificate_chain):
breakdown = _assembly_utilities.break_down_service(methods) breakdown = _face_utilities.break_down_service(methods)
return _Server(breakdown, port, private_key, certificate_chain) return _Server(breakdown, port, private_key, certificate_chain)
@ -117,7 +118,7 @@ def insecure_stub(methods, host, port):
Returns: Returns:
An interfaces.Stub affording RPC invocation. An interfaces.Stub affording RPC invocation.
""" """
breakdown = _assembly_utilities.break_down_invocation(methods) breakdown = _face_utilities.break_down_invocation(methods)
activated_rear_link = _rear.activated_rear_link( activated_rear_link = _rear.activated_rear_link(
host, port, breakdown.request_serializers, host, port, breakdown.request_serializers,
breakdown.response_deserializers) breakdown.response_deserializers)
@ -147,7 +148,7 @@ def secure_stub(
Returns: Returns:
An interfaces.Stub affording RPC invocation. An interfaces.Stub affording RPC invocation.
""" """
breakdown = _assembly_utilities.break_down_invocation(methods) breakdown = _face_utilities.break_down_invocation(methods)
activated_rear_link = _rear.secure_activated_rear_link( activated_rear_link = _rear.secure_activated_rear_link(
host, port, breakdown.request_serializers, host, port, breakdown.request_serializers,
breakdown.response_deserializers, root_certificates, private_key, breakdown.response_deserializers, root_certificates, private_key,

111
src/python/src/grpc/framework/assembly/implementations.py
Unescape View File

@ -66,7 +66,7 @@ class _FaceStub(object):
self._rear_link.start() self._rear_link.start()
self._rear_link.join_fore_link(self._front) self._rear_link.join_fore_link(self._front)
self._front.join_rear_link(self._rear_link) self._front.join_rear_link(self._rear_link)
self._under_stub = face_implementations.stub(self._front, self._pool) self._under_stub = face_implementations.generic_stub(self._front, self._pool)
def __exit__(self, exc_type, exc_val, exc_tb): def __exit__(self, exc_type, exc_val, exc_tb):
with self._lock: with self._lock:
@ -86,18 +86,18 @@ class _FaceStub(object):
return getattr(self._under_stub, attr) return getattr(self._under_stub, attr)
def _behaviors(implementations, front, pool): def _behaviors(method_cardinalities, front, pool):
behaviors = {} behaviors = {}
stub = face_implementations.stub(front, pool) stub = face_implementations.generic_stub(front, pool)
for name, implementation in implementations.iteritems(): for name, method_cardinality in method_cardinalities.iteritems():
if implementation.cardinality is cardinality.Cardinality.UNARY_UNARY: if method_cardinality is cardinality.Cardinality.UNARY_UNARY:
behaviors[name] = stub.unary_unary_sync_async(name) behaviors[name] = stub.unary_unary_multi_callable(name)
elif implementation.cardinality is cardinality.Cardinality.UNARY_STREAM: elif method_cardinality is cardinality.Cardinality.UNARY_STREAM:
behaviors[name] = lambda request, context, bound_name=name: ( behaviors[name] = lambda request, context, bound_name=name: (
stub.inline_value_in_stream_out(bound_name, request, context)) stub.inline_value_in_stream_out(bound_name, request, context))
elif implementation.cardinality is cardinality.Cardinality.STREAM_UNARY: elif method_cardinality is cardinality.Cardinality.STREAM_UNARY:
behaviors[name] = stub.stream_unary_sync_async(name) behaviors[name] = stub.stream_unary_multi_callable(name)
elif implementation.cardinality is cardinality.Cardinality.STREAM_STREAM: elif method_cardinality is cardinality.Cardinality.STREAM_STREAM:
behaviors[name] = lambda request_iterator, context, bound_name=name: ( behaviors[name] = lambda request_iterator, context, bound_name=name: (
stub.inline_stream_in_stream_out( stub.inline_stream_in_stream_out(
bound_name, request_iterator, context)) bound_name, request_iterator, context))
@ -106,8 +106,8 @@ def _behaviors(implementations, front, pool):
class _DynamicInlineStub(object): class _DynamicInlineStub(object):
def __init__(self, implementations, rear_link): def __init__(self, cardinalities, rear_link):
self._implementations = implementations self._cardinalities = cardinalities
self._rear_link = rear_link self._rear_link = rear_link
self._lock = threading.Lock() self._lock = threading.Lock()
self._pool = None self._pool = None
@ -123,7 +123,7 @@ class _DynamicInlineStub(object):
self._rear_link.join_fore_link(self._front) self._rear_link.join_fore_link(self._front)
self._front.join_rear_link(self._rear_link) self._front.join_rear_link(self._rear_link)
self._behaviors = _behaviors( self._behaviors = _behaviors(
self._implementations, self._front, self._pool) self._cardinalities, self._front, self._pool)
return self return self
def __exit__(self, exc_type, exc_val, exc_tb): def __exit__(self, exc_type, exc_val, exc_tb):
@ -151,58 +151,6 @@ class _DynamicInlineStub(object):
return behavior return behavior
def _servicer(implementations, pool):
inline_value_in_value_out_methods = {}
inline_value_in_stream_out_methods = {}
inline_stream_in_value_out_methods = {}
inline_stream_in_stream_out_methods = {}
event_value_in_value_out_methods = {}
event_value_in_stream_out_methods = {}
event_stream_in_value_out_methods = {}
event_stream_in_stream_out_methods = {}
for name, implementation in implementations.iteritems():
if implementation.cardinality is cardinality.Cardinality.UNARY_UNARY:
if implementation.style is style.Service.INLINE:
inline_value_in_value_out_methods[name] = (
face_utilities.inline_unary_unary_method(implementation.unary_unary_inline))
elif implementation.style is style.Service.EVENT:
event_value_in_value_out_methods[name] = (
face_utilities.event_unary_unary_method(implementation.unary_unary_event))
elif implementation.cardinality is cardinality.Cardinality.UNARY_STREAM:
if implementation.style is style.Service.INLINE:
inline_value_in_stream_out_methods[name] = (
face_utilities.inline_unary_stream_method(implementation.unary_stream_inline))
elif implementation.style is style.Service.EVENT:
event_value_in_stream_out_methods[name] = (
face_utilities.event_unary_stream_method(implementation.unary_stream_event))
if implementation.cardinality is cardinality.Cardinality.STREAM_UNARY:
if implementation.style is style.Service.INLINE:
inline_stream_in_value_out_methods[name] = (
face_utilities.inline_stream_unary_method(implementation.stream_unary_inline))
elif implementation.style is style.Service.EVENT:
event_stream_in_value_out_methods[name] = (
face_utilities.event_stream_unary_method(implementation.stream_unary_event))
elif implementation.cardinality is cardinality.Cardinality.STREAM_STREAM:
if implementation.style is style.Service.INLINE:
inline_stream_in_stream_out_methods[name] = (
face_utilities.inline_stream_stream_method(implementation.stream_stream_inline))
elif implementation.style is style.Service.EVENT:
event_stream_in_stream_out_methods[name] = (
face_utilities.event_stream_stream_method(implementation.stream_stream_event))
return face_implementations.servicer(
pool,
inline_value_in_value_out_methods=inline_value_in_value_out_methods,
inline_value_in_stream_out_methods=inline_value_in_stream_out_methods,
inline_stream_in_value_out_methods=inline_stream_in_value_out_methods,
inline_stream_in_stream_out_methods=inline_stream_in_stream_out_methods,
event_value_in_value_out_methods=event_value_in_value_out_methods,
event_value_in_stream_out_methods=event_value_in_stream_out_methods,
event_stream_in_value_out_methods=event_stream_in_value_out_methods,
event_stream_in_stream_out_methods=event_stream_in_stream_out_methods)
class _ServiceAssembly(interfaces.Server): class _ServiceAssembly(interfaces.Server):
def __init__(self, implementations, fore_link): def __init__(self, implementations, fore_link):
@ -215,7 +163,8 @@ class _ServiceAssembly(interfaces.Server):
def _start(self): def _start(self):
with self._lock: with self._lock:
self._pool = logging_pool.pool(_THREAD_POOL_SIZE) self._pool = logging_pool.pool(_THREAD_POOL_SIZE)
servicer = _servicer(self._implementations, self._pool) servicer = face_implementations.servicer(
self._pool, self._implementations, None)
self._back = tickets_implementations.back( self._back = tickets_implementations.back(
servicer, self._pool, self._pool, self._pool, _ONE_DAY_IN_SECONDS, servicer, self._pool, self._pool, self._pool, _ONE_DAY_IN_SECONDS,
_ONE_DAY_IN_SECONDS) _ONE_DAY_IN_SECONDS)
@ -251,7 +200,7 @@ class _ServiceAssembly(interfaces.Server):
def assemble_face_stub(activated_rear_link): def assemble_face_stub(activated_rear_link):
"""Assembles a face_interfaces.Stub. """Assembles a face_interfaces.GenericStub.
The returned object is a context manager and may only be used in context to The returned object is a context manager and may only be used in context to
invoke RPCs. invoke RPCs.
@ -262,12 +211,12 @@ def assemble_face_stub(activated_rear_link):
when passed to this method. when passed to this method.
Returns: Returns:
A face_interfaces.Stub on which, in context, RPCs can be invoked. A face_interfaces.GenericStub on which, in context, RPCs can be invoked.
""" """
return _FaceStub(activated_rear_link) return _FaceStub(activated_rear_link)
def assemble_dynamic_inline_stub(implementations, activated_rear_link): def assemble_dynamic_inline_stub(cardinalities, activated_rear_link):
"""Assembles a stub with method names for attributes. """Assembles a stub with method names for attributes.
The returned object is a context manager and may only be used in context to The returned object is a context manager and may only be used in context to
@ -276,29 +225,27 @@ def assemble_dynamic_inline_stub(implementations, activated_rear_link):
The returned object, when used in context, will respond to attribute access The returned object, when used in context, will respond to attribute access
as follows: if the requested attribute is the name of a unary-unary RPC as follows: if the requested attribute is the name of a unary-unary RPC
method, the value of the attribute will be a method, the value of the attribute will be a
face_interfaces.UnaryUnarySyncAsync with which to invoke the RPC method. If face_interfaces.UnaryUnaryMultiCallable with which to invoke the RPC method.
the requested attribute is the name of a unary-stream RPC method, the value If the requested attribute is the name of a unary-stream RPC method, the
of the attribute will be a callable with the semantics of value of the attribute will be a face_interfaces.UnaryStreamMultiCallable
face_interfaces.Stub.inline_value_in_stream_out, minus the "name" parameter,
with which to invoke the RPC method. If the requested attribute is the name with which to invoke the RPC method. If the requested attribute is the name
of a stream-unary RPC method, the value of the attribute will be a of a stream-unary RPC method, the value of the attribute will be a
face_interfaces.StreamUnarySyncAsync with which to invoke the RPC method. If face_interfaces.StreamUnaryMultiCallable with which to invoke the RPC method.
the requested attribute is the name of a stream-stream RPC method, the value If the requested attribute is the name of a stream-stream RPC method, the
of the attribute will be a callable with the semantics of value of the attribute will be a face_interfaces.StreamStreamMultiCallable
face_interfaces.Stub.inline_stream_in_stream_out, minus the "name" parameter,
with which to invoke the RPC method. with which to invoke the RPC method.
Args: Args:
implementations: A dictionary from RPC method name to cardinalities: A dictionary from RPC method name to cardinality.Cardinality
interfaces.MethodImplementation. value identifying the cardinality of the named RPC method.
activated_rear_link: An object that is both a tickets_interfaces.RearLink activated_rear_link: An object that is both a tickets_interfaces.RearLink
and an activated.Activated. The object should be in the inactive state and an activated.Activated. The object should be in the inactive state
when passed to this method. when passed to this method.
Returns: Returns:
A stub on which, in context, RPCs can be invoked. A face_interfaces.DynamicStub on which, in context, RPCs can be invoked.
""" """
return _DynamicInlineStub(implementations, activated_rear_link) return _DynamicInlineStub(cardinalities, activated_rear_link)
def assemble_service(implementations, activated_fore_link): def assemble_service(implementations, activated_fore_link):
@ -306,7 +253,7 @@ def assemble_service(implementations, activated_fore_link):
Args: Args:
implementations: A dictionary from RPC method name to implementations: A dictionary from RPC method name to
interfaces.MethodImplementation. face_interfaces.MethodImplementation.
activated_fore_link: An object that is both a tickets_interfaces.ForeLink activated_fore_link: An object that is both a tickets_interfaces.ForeLink
and an activated.Activated. The object should be in the inactive state and an activated.Activated. The object should be in the inactive state
when passed to this method. when passed to this method.

28
src/python/src/grpc/framework/assembly/implementations_test.py
Unescape View File

@ -35,11 +35,11 @@ import threading
import unittest import unittest
from grpc.framework.assembly import implementations from grpc.framework.assembly import implementations
from grpc.framework.assembly import utilities
from grpc.framework.base import interfaces from grpc.framework.base import interfaces
from grpc.framework.base.packets import packets as tickets from grpc.framework.base.packets import packets as tickets
from grpc.framework.base.packets import interfaces as tickets_interfaces from grpc.framework.base.packets import interfaces as tickets_interfaces
from grpc.framework.base.packets import null from grpc.framework.base.packets import null
from grpc.framework.face import utilities as face_utilities
from grpc.framework.foundation import logging_pool from grpc.framework.foundation import logging_pool
from grpc._junkdrawer import math_pb2 from grpc._junkdrawer import math_pb2
@ -81,12 +81,16 @@ def _sum(request_iterator, unused_context):
_IMPLEMENTATIONS = { _IMPLEMENTATIONS = {
DIV: utilities.unary_unary_inline(_div), DIV: face_utilities.unary_unary_inline(_div),
DIV_MANY: utilities.stream_stream_inline(_div_many), DIV_MANY: face_utilities.stream_stream_inline(_div_many),
FIB: utilities.unary_stream_inline(_fib), FIB: face_utilities.unary_stream_inline(_fib),
SUM: utilities.stream_unary_inline(_sum), SUM: face_utilities.stream_unary_inline(_sum),
} }
_CARDINALITIES = {
name: implementation.cardinality
for name, implementation in _IMPLEMENTATIONS.iteritems()}
_TIMEOUT = 10 _TIMEOUT = 10
@ -170,8 +174,8 @@ class FaceStubTest(unittest.TestCase):
face_stub = implementations.assemble_face_stub(pipe) face_stub = implementations.assemble_face_stub(pipe)
with service, face_stub: with service, face_stub:
sync_async = face_stub.stream_unary_sync_async(SUM) multi_callable = face_stub.stream_unary_multi_callable(SUM)
response_future = sync_async.async( response_future = multi_callable.future(
(math_pb2.Num(num=index) for index in range(stream_length)), (math_pb2.Num(num=index) for index in range(stream_length)),
_TIMEOUT) _TIMEOUT)
self.assertEqual( self.assertEqual(
@ -214,7 +218,7 @@ class DynamicInlineStubTest(unittest.TestCase):
pipe = PipeLink() pipe = PipeLink()
service = implementations.assemble_service(_IMPLEMENTATIONS, pipe) service = implementations.assemble_service(_IMPLEMENTATIONS, pipe)
dynamic_stub = implementations.assemble_dynamic_inline_stub( dynamic_stub = implementations.assemble_dynamic_inline_stub(
_IMPLEMENTATIONS, pipe) _CARDINALITIES, pipe)
service.start() service.start()
with dynamic_stub: with dynamic_stub:
@ -229,7 +233,7 @@ class DynamicInlineStubTest(unittest.TestCase):
pipe = PipeLink() pipe = PipeLink()
service = implementations.assemble_service(_IMPLEMENTATIONS, pipe) service = implementations.assemble_service(_IMPLEMENTATIONS, pipe)
dynamic_stub = implementations.assemble_dynamic_inline_stub( dynamic_stub = implementations.assemble_dynamic_inline_stub(
_IMPLEMENTATIONS, pipe) _CARDINALITIES, pipe)
with service, dynamic_stub: with service, dynamic_stub:
response_iterator = dynamic_stub.Fib( response_iterator = dynamic_stub.Fib(
@ -244,10 +248,10 @@ class DynamicInlineStubTest(unittest.TestCase):
pipe = PipeLink() pipe = PipeLink()
service = implementations.assemble_service(_IMPLEMENTATIONS, pipe) service = implementations.assemble_service(_IMPLEMENTATIONS, pipe)
dynamic_stub = implementations.assemble_dynamic_inline_stub( dynamic_stub = implementations.assemble_dynamic_inline_stub(
_IMPLEMENTATIONS, pipe) _CARDINALITIES, pipe)
with service, dynamic_stub: with service, dynamic_stub:
response_future = dynamic_stub.Sum.async( response_future = dynamic_stub.Sum.future(
(math_pb2.Num(num=index) for index in range(stream_length)), (math_pb2.Num(num=index) for index in range(stream_length)),
_TIMEOUT) _TIMEOUT)
self.assertEqual( self.assertEqual(
@ -261,7 +265,7 @@ class DynamicInlineStubTest(unittest.TestCase):
pipe = PipeLink() pipe = PipeLink()
service = implementations.assemble_service(_IMPLEMENTATIONS, pipe) service = implementations.assemble_service(_IMPLEMENTATIONS, pipe)
dynamic_stub = implementations.assemble_dynamic_inline_stub( dynamic_stub = implementations.assemble_dynamic_inline_stub(
_IMPLEMENTATIONS, pipe) _CARDINALITIES, pipe)
with service, dynamic_stub: with service, dynamic_stub:
response_iterator = dynamic_stub.DivMany( response_iterator = dynamic_stub.DivMany(

56
src/python/src/grpc/framework/assembly/interfaces.py
Unescape View File

@ -33,63 +33,7 @@
import abc import abc
# cardinality, style, and stream are referenced from specification in this
# module.
from grpc.framework.common import cardinality # pylint: disable=unused-import
from grpc.framework.common import style # pylint: disable=unused-import
from grpc.framework.foundation import activated from grpc.framework.foundation import activated
from grpc.framework.foundation import stream # pylint: disable=unused-import
class MethodImplementation(object):
"""A sum type that describes an RPC method implementation.
Attributes:
cardinality: A cardinality.Cardinality value.
style: A style.Service value.
unary_unary_inline: The implementation of the RPC method as a callable
value that takes a request value and a face_interfaces.RpcContext object
and returns a response value. Only non-None if cardinality is
cardinality.Cardinality.UNARY_UNARY and style is style.Service.INLINE.
unary_stream_inline: The implementation of the RPC method as a callable
value that takes a request value and a face_interfaces.RpcContext object
and returns an iterator of response values. Only non-None if cardinality
is cardinality.Cardinality.UNARY_STREAM and style is
style.Service.INLINE.
stream_unary_inline: The implementation of the RPC method as a callable
value that takes an iterator of request values and a
face_interfaces.RpcContext object and returns a response value. Only
non-None if cardinality is cardinality.Cardinality.STREAM_UNARY and style
is style.Service.INLINE.
stream_stream_inline: The implementation of the RPC method as a callable
value that takes an iterator of request values and a
face_interfaces.RpcContext object and returns an iterator of response
values. Only non-None if cardinality is
cardinality.Cardinality.STREAM_STREAM and style is style.Service.INLINE.
unary_unary_event: The implementation of the RPC method as a callable value
that takes a request value, a response callback to which to pass the
response value of the RPC, and a face_interfaces.RpcContext. Only
non-None if cardinality is cardinality.Cardinality.UNARY_UNARY and style
is style.Service.EVENT.
unary_stream_event: The implementation of the RPC method as a callable
value that takes a request value, a stream.Consumer to which to pass the
the response values of the RPC, and a face_interfaces.RpcContext. Only
non-None if cardinality is cardinality.Cardinality.UNARY_STREAM and style
is style.Service.EVENT.
stream_unary_event: The implementation of the RPC method as a callable
value that takes a response callback to which to pass the response value
of the RPC and a face_interfaces.RpcContext and returns a stream.Consumer
to which the request values of the RPC should be passed. Only non-None if
cardinality is cardinality.Cardinality.STREAM_UNARY and style is
style.Service.EVENT.
stream_stream_event: The implementation of the RPC method as a callable
value that takes a stream.Consumer to which to pass the response values
of the RPC and a face_interfaces.RpcContext and returns a stream.Consumer
to which the request values of the RPC should be passed. Only non-None if
cardinality is cardinality.Cardinality.STREAM_STREAM and style is
style.Service.EVENT.
"""
__metaclass__ = abc.ABCMeta
class Server(activated.Activated): class Server(activated.Activated):

179
src/python/src/grpc/framework/assembly/utilities.py
Unescape View File

@ -1,179 +0,0 @@
# 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.
"""Utilities for assembling RPC framework values."""
import collections
from grpc.framework.assembly import interfaces
from grpc.framework.common import cardinality
from grpc.framework.common import style
from grpc.framework.face import interfaces as face_interfaces
from grpc.framework.foundation import stream
class _MethodImplementation(
interfaces.MethodImplementation,
collections.namedtuple(
'_MethodImplementation',
['cardinality', 'style', 'unary_unary_inline', 'unary_stream_inline',
'stream_unary_inline', 'stream_stream_inline', 'unary_unary_event',
'unary_stream_event', 'stream_unary_event', 'stream_stream_event',])):
pass
def unary_unary_inline(behavior):
"""Creates an interfaces.MethodImplementation for the given behavior.
Args:
behavior: The implementation of a unary-unary RPC method as a callable value
that takes a request value and a face_interfaces.RpcContext object and
returns a response value.
Returns:
An interfaces.MethodImplementation derived from the given behavior.
"""
return _MethodImplementation(
cardinality.Cardinality.UNARY_UNARY, style.Service.INLINE, behavior,
None, None, None, None, None, None, None)
def unary_stream_inline(behavior):
"""Creates an interfaces.MethodImplementation for the given behavior.
Args:
behavior: The implementation of a unary-stream RPC method as a callable
value that takes a request value and a face_interfaces.RpcContext object
and returns an iterator of response values.
Returns:
An interfaces.MethodImplementation derived from the given behavior.
"""
return _MethodImplementation(
cardinality.Cardinality.UNARY_STREAM, style.Service.INLINE, None,
behavior, None, None, None, None, None, None)
def stream_unary_inline(behavior):
"""Creates an interfaces.MethodImplementation for the given behavior.
Args:
behavior: The implementation of a stream-unary RPC method as a callable
value that takes an iterator of request values and a
face_interfaces.RpcContext object and returns a response value.
Returns:
An interfaces.MethodImplementation derived from the given behavior.
"""
return _MethodImplementation(
cardinality.Cardinality.STREAM_UNARY, style.Service.INLINE, None, None,
behavior, None, None, None, None, None)
def stream_stream_inline(behavior):
"""Creates an interfaces.MethodImplementation for the given behavior.
Args:
behavior: The implementation of a stream-stream RPC method as a callable
value that takes an iterator of request values and a
face_interfaces.RpcContext object and returns an iterator of response
values.
Returns:
An interfaces.MethodImplementation derived from the given behavior.
"""
return _MethodImplementation(
cardinality.Cardinality.STREAM_STREAM, style.Service.INLINE, None, None,
None, behavior, None, None, None, None)
def unary_unary_event(behavior):
"""Creates an interfaces.MethodImplementation for the given behavior.
Args:
behavior: The implementation of a unary-unary RPC method as a callable
value that takes a request value, a response callback to which to pass
the response value of the RPC, and a face_interfaces.RpcContext.
Returns:
An interfaces.MethodImplementation derived from the given behavior.
"""
return _MethodImplementation(
cardinality.Cardinality.UNARY_UNARY, style.Service.EVENT, None, None,
None, None, behavior, None, None, None)
def unary_stream_event(behavior):
"""Creates an interfaces.MethodImplementation for the given behavior.
Args:
behavior: The implementation of a unary-stream RPC method as a callable
value that takes a request value, a stream.Consumer to which to pass the
the response values of the RPC, and a face_interfaces.RpcContext.
Returns:
An interfaces.MethodImplementation derived from the given behavior.
"""
return _MethodImplementation(
cardinality.Cardinality.UNARY_STREAM, style.Service.EVENT, None, None,
None, None, None, behavior, None, None)
def stream_unary_event(behavior):
"""Creates an interfaces.MethodImplementation for the given behavior.
Args:
behavior: The implementation of a stream-unary RPC method as a callable
value that takes a response callback to which to pass the response value
of the RPC and a face_interfaces.RpcContext and returns a stream.Consumer
to which the request values of the RPC should be passed.
Returns:
An interfaces.MethodImplementation derived from the given behavior.
"""
return _MethodImplementation(
cardinality.Cardinality.STREAM_UNARY, style.Service.EVENT, None, None,
None, None, None, None, behavior, None)
def stream_stream_event(behavior):
"""Creates an interfaces.MethodImplementation for the given behavior.
Args:
behavior: The implementation of a stream-stream RPC method as a callable
value that takes a stream.Consumer to which to pass the response values
of the RPC and a face_interfaces.RpcContext and returns a stream.Consumer
to which the request values of the RPC should be passed.
Returns:
An interfaces.MethodImplementation derived from the given behavior.
"""
return _MethodImplementation(
cardinality.Cardinality.STREAM_STREAM, style.Service.EVENT, None, None,
None, None, None, None, None, behavior)

18
src/python/src/grpc/framework/face/_service.py
Unescape View File

@ -105,15 +105,14 @@ def adapt_inline_value_in_value_out(method):
def adaptation(response_consumer, operation_context): def adaptation(response_consumer, operation_context):
rpc_context = _control.RpcContext(operation_context) rpc_context = _control.RpcContext(operation_context)
return stream_util.TransformingConsumer( return stream_util.TransformingConsumer(
lambda request: method.service(request, rpc_context), response_consumer) lambda request: method(request, rpc_context), response_consumer)
return adaptation return adaptation
def adapt_inline_value_in_stream_out(method): def adapt_inline_value_in_stream_out(method):
def adaptation(response_consumer, operation_context): def adaptation(response_consumer, operation_context):
rpc_context = _control.RpcContext(operation_context) rpc_context = _control.RpcContext(operation_context)
return _ValueInStreamOutConsumer( return _ValueInStreamOutConsumer(method, rpc_context, response_consumer)
method.service, rpc_context, response_consumer)
return adaptation return adaptation
@ -123,7 +122,7 @@ def adapt_inline_stream_in_value_out(method, pool):
operation_context.add_termination_callback(rendezvous.set_outcome) operation_context.add_termination_callback(rendezvous.set_outcome)
def in_pool_thread(): def in_pool_thread():
response_consumer.consume_and_terminate( response_consumer.consume_and_terminate(
method.service(rendezvous, _control.RpcContext(operation_context))) method(rendezvous, _control.RpcContext(operation_context)))
pool.submit(_pool_wrap(in_pool_thread, operation_context)) pool.submit(_pool_wrap(in_pool_thread, operation_context))
return rendezvous return rendezvous
return adaptation return adaptation
@ -149,7 +148,7 @@ def adapt_inline_stream_in_stream_out(method, pool):
operation_context.add_termination_callback(rendezvous.set_outcome) operation_context.add_termination_callback(rendezvous.set_outcome)
def in_pool_thread(): def in_pool_thread():
_control.pipe_iterator_to_consumer( _control.pipe_iterator_to_consumer(
method.service(rendezvous, _control.RpcContext(operation_context)), method(rendezvous, _control.RpcContext(operation_context)),
response_consumer, operation_context.is_active, True) response_consumer, operation_context.is_active, True)
pool.submit(_pool_wrap(in_pool_thread, operation_context)) pool.submit(_pool_wrap(in_pool_thread, operation_context))
return rendezvous return rendezvous
@ -159,7 +158,7 @@ def adapt_inline_stream_in_stream_out(method, pool):
def adapt_event_value_in_value_out(method): def adapt_event_value_in_value_out(method):
def adaptation(response_consumer, operation_context): def adaptation(response_consumer, operation_context):
def on_payload(payload): def on_payload(payload):
method.service( method(
payload, response_consumer.consume_and_terminate, payload, response_consumer.consume_and_terminate,
_control.RpcContext(operation_context)) _control.RpcContext(operation_context))
return _control.UnaryConsumer(on_payload) return _control.UnaryConsumer(on_payload)
@ -169,7 +168,7 @@ def adapt_event_value_in_value_out(method):
def adapt_event_value_in_stream_out(method): def adapt_event_value_in_stream_out(method):
def adaptation(response_consumer, operation_context): def adaptation(response_consumer, operation_context):
def on_payload(payload): def on_payload(payload):
method.service( method(
payload, response_consumer, _control.RpcContext(operation_context)) payload, response_consumer, _control.RpcContext(operation_context))
return _control.UnaryConsumer(on_payload) return _control.UnaryConsumer(on_payload)
return adaptation return adaptation
@ -178,12 +177,11 @@ def adapt_event_value_in_stream_out(method):
def adapt_event_stream_in_value_out(method): def adapt_event_stream_in_value_out(method):
def adaptation(response_consumer, operation_context): def adaptation(response_consumer, operation_context):
rpc_context = _control.RpcContext(operation_context) rpc_context = _control.RpcContext(operation_context)
return method.service(response_consumer.consume_and_terminate, rpc_context) return method(response_consumer.consume_and_terminate, rpc_context)
return adaptation return adaptation
def adapt_event_stream_in_stream_out(method): def adapt_event_stream_in_stream_out(method):
def adaptation(response_consumer, operation_context): def adaptation(response_consumer, operation_context):
return method.service( return method(response_consumer, _control.RpcContext(operation_context))
response_consumer, _control.RpcContext(operation_context))
return adaptation return adaptation

30
src/python/src/grpc/framework/face/_test_case.py
Unescape View File

@ -42,37 +42,17 @@ class FaceTestCase(test_case.FaceTestCase):
"""Provides abstract Face-layer tests an in-memory implementation.""" """Provides abstract Face-layer tests an in-memory implementation."""
def set_up_implementation( def set_up_implementation(
self, self, name, methods, method_implementations,
name, multi_method_implementation):
methods,
inline_value_in_value_out_methods,
inline_value_in_stream_out_methods,
inline_stream_in_value_out_methods,
inline_stream_in_stream_out_methods,
event_value_in_value_out_methods,
event_value_in_stream_out_methods,
event_stream_in_value_out_methods,
event_stream_in_stream_out_methods,
multi_method):
servicer_pool = logging_pool.pool(_MAXIMUM_POOL_SIZE) servicer_pool = logging_pool.pool(_MAXIMUM_POOL_SIZE)
stub_pool = logging_pool.pool(_MAXIMUM_POOL_SIZE) stub_pool = logging_pool.pool(_MAXIMUM_POOL_SIZE)
servicer = implementations.servicer( servicer = implementations.servicer(
servicer_pool, servicer_pool, method_implementations, multi_method_implementation)
inline_value_in_value_out_methods=inline_value_in_value_out_methods,
inline_value_in_stream_out_methods=inline_value_in_stream_out_methods,
inline_stream_in_value_out_methods=inline_stream_in_value_out_methods,
inline_stream_in_stream_out_methods=inline_stream_in_stream_out_methods,
event_value_in_value_out_methods=event_value_in_value_out_methods,
event_value_in_stream_out_methods=event_value_in_stream_out_methods,
event_stream_in_value_out_methods=event_stream_in_value_out_methods,
event_stream_in_stream_out_methods=event_stream_in_stream_out_methods,
multi_method=multi_method)
linked_pair = base_util.linked_pair(servicer, _TIMEOUT) linked_pair = base_util.linked_pair(servicer, _TIMEOUT)
server = implementations.server() stub = implementations.generic_stub(linked_pair.front, stub_pool)
stub = implementations.stub(linked_pair.front, stub_pool) return stub, (servicer_pool, stub_pool, linked_pair)
return server, stub, (servicer_pool, stub_pool, linked_pair)
def tear_down_implementation(self, memo): def tear_down_implementation(self, memo):
servicer_pool, stub_pool, linked_pair = memo servicer_pool, stub_pool, linked_pair = memo

268
src/python/src/grpc/framework/face/implementations.py
Unescape View File

@ -29,6 +29,8 @@
"""Entry points into the Face layer of RPC Framework.""" """Entry points into the Face layer of RPC Framework."""
from grpc.framework.common import cardinality
from grpc.framework.common import style
from grpc.framework.base import exceptions as _base_exceptions from grpc.framework.base import exceptions as _base_exceptions
from grpc.framework.base import interfaces as base_interfaces from grpc.framework.base import interfaces as base_interfaces
from grpc.framework.face import _calls from grpc.framework.face import _calls
@ -56,7 +58,7 @@ class _BaseServicer(base_interfaces.Servicer):
raise _base_exceptions.NoSuchMethodError() raise _base_exceptions.NoSuchMethodError()
class _UnaryUnarySyncAsync(interfaces.UnaryUnarySyncAsync): class _UnaryUnaryMultiCallable(interfaces.UnaryUnaryMultiCallable):
def __init__(self, front, name): def __init__(self, front, name):
self._front = front self._front = front
@ -66,12 +68,33 @@ class _UnaryUnarySyncAsync(interfaces.UnaryUnarySyncAsync):
return _calls.blocking_value_in_value_out( return _calls.blocking_value_in_value_out(
self._front, self._name, request, timeout, 'unused trace ID') self._front, self._name, request, timeout, 'unused trace ID')
def async(self, request, timeout): def future(self, request, timeout):
return _calls.future_value_in_value_out( return _calls.future_value_in_value_out(
self._front, self._name, request, timeout, 'unused trace ID') self._front, self._name, request, timeout, 'unused trace ID')
def event(self, request, response_callback, abortion_callback, timeout):
return _calls.event_value_in_value_out(
self._front, self._name, request, response_callback, abortion_callback,
timeout, 'unused trace ID')
class _UnaryStreamMultiCallable(interfaces.UnaryStreamMultiCallable):
def __init__(self, front, name):
self._front = front
self._name = name
class _StreamUnarySyncAsync(interfaces.StreamUnarySyncAsync): def __call__(self, request, timeout):
return _calls.inline_value_in_stream_out(
self._front, self._name, request, timeout, 'unused trace ID')
def event(self, request, response_consumer, abortion_callback, timeout):
return _calls.event_value_in_stream_out(
self._front, self._name, request, response_consumer, abortion_callback,
timeout, 'unused trace ID')
class _StreamUnaryMultiCallable(interfaces.StreamUnaryMultiCallable):
def __init__(self, front, name, pool): def __init__(self, front, name, pool):
self._front = front self._front = front
@ -82,18 +105,37 @@ class _StreamUnarySyncAsync(interfaces.StreamUnarySyncAsync):
return _calls.blocking_stream_in_value_out( return _calls.blocking_stream_in_value_out(
self._front, self._name, request_iterator, timeout, 'unused trace ID') self._front, self._name, request_iterator, timeout, 'unused trace ID')
def async(self, request_iterator, timeout): def future(self, request_iterator, timeout):
return _calls.future_stream_in_value_out( return _calls.future_stream_in_value_out(
self._front, self._name, request_iterator, timeout, 'unused trace ID', self._front, self._name, request_iterator, timeout, 'unused trace ID',
self._pool) self._pool)
def event(self, response_callback, abortion_callback, timeout):
return _calls.event_stream_in_value_out(
self._front, self._name, response_callback, abortion_callback, timeout,
'unused trace ID')
class _Server(interfaces.Server):
"""An interfaces.Server implementation."""
class _StreamStreamMultiCallable(interfaces.StreamStreamMultiCallable):
class _Stub(interfaces.Stub): def __init__(self, front, name, pool):
"""An interfaces.Stub implementation.""" self._front = front
self._name = name
self._pool = pool
def __call__(self, request_iterator, timeout):
return _calls.inline_stream_in_stream_out(
self._front, self._name, request_iterator, timeout, 'unused trace ID',
self._pool)
def event(self, response_consumer, abortion_callback, timeout):
return _calls.event_stream_in_stream_out(
self._front, self._name, response_consumer, abortion_callback, timeout,
'unused trace ID')
class _GenericStub(interfaces.GenericStub):
"""An interfaces.GenericStub implementation."""
def __init__(self, front, pool): def __init__(self, front, pool):
self._front = front self._front = front
@ -149,136 +191,128 @@ class _Stub(interfaces.Stub):
self._front, name, response_consumer, abortion_callback, timeout, self._front, name, response_consumer, abortion_callback, timeout,
'unused trace ID') 'unused trace ID')
def unary_unary_sync_async(self, name): def unary_unary_multi_callable(self, name):
return _UnaryUnarySyncAsync(self._front, name) return _UnaryUnaryMultiCallable(self._front, name)
def stream_unary_sync_async(self, name): def unary_stream_multi_callable(self, name):
return _StreamUnarySyncAsync(self._front, name, self._pool) return _UnaryStreamMultiCallable(self._front, name)
def stream_unary_multi_callable(self, name):
def _aggregate_methods( return _StreamUnaryMultiCallable(self._front, name, self._pool)
pool,
inline_value_in_value_out_methods, def stream_stream_multi_callable(self, name):
inline_value_in_stream_out_methods, return _StreamStreamMultiCallable(self._front, name, self._pool)
inline_stream_in_value_out_methods,
inline_stream_in_stream_out_methods,
event_value_in_value_out_methods, class _DynamicStub(interfaces.DynamicStub):
event_value_in_stream_out_methods, """An interfaces.DynamicStub implementation."""
event_stream_in_value_out_methods,
event_stream_in_stream_out_methods): def __init__(self, cardinalities, front, pool):
"""Aggregates methods coded in according to different interfaces.""" self._cardinalities = cardinalities
methods = {} self._front = front
self._pool = pool
def adapt_unpooled_methods(adapted_methods, unadapted_methods, adaptation):
if unadapted_methods is not None: def __getattr__(self, attr):
for name, unadapted_method in unadapted_methods.iteritems(): cardinality = self._cardinalities.get(attr)
adapted_methods[name] = adaptation(unadapted_method) if cardinality is cardinality.Cardinality.UNARY_UNARY:
return _UnaryUnaryMultiCallable(self._front, attr)
def adapt_pooled_methods(adapted_methods, unadapted_methods, adaptation): elif cardinality is cardinality.Cardinality.UNARY_STREAM:
if unadapted_methods is not None: return _UnaryStreamMultiCallable(self._front, attr)
for name, unadapted_method in unadapted_methods.iteritems(): elif cardinality is cardinality.Cardinality.STREAM_UNARY:
adapted_methods[name] = adaptation(unadapted_method, pool) return _StreamUnaryMultiCallable(self._front, attr, self._pool)
elif cardinality is cardinality.Cardinality.STREAM_STREAM:
adapt_unpooled_methods( return _StreamStreamMultiCallable(self._front, attr, self._pool)
methods, inline_value_in_value_out_methods, else:
_service.adapt_inline_value_in_value_out) raise AttributeError('_DynamicStub object has no attribute "%s"!' % attr)
adapt_unpooled_methods(
methods, inline_value_in_stream_out_methods,
_service.adapt_inline_value_in_stream_out) def _adapt_method_implementations(method_implementations, pool):
adapt_pooled_methods( adapted_implementations = {}
methods, inline_stream_in_value_out_methods, for name, method_implementation in method_implementations.iteritems():
_service.adapt_inline_stream_in_value_out) if method_implementation.style is style.Service.INLINE:
adapt_pooled_methods( if method_implementation.cardinality is cardinality.Cardinality.UNARY_UNARY:
methods, inline_stream_in_stream_out_methods, adapted_implementations[name] = _service.adapt_inline_value_in_value_out(
_service.adapt_inline_stream_in_stream_out) method_implementation.unary_unary_inline)
adapt_unpooled_methods( elif method_implementation.cardinality is cardinality.Cardinality.UNARY_STREAM:
methods, event_value_in_value_out_methods, adapted_implementations[name] = _service.adapt_inline_value_in_stream_out(
_service.adapt_event_value_in_value_out) method_implementation.unary_stream_inline)
adapt_unpooled_methods( elif method_implementation.cardinality is cardinality.Cardinality.STREAM_UNARY:
methods, event_value_in_stream_out_methods, adapted_implementations[name] = _service.adapt_inline_stream_in_value_out(
_service.adapt_event_value_in_stream_out) method_implementation.stream_unary_inline, pool)
adapt_unpooled_methods( elif method_implementation.cardinality is cardinality.Cardinality.STREAM_STREAM:
methods, event_stream_in_value_out_methods, adapted_implementations[name] = _service.adapt_inline_stream_in_stream_out(
_service.adapt_event_stream_in_value_out) method_implementation.stream_stream_inline, pool)
adapt_unpooled_methods( elif method_implementation.style is style.Service.EVENT:
methods, event_stream_in_stream_out_methods, if method_implementation.cardinality is cardinality.Cardinality.UNARY_UNARY:
_service.adapt_event_stream_in_stream_out) adapted_implementations[name] = _service.adapt_event_value_in_value_out(
method_implementation.unary_unary_event)
return methods elif method_implementation.cardinality is cardinality.Cardinality.UNARY_STREAM:
adapted_implementations[name] = _service.adapt_event_value_in_stream_out(
method_implementation.unary_stream_event)
def servicer( elif method_implementation.cardinality is cardinality.Cardinality.STREAM_UNARY:
pool, adapted_implementations[name] = _service.adapt_event_stream_in_value_out(
inline_value_in_value_out_methods=None, method_implementation.stream_unary_event)
inline_value_in_stream_out_methods=None, elif method_implementation.cardinality is cardinality.Cardinality.STREAM_STREAM:
inline_stream_in_value_out_methods=None, adapted_implementations[name] = _service.adapt_event_stream_in_stream_out(
inline_stream_in_stream_out_methods=None, method_implementation.stream_stream_event)
event_value_in_value_out_methods=None, return adapted_implementations
event_value_in_stream_out_methods=None,
event_stream_in_value_out_methods=None,
event_stream_in_stream_out_methods=None, def servicer(pool, method_implementations, multi_method_implementation):
multi_method=None):
"""Creates a base_interfaces.Servicer. """Creates a base_interfaces.Servicer.
The key sets of the passed dictionaries must be disjoint. It is guaranteed It is guaranteed that any passed interfaces.MultiMethodImplementation will
that any passed MultiMethod implementation will only be called to service an only be called to service an RPC if there is no
RPC if the RPC method name is not present in the key sets of the passed interfaces.MethodImplementation for the RPC method in the passed
dictionaries. method_implementations dictionary.
Args: Args:
pool: A thread pool. pool: A thread pool.
inline_value_in_value_out_methods: A dictionary mapping method names to method_implementations: A dictionary from RPC method name to
interfaces.InlineValueInValueOutMethod implementations. interfaces.MethodImplementation object to be used to service the named
inline_value_in_stream_out_methods: A dictionary mapping method names to RPC method.
interfaces.InlineValueInStreamOutMethod implementations. multi_method_implementation: An interfaces.MultiMethodImplementation to be
inline_stream_in_value_out_methods: A dictionary mapping method names to used to service any RPCs not serviced by the
interfaces.InlineStreamInValueOutMethod implementations. interfaces.MethodImplementations given in the method_implementations
inline_stream_in_stream_out_methods: A dictionary mapping method names to dictionary, or None.
interfaces.InlineStreamInStreamOutMethod implementations.
event_value_in_value_out_methods: A dictionary mapping method names to
interfaces.EventValueInValueOutMethod implementations.
event_value_in_stream_out_methods: A dictionary mapping method names to
interfaces.EventValueInStreamOutMethod implementations.
event_stream_in_value_out_methods: A dictionary mapping method names to
interfaces.EventStreamInValueOutMethod implementations.
event_stream_in_stream_out_methods: A dictionary mapping method names to
interfaces.EventStreamInStreamOutMethod implementations.
multi_method: An implementation of interfaces.MultiMethod.
Returns: Returns:
A base_interfaces.Servicer that services RPCs via the given implementations. A base_interfaces.Servicer that services RPCs via the given implementations.
""" """
methods = _aggregate_methods( adapted_implementations = _adapt_method_implementations(
pool, method_implementations, pool)
inline_value_in_value_out_methods, return _BaseServicer(adapted_implementations, multi_method_implementation)
inline_value_in_stream_out_methods,
inline_stream_in_value_out_methods,
inline_stream_in_stream_out_methods,
event_value_in_value_out_methods,
event_value_in_stream_out_methods,
event_stream_in_value_out_methods,
event_stream_in_stream_out_methods)
return _BaseServicer(methods, multi_method)
def generic_stub(front, pool):
"""Creates an interfaces.GenericStub.
def server(): Args:
"""Creates an interfaces.Server. front: A base_interfaces.Front.
pool: A futures.ThreadPoolExecutor.
Returns: Returns:
An interfaces.Server. An interfaces.GenericStub that performs RPCs via the given
base_interfaces.Front.
""" """
return _Server() return _GenericStub(front, pool)
def stub(front, pool): def dynamic_stub(cardinalities, front, pool, prefix):
"""Creates an interfaces.Stub. """Creates an interfaces.DynamicStub.
Args: Args:
cardinalities: A dict from RPC method name to cardinality.Cardinality
value identifying the cardinality of every RPC method to be supported by
the created interfaces.DynamicStub.
front: A base_interfaces.Front. front: A base_interfaces.Front.
pool: A futures.ThreadPoolExecutor. pool: A futures.ThreadPoolExecutor.
prefix: A string to prepend when mapping requested attribute name to RPC
method name during attribute access on the created
interfaces.DynamicStub.
Returns: Returns:
An interfaces.Stub that performs RPCs via the given base_interfaces.Front. An interfaces.DynamicStub that performs RPCs via the given
base_interfaces.Front.
""" """
return _Stub(front, pool) return _DynamicStub(cardinalities, front, pool, prefix)

462
src/python/src/grpc/framework/face/interfaces.py
Unescape View File

@ -32,11 +32,24 @@
import abc import abc
import enum import enum
# exceptions, abandonment, and future are referenced from specification in this # cardinality, style, exceptions, abandonment, future, and stream are
# module. # referenced from specification in this module.
from grpc.framework.common import cardinality # pylint: disable=unused-import
from grpc.framework.common import style # pylint: disable=unused-import
from grpc.framework.face import exceptions # pylint: disable=unused-import from grpc.framework.face import exceptions # pylint: disable=unused-import
from grpc.framework.foundation import abandonment # pylint: disable=unused-import from grpc.framework.foundation import abandonment # pylint: disable=unused-import
from grpc.framework.foundation import future # pylint: disable=unused-import from grpc.framework.foundation import future # pylint: disable=unused-import
from grpc.framework.foundation import stream # pylint: disable=unused-import
@enum.unique
class Abortion(enum.Enum):
"""Categories of RPC abortion."""
CANCELLED = 'cancelled'
EXPIRED = 'expired'
NETWORK_FAILURE = 'network failure'
SERVICED_FAILURE = 'serviced failure'
SERVICER_FAILURE = 'servicer failure'
class CancellableIterator(object): class CancellableIterator(object):
@ -59,69 +72,61 @@ class CancellableIterator(object):
raise NotImplementedError() raise NotImplementedError()
class UnaryUnarySyncAsync(object): class RpcContext(object):
"""Affords invoking a unary-unary RPC synchronously or asynchronously. """Provides RPC-related information and action."""
Values implementing this interface are directly callable and present an
"async" method. Both calls take a request value and a numeric timeout.
Direct invocation of a value of this type invokes its associated RPC and
blocks until the RPC's response is available. Calling the "async" method
of a value of this type invokes its associated RPC and immediately returns a
future.Future bound to the asynchronous execution of the RPC.
"""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def __call__(self, request, timeout): def is_active(self):
"""Synchronously invokes the underlying RPC. """Describes whether the RPC is active or has terminated."""
raise NotImplementedError()
Args: @abc.abstractmethod
request: The request value for the RPC. def time_remaining(self):
timeout: A duration of time in seconds to allow for the RPC. """Describes the length of allowed time remaining for the RPC.
Returns: Returns:
The response value for the RPC. A nonnegative float indicating the length of allowed time in seconds
remaining for the RPC to complete before it is considered to have timed
Raises: out.
exceptions.RpcError: Indicating that the RPC was aborted.
""" """
raise NotImplementedError() raise NotImplementedError()
@abc.abstractmethod @abc.abstractmethod
def async(self, request, timeout): def add_abortion_callback(self, abortion_callback):
"""Asynchronously invokes the underlying RPC. """Registers a callback to be called if the RPC is aborted.
Args: Args:
request: The request value for the RPC. abortion_callback: A callable to be called and passed an Abortion value
timeout: A duration of time in seconds to allow for the RPC. in the event of RPC abortion.
Returns:
A future.Future representing the RPC. In the event of RPC completion, the
returned Future's result value will be the response value of the RPC.
In the event of RPC abortion, the returned Future's exception value
will be an exceptions.RpcError.
""" """
raise NotImplementedError() raise NotImplementedError()
class StreamUnarySyncAsync(object): class Call(object):
"""Affords invoking a stream-unary RPC synchronously or asynchronously. """Invocation-side representation of an RPC.
Values implementing this interface are directly callable and present an Attributes:
"async" method. Both calls take an iterator of request values and a numeric context: An RpcContext affording information about the RPC.
timeout. Direct invocation of a value of this type invokes its associated RPC
and blocks until the RPC's response is available. Calling the "async" method
of a value of this type invokes its associated RPC and immediately returns a
future.Future bound to the asynchronous execution of the RPC.
""" """
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def __call__(self, request_iterator, timeout): def cancel(self):
"""Requests cancellation of the RPC."""
raise NotImplementedError()
class UnaryUnaryMultiCallable(object):
"""Affords invoking a unary-unary RPC in any call style."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def __call__(self, request, timeout):
"""Synchronously invokes the underlying RPC. """Synchronously invokes the underlying RPC.
Args: Args:
request_iterator: An iterator that yields request values for the RPC. request: The request value for the RPC.
timeout: A duration of time in seconds to allow for the RPC. timeout: A duration of time in seconds to allow for the RPC.
Returns: Returns:
@ -133,11 +138,11 @@ class StreamUnarySyncAsync(object):
raise NotImplementedError() raise NotImplementedError()
@abc.abstractmethod @abc.abstractmethod
def async(self, request, timeout): def future(self, request, timeout):
"""Asynchronously invokes the underlying RPC. """Asynchronously invokes the underlying RPC.
Args: Args:
request_iterator: An iterator that yields request values for the RPC. request: The request value for the RPC.
timeout: A duration of time in seconds to allow for the RPC. timeout: A duration of time in seconds to allow for the RPC.
Returns: Returns:
@ -148,248 +153,204 @@ class StreamUnarySyncAsync(object):
""" """
raise NotImplementedError() raise NotImplementedError()
@enum.unique
class Abortion(enum.Enum):
"""Categories of RPC abortion."""
CANCELLED = 'cancelled'
EXPIRED = 'expired'
NETWORK_FAILURE = 'network failure'
SERVICED_FAILURE = 'serviced failure'
SERVICER_FAILURE = 'servicer failure'
class RpcContext(object):
"""Provides RPC-related information and action."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def is_active(self):
"""Describes whether the RPC is active or has terminated."""
raise NotImplementedError()
@abc.abstractmethod
def time_remaining(self):
"""Describes the length of allowed time remaining for the RPC.
Returns:
A nonnegative float indicating the length of allowed time in seconds
remaining for the RPC to complete before it is considered to have timed
out.
"""
raise NotImplementedError()
@abc.abstractmethod @abc.abstractmethod
def add_abortion_callback(self, abortion_callback): def event(self, request, response_callback, abortion_callback, timeout):
"""Registers a callback to be called if the RPC is aborted. """Asynchronously invokes the underlying RPC.
Args: Args:
abortion_callback: A callable to be called and passed an Abortion value request: The request value for the RPC.
response_callback: A callback to be called to accept the restponse value
of the RPC.
abortion_callback: A callback to be called and passed an Abortion value
in the event of RPC abortion. in the event of RPC abortion.
""" timeout: A duration of time in seconds to allow for the RPC.
raise NotImplementedError()
class InlineValueInValueOutMethod(object):
"""A type for inline unary-request-unary-response RPC methods."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def service(self, request, context):
"""Services an RPC that accepts one value and produces one value.
Args:
request: The single request value for the RPC.
context: An RpcContext object.
Returns: Returns:
The single response value for the RPC. A Call object for the RPC.
Raises:
abandonment.Abandoned: If no response is necessary because the RPC has
been aborted.
""" """
raise NotImplementedError() raise NotImplementedError()
class InlineValueInStreamOutMethod(object): class UnaryStreamMultiCallable(object):
"""A type for inline unary-request-stream-response RPC methods.""" """Affords invoking a unary-stream RPC in any call style."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def service(self, request, context): def __call__(self, request, timeout):
"""Services an RPC that accepts one value and produces a stream of values. """Synchronously invokes the underlying RPC.
Args: Args:
request: The single request value for the RPC. request: The request value for the RPC.
context: An RpcContext object. timeout: A duration of time in seconds to allow for the RPC.
Yields:
The values that comprise the response stream of the RPC.
Raises: Returns:
abandonment.Abandoned: If completing the response stream is not necessary A CancellableIterator that yields the response values of the RPC and
because the RPC has been aborted. affords RPC cancellation. Drawing response values from the returned
CancellableIterator may raise exceptions.RpcError indicating abortion
of the RPC.
""" """
raise NotImplementedError() raise NotImplementedError()
class InlineStreamInValueOutMethod(object):
"""A type for inline stream-request-unary-response RPC methods."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def service(self, request_iterator, context): def event(self, request, response_consumer, abortion_callback, timeout):
"""Services an RPC that accepts a stream of values and produces one value. """Asynchronously invokes the underlying RPC.
Args: Args:
request_iterator: An iterator that yields the request values of the RPC. request: The request value for the RPC.
Drawing values from this iterator may also raise exceptions.RpcError to response_consumer: A stream.Consumer to be called to accept the restponse
indicate abortion of the RPC. values of the RPC.
context: An RpcContext object. abortion_callback: A callback to be called and passed an Abortion value
in the event of RPC abortion.
Yields: timeout: A duration of time in seconds to allow for the RPC.
The values that comprise the response stream of the RPC.
Raises: Returns:
abandonment.Abandoned: If no response is necessary because the RPC has A Call object for the RPC.
been aborted.
exceptions.RpcError: Implementations of this method must not deliberately
raise exceptions.RpcError but may allow such errors raised by the
request_iterator passed to them to propagate through their bodies
uncaught.
""" """
raise NotImplementedError() raise NotImplementedError()
class InlineStreamInStreamOutMethod(object): class StreamUnaryMultiCallable(object):
"""A type for inline stream-request-stream-response RPC methods.""" """Affords invoking a stream-unary RPC in any call style."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def service(self, request_iterator, context): def __call__(self, request_iterator, timeout):
"""Services an RPC that accepts and produces streams of values. """Synchronously invokes the underlying RPC.
Args: Args:
request_iterator: An iterator that yields the request values of the RPC. request_iterator: An iterator that yields request values for the RPC.
Drawing values from this iterator may also raise exceptions.RpcError to timeout: A duration of time in seconds to allow for the RPC.
indicate abortion of the RPC.
context: An RpcContext object.
Yields: Returns:
The values that comprise the response stream of the RPC. The response value for the RPC.
Raises: Raises:
abandonment.Abandoned: If completing the response stream is not necessary exceptions.RpcError: Indicating that the RPC was aborted.
because the RPC has been aborted.
exceptions.RpcError: Implementations of this method must not deliberately
raise exceptions.RpcError but may allow such errors raised by the
request_iterator passed to them to propagate through their bodies
uncaught.
""" """
raise NotImplementedError() raise NotImplementedError()
class EventValueInValueOutMethod(object):
"""A type for event-driven unary-request-unary-response RPC methods."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def service(self, request, response_callback, context): def future(self, request_iterator, timeout):
"""Services an RPC that accepts one value and produces one value. """Asynchronously invokes the underlying RPC.
Args: Args:
request: The single request value for the RPC. request_iterator: An iterator that yields request values for the RPC.
response_callback: A callback to be called to accept the response value of timeout: A duration of time in seconds to allow for the RPC.
the RPC.
context: An RpcContext object.
Raises: Returns:
abandonment.Abandoned: May or may not be raised when the RPC has been A future.Future representing the RPC. In the event of RPC completion, the
aborted. returned Future's result value will be the response value of the RPC.
In the event of RPC abortion, the returned Future's exception value
will be an exceptions.RpcError.
""" """
raise NotImplementedError() raise NotImplementedError()
class EventValueInStreamOutMethod(object):
"""A type for event-driven unary-request-stream-response RPC methods."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def service(self, request, response_consumer, context): def event(self, response_callback, abortion_callback, timeout):
"""Services an RPC that accepts one value and produces a stream of values. """Asynchronously invokes the underlying RPC.
Args: Args:
request: The single request value for the RPC. request: The request value for the RPC.
response_consumer: A stream.Consumer to be called to accept the response response_callback: A callback to be called to accept the restponse value
values of the RPC. of the RPC.
context: An RpcContext object. abortion_callback: A callback to be called and passed an Abortion value
in the event of RPC abortion.
timeout: A duration of time in seconds to allow for the RPC.
Raises: Returns:
abandonment.Abandoned: May or may not be raised when the RPC has been A pair of a Call object for the RPC and a stream.Consumer to which the
aborted. request values of the RPC should be passed.
""" """
raise NotImplementedError() raise NotImplementedError()
class EventStreamInValueOutMethod(object): class StreamStreamMultiCallable(object):
"""A type for event-driven stream-request-unary-response RPC methods.""" """Affords invoking a stream-stream RPC in any call style."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def service(self, response_callback, context): def __call__(self, request_iterator, timeout):
"""Services an RPC that accepts a stream of values and produces one value. """Synchronously invokes the underlying RPC.
Args: Args:
response_callback: A callback to be called to accept the response value of request_iterator: An iterator that yields request values for the RPC.
the RPC. timeout: A duration of time in seconds to allow for the RPC.
context: An RpcContext object.
Returns: Returns:
A stream.Consumer with which to accept the request values of the RPC. The A CancellableIterator that yields the response values of the RPC and
consumer returned from this method may or may not be invoked to affords RPC cancellation. Drawing response values from the returned
completion: in the case of RPC abortion, RPC Framework will simply stop CancellableIterator may raise exceptions.RpcError indicating abortion
passing values to this object. Implementations must not assume that this of the RPC.
object will be called to completion of the request stream or even called
at all.
Raises:
abandonment.Abandoned: May or may not be raised when the RPC has been
aborted.
""" """
raise NotImplementedError() raise NotImplementedError()
class EventStreamInStreamOutMethod(object):
"""A type for event-driven stream-request-stream-response RPC methods."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def service(self, response_consumer, context): def event(self, response_consumer, abortion_callback, timeout):
"""Services an RPC that accepts and produces streams of values. """Asynchronously invokes the underlying RPC.
Args: l Args:
response_consumer: A stream.Consumer to be called to accept the response response_consumer: A stream.Consumer to be called to accept the restponse
values of the RPC. values of the RPC.
context: An RpcContext object. abortion_callback: A callback to be called and passed an Abortion value
in the event of RPC abortion.
timeout: A duration of time in seconds to allow for the RPC.
Returns: Returns:
A stream.Consumer with which to accept the request values of the RPC. The A pair of a Call object for the RPC and a stream.Consumer to which the
consumer returned from this method may or may not be invoked to request values of the RPC should be passed.
completion: in the case of RPC abortion, RPC Framework will simply stop
passing values to this object. Implementations must not assume that this
object will be called to completion of the request stream or even called
at all.
Raises:
abandonment.Abandoned: May or may not be raised when the RPC has been
aborted.
""" """
raise NotImplementedError() raise NotImplementedError()
class MultiMethod(object): class MethodImplementation(object):
"""A sum type that describes an RPC method implementation.
Attributes:
cardinality: A cardinality.Cardinality value.
style: A style.Service value.
unary_unary_inline: The implementation of the RPC method as a callable
value that takes a request value and an RpcContext object and returns a
response value. Only non-None if cardinality is
cardinality.Cardinality.UNARY_UNARY and style is style.Service.INLINE.
unary_stream_inline: The implementation of the RPC method as a callable
value that takes a request value and an RpcContext object and returns an
iterator of response values. Only non-None if cardinality is
cardinality.Cardinality.UNARY_STREAM and style is style.Service.INLINE.
stream_unary_inline: The implementation of the RPC method as a callable
value that takes an iterator of request values and an RpcContext object
and returns a response value. Only non-None if cardinality is
cardinality.Cardinality.STREAM_UNARY and style is style.Service.INLINE.
stream_stream_inline: The implementation of the RPC method as a callable
value that takes an iterator of request values and an RpcContext object
and returns an iterator of response values. Only non-None if cardinality
is cardinality.Cardinality.STREAM_STREAM and style is
style.Service.INLINE.
unary_unary_event: The implementation of the RPC method as a callable value
that takes a request value, a response callback to which to pass the
response value of the RPC, and an RpcContext. Only non-None if
cardinality is cardinality.Cardinality.UNARY_UNARY and style is
style.Service.EVENT.
unary_stream_event: The implementation of the RPC method as a callable
value that takes a request value, a stream.Consumer to which to pass the
the response values of the RPC, and an RpcContext. Only non-None if
cardinality is cardinality.Cardinality.UNARY_STREAM and style is
style.Service.EVENT.
stream_unary_event: The implementation of the RPC method as a callable
value that takes a response callback to which to pass the response value
of the RPC and an RpcContext and returns a stream.Consumer to which the
request values of the RPC should be passed. Only non-None if cardinality
is cardinality.Cardinality.STREAM_UNARY and style is style.Service.EVENT.
stream_stream_event: The implementation of the RPC method as a callable
value that takes a stream.Consumer to which to pass the response values
of the RPC and an RpcContext and returns a stream.Consumer to which the
request values of the RPC should be passed. Only non-None if cardinality
is cardinality.Cardinality.STREAM_STREAM and style is
style.Service.EVENT.
"""
__metaclass__ = abc.ABCMeta
class MultiMethodImplementation(object):
"""A general type able to service many RPC methods.""" """A general type able to service many RPC methods."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@ -420,26 +381,7 @@ class MultiMethod(object):
raise NotImplementedError() raise NotImplementedError()
class Server(object): class GenericStub(object):
"""Specification of a running server that services RPCs."""
__metaclass__ = abc.ABCMeta
class Call(object):
"""Invocation-side representation of an RPC.
Attributes:
context: An RpcContext affording information about the RPC.
"""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def cancel(self):
"""Requests cancellation of the RPC."""
raise NotImplementedError()
class Stub(object):
"""Affords RPC methods to callers.""" """Affords RPC methods to callers."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@ -632,25 +574,67 @@ class Stub(object):
raise NotImplementedError() raise NotImplementedError()
@abc.abstractmethod @abc.abstractmethod
def unary_unary_sync_async(self, name): def unary_unary_multi_callable(self, name):
"""Creates a UnaryUnarySyncAsync value for a unary-unary RPC method. """Creates a UnaryUnaryMultiCallable for a unary-unary RPC method.
Args:
name: The RPC method name.
Returns:
A UnaryUnaryMultiCallable value for the named unary-unary RPC method.
"""
raise NotImplementedError()
@abc.abstractmethod
def unary_stream_multi_callable(self, name):
"""Creates a UnaryStreamMultiCallable for a unary-stream RPC method.
Args: Args:
name: The RPC method name. name: The RPC method name.
Returns: Returns:
A UnaryUnarySyncAsync value for the named unary-unary RPC method. A UnaryStreamMultiCallable value for the name unary-stream RPC method.
""" """
raise NotImplementedError() raise NotImplementedError()
@abc.abstractmethod @abc.abstractmethod
def stream_unary_sync_async(self, name): def stream_unary_multi_callable(self, name):
"""Creates a StreamUnarySyncAsync value for a stream-unary RPC method. """Creates a StreamUnaryMultiCallable for a stream-unary RPC method.
Args: Args:
name: The RPC method name. name: The RPC method name.
Returns: Returns:
A StreamUnarySyncAsync value for the named stream-unary RPC method. A StreamUnaryMultiCallable value for the named stream-unary RPC method.
""" """
raise NotImplementedError() raise NotImplementedError()
@abc.abstractmethod
def stream_stream_multi_callable(self, name):
"""Creates a StreamStreamMultiCallable for a stream-stream RPC method.
Args:
name: The RPC method name.
Returns:
A StreamStreamMultiCallable value for the named stream-stream RPC method.
"""
raise NotImplementedError()
class DynamicStub(object):
"""A stub with RPC-method-bound multi-callable attributes.
Instances of this type responsd to attribute access as follows: if the
requested attribute is the name of a unary-unary RPC method, the value of the
attribute will be a UnaryUnaryMultiCallable with which to invoke the RPC
method; if the requested attribute is the name of a unary-stream RPC method,
the value of the attribute will be a UnaryStreamMultiCallable with which to
invoke the RPC method; if the requested attribute is the name of a
stream-unary RPC method, the value of the attribute will be a
StreamUnaryMultiCallable with which to invoke the RPC method; and if the
requested attribute is the name of a stream-stream RPC method, the value of
the attribute will be a StreamStreamMultiCallable with which to invoke the
RPC method.
"""
__metaclass__ = abc.ABCMeta

16
src/python/src/grpc/framework/face/testing/blocking_invocation_inline_service_test_case.py
Unescape View File

@ -61,13 +61,9 @@ class BlockingInvocationInlineServiceTestCase(
self.digest = digest.digest( self.digest = digest.digest(
stock_service.STOCK_TEST_SERVICE, self.control, None) stock_service.STOCK_TEST_SERVICE, self.control, None)
self.server, self.stub, self.memo = self.set_up_implementation( self.stub, self.memo = self.set_up_implementation(
self.digest.name, self.digest.methods, self.digest.name, self.digest.methods,
self.digest.inline_unary_unary_methods, self.digest.inline_method_implementations, None)
self.digest.inline_unary_stream_methods,
self.digest.inline_stream_unary_methods,
self.digest.inline_stream_stream_methods,
{}, {}, {}, {}, None)
def tearDown(self): def tearDown(self):
"""See unittest.TestCase.tearDown for full specification. """See unittest.TestCase.tearDown for full specification.
@ -147,8 +143,8 @@ class BlockingInvocationInlineServiceTestCase(
with self.control.pause(), self.assertRaises( with self.control.pause(), self.assertRaises(
exceptions.ExpirationError): exceptions.ExpirationError):
sync_async = self.stub.unary_unary_sync_async(name) multi_callable = self.stub.unary_unary_multi_callable(name)
sync_async(request, _TIMEOUT) multi_callable(request, _TIMEOUT)
def testExpiredUnaryRequestStreamResponse(self): def testExpiredUnaryRequestStreamResponse(self):
for name, test_messages_sequence in ( for name, test_messages_sequence in (
@ -170,8 +166,8 @@ class BlockingInvocationInlineServiceTestCase(
with self.control.pause(), self.assertRaises( with self.control.pause(), self.assertRaises(
exceptions.ExpirationError): exceptions.ExpirationError):
sync_async = self.stub.stream_unary_sync_async(name) multi_callable = self.stub.stream_unary_multi_callable(name)
sync_async(iter(requests), _TIMEOUT) multi_callable(iter(requests), _TIMEOUT)
def testExpiredStreamRequestStreamResponse(self): def testExpiredStreamRequestStreamResponse(self):
for name, test_messages_sequence in ( for name, test_messages_sequence in (

122
src/python/src/grpc/framework/face/testing/digest.py
Unescape View File

@ -34,6 +34,8 @@ import threading
# testing_control, interfaces, and testing_service are referenced from # testing_control, interfaces, and testing_service are referenced from
# specification in this module. # specification in this module.
from grpc.framework.common import cardinality
from grpc.framework.common import style
from grpc.framework.face import exceptions from grpc.framework.face import exceptions
from grpc.framework.face import interfaces as face_interfaces from grpc.framework.face import interfaces as face_interfaces
from grpc.framework.face.testing import control as testing_control # pylint: disable=unused-import from grpc.framework.face.testing import control as testing_control # pylint: disable=unused-import
@ -50,15 +52,9 @@ class TestServiceDigest(
'TestServiceDigest', 'TestServiceDigest',
['name', ['name',
'methods', 'methods',
'inline_unary_unary_methods', 'inline_method_implementations',
'inline_unary_stream_methods', 'event_method_implementations',
'inline_stream_unary_methods', 'multi_method_implementation',
'inline_stream_stream_methods',
'event_unary_unary_methods',
'event_unary_stream_methods',
'event_stream_unary_methods',
'event_stream_stream_methods',
'multi_method',
'unary_unary_messages_sequences', 'unary_unary_messages_sequences',
'unary_stream_messages_sequences', 'unary_stream_messages_sequences',
'stream_unary_messages_sequences', 'stream_unary_messages_sequences',
@ -69,32 +65,14 @@ class TestServiceDigest(
name: The RPC service name to be used in the test. name: The RPC service name to be used in the test.
methods: A sequence of interfaces.Method objects describing the RPC methods: A sequence of interfaces.Method objects describing the RPC
methods that will be called during the test. methods that will be called during the test.
inline_unary_unary_methods: A dict from method name to inline_method_implementations: A dict from RPC method name to
face_interfaces.InlineValueInValueOutMethod object to be used in tests of face_interfaces.MethodImplementation object to be used in tests of
in-line calls to behaviors under test. in-line calls to behaviors under test.
inline_unary_stream_methods: A dict from method name to event_method_implementations: A dict from RPC method name to
face_interfaces.InlineValueInStreamOutMethod object to be used in tests of face_interfaces.MethodImplementation object to be used in tests of
in-line calls to behaviors under test.
inline_stream_unary_methods: A dict from method name to
face_interfaces.InlineStreamInValueOutMethod object to be used in tests of
in-line calls to behaviors under test.
inline_stream_stream_methods: A dict from method name to
face_interfaces.InlineStreamInStreamOutMethod object to be used in tests
of in-line calls to behaviors under test.
event_unary_unary_methods: A dict from method name to
face_interfaces.EventValueInValueOutMethod object to be used in tests of
event-driven calls to behaviors under test.
event_unary_stream_methods: A dict from method name to
face_interfaces.EventValueInStreamOutMethod object to be used in tests of
event-driven calls to behaviors under test.
event_stream_unary_methods: A dict from method name to
face_interfaces.EventStreamInValueOutMethod object to be used in tests of
event-driven calls to behaviors under test. event-driven calls to behaviors under test.
event_stream_stream_methods: A dict from method name to multi_method_implementation: A face_interfaces.MultiMethodImplementation to
face_interfaces.EventStreamInStreamOutMethod object to be used in tests of be used in tests of generic calls to behaviors under test.
event-driven calls to behaviors under test.
multi_method: A face_interfaces.MultiMethod to be used in tests of generic
calls to behaviors under test.
unary_unary_messages_sequences: A dict from method name to sequence of unary_unary_messages_sequences: A dict from method name to sequence of
service.UnaryUnaryTestMessages objects to be used to test the method service.UnaryUnaryTestMessages objects to be used to test the method
with the given name. with the given name.
@ -130,27 +108,33 @@ class _BufferingConsumer(stream.Consumer):
self.terminated = True self.terminated = True
class _InlineUnaryUnaryMethod(face_interfaces.InlineValueInValueOutMethod): class _InlineUnaryUnaryMethod(face_interfaces.MethodImplementation):
def __init__(self, unary_unary_test_method, control): def __init__(self, unary_unary_test_method, control):
self._test_method = unary_unary_test_method self._test_method = unary_unary_test_method
self._control = control self._control = control
def service(self, request, context): self.cardinality = cardinality.Cardinality.UNARY_UNARY
self.style = style.Service.INLINE
def unary_unary_inline(self, request, context):
response_list = [] response_list = []
self._test_method.service( self._test_method.service(
request, response_list.append, context, self._control) request, response_list.append, context, self._control)
return response_list.pop(0) return response_list.pop(0)
class _EventUnaryUnaryMethod(face_interfaces.EventValueInValueOutMethod): class _EventUnaryUnaryMethod(face_interfaces.MethodImplementation):
def __init__(self, unary_unary_test_method, control, pool): def __init__(self, unary_unary_test_method, control, pool):
self._test_method = unary_unary_test_method self._test_method = unary_unary_test_method
self._control = control self._control = control
self._pool = pool self._pool = pool
def service(self, request, response_callback, context): self.cardinality = cardinality.Cardinality.UNARY_UNARY
self.style = style.Service.EVENT
def unary_unary_event(self, request, response_callback, context):
if self._pool is None: if self._pool is None:
self._test_method.service( self._test_method.service(
request, response_callback, context, self._control) request, response_callback, context, self._control)
@ -160,13 +144,16 @@ class _EventUnaryUnaryMethod(face_interfaces.EventValueInValueOutMethod):
self._control) self._control)
class _InlineUnaryStreamMethod(face_interfaces.InlineValueInStreamOutMethod): class _InlineUnaryStreamMethod(face_interfaces.MethodImplementation):
def __init__(self, unary_stream_test_method, control): def __init__(self, unary_stream_test_method, control):
self._test_method = unary_stream_test_method self._test_method = unary_stream_test_method
self._control = control self._control = control
def service(self, request, context): self.cardinality = cardinality.Cardinality.UNARY_STREAM
self.style = style.Service.INLINE
def unary_stream_inline(self, request, context):
response_consumer = _BufferingConsumer() response_consumer = _BufferingConsumer()
self._test_method.service( self._test_method.service(
request, response_consumer, context, self._control) request, response_consumer, context, self._control)
@ -174,14 +161,17 @@ class _InlineUnaryStreamMethod(face_interfaces.InlineValueInStreamOutMethod):
yield response yield response
class _EventUnaryStreamMethod(face_interfaces.EventValueInStreamOutMethod): class _EventUnaryStreamMethod(face_interfaces.MethodImplementation):
def __init__(self, unary_stream_test_method, control, pool): def __init__(self, unary_stream_test_method, control, pool):
self._test_method = unary_stream_test_method self._test_method = unary_stream_test_method
self._control = control self._control = control
self._pool = pool self._pool = pool
def service(self, request, response_consumer, context): self.cardinality = cardinality.Cardinality.UNARY_STREAM
self.style = style.Service.EVENT
def unary_stream_event(self, request, response_consumer, context):
if self._pool is None: if self._pool is None:
self._test_method.service( self._test_method.service(
request, response_consumer, context, self._control) request, response_consumer, context, self._control)
@ -191,13 +181,16 @@ class _EventUnaryStreamMethod(face_interfaces.EventValueInStreamOutMethod):
self._control) self._control)
class _InlineStreamUnaryMethod(face_interfaces.InlineStreamInValueOutMethod): class _InlineStreamUnaryMethod(face_interfaces.MethodImplementation):
def __init__(self, stream_unary_test_method, control): def __init__(self, stream_unary_test_method, control):
self._test_method = stream_unary_test_method self._test_method = stream_unary_test_method
self._control = control self._control = control
def service(self, request_iterator, context): self.cardinality = cardinality.Cardinality.STREAM_UNARY
self.style = style.Service.INLINE
def stream_unary_inline(self, request_iterator, context):
response_list = [] response_list = []
request_consumer = self._test_method.service( request_consumer = self._test_method.service(
response_list.append, context, self._control) response_list.append, context, self._control)
@ -207,14 +200,17 @@ class _InlineStreamUnaryMethod(face_interfaces.InlineStreamInValueOutMethod):
return response_list.pop(0) return response_list.pop(0)
class _EventStreamUnaryMethod(face_interfaces.EventStreamInValueOutMethod): class _EventStreamUnaryMethod(face_interfaces.MethodImplementation):
def __init__(self, stream_unary_test_method, control, pool): def __init__(self, stream_unary_test_method, control, pool):
self._test_method = stream_unary_test_method self._test_method = stream_unary_test_method
self._control = control self._control = control
self._pool = pool self._pool = pool
def service(self, response_callback, context): self.cardinality = cardinality.Cardinality.STREAM_UNARY
self.style = style.Service.EVENT
def stream_unary_event(self, response_callback, context):
request_consumer = self._test_method.service( request_consumer = self._test_method.service(
response_callback, context, self._control) response_callback, context, self._control)
if self._pool is None: if self._pool is None:
@ -223,13 +219,16 @@ class _EventStreamUnaryMethod(face_interfaces.EventStreamInValueOutMethod):
return stream_util.ThreadSwitchingConsumer(request_consumer, self._pool) return stream_util.ThreadSwitchingConsumer(request_consumer, self._pool)
class _InlineStreamStreamMethod(face_interfaces.InlineStreamInStreamOutMethod): class _InlineStreamStreamMethod(face_interfaces.MethodImplementation):
def __init__(self, stream_stream_test_method, control): def __init__(self, stream_stream_test_method, control):
self._test_method = stream_stream_test_method self._test_method = stream_stream_test_method
self._control = control self._control = control
def service(self, request_iterator, context): self.cardinality = cardinality.Cardinality.STREAM_STREAM
self.style = style.Service.INLINE
def stream_stream_inline(self, request_iterator, context):
response_consumer = _BufferingConsumer() response_consumer = _BufferingConsumer()
request_consumer = self._test_method.service( request_consumer = self._test_method.service(
response_consumer, context, self._control) response_consumer, context, self._control)
@ -241,14 +240,17 @@ class _InlineStreamStreamMethod(face_interfaces.InlineStreamInStreamOutMethod):
response_consumer.terminate() response_consumer.terminate()
class _EventStreamStreamMethod(face_interfaces.EventStreamInStreamOutMethod): class _EventStreamStreamMethod(face_interfaces.MethodImplementation):
def __init__(self, stream_stream_test_method, control, pool): def __init__(self, stream_stream_test_method, control, pool):
self._test_method = stream_stream_test_method self._test_method = stream_stream_test_method
self._control = control self._control = control
self._pool = pool self._pool = pool
def service(self, response_consumer, context): self.cardinality = cardinality.Cardinality.STREAM_STREAM
self.style = style.Service.EVENT
def stream_stream_event(self, response_consumer, context):
request_consumer = self._test_method.service( request_consumer = self._test_method.service(
response_consumer, context, self._control) response_consumer, context, self._control)
if self._pool is None: if self._pool is None:
@ -332,7 +334,7 @@ class _StreamUnaryAdaptation(object):
response_consumer.consume_and_terminate, context, control) response_consumer.consume_and_terminate, context, control)
class _MultiMethod(face_interfaces.MultiMethod): class _MultiMethodImplementation(face_interfaces.MultiMethodImplementation):
def __init__(self, methods, control, pool): def __init__(self, methods, control, pool):
self._methods = methods self._methods = methods
@ -427,19 +429,21 @@ def digest(service, control, pool):
adaptations.update(unary_stream.adaptations) adaptations.update(unary_stream.adaptations)
adaptations.update(stream_unary.adaptations) adaptations.update(stream_unary.adaptations)
adaptations.update(stream_stream.adaptations) adaptations.update(stream_stream.adaptations)
inlines = dict(unary_unary.inlines)
inlines.update(unary_stream.inlines)
inlines.update(stream_unary.inlines)
inlines.update(stream_stream.inlines)
events = dict(unary_unary.events)
events.update(unary_stream.events)
events.update(stream_unary.events)
events.update(stream_stream.events)
return TestServiceDigest( return TestServiceDigest(
service.name(), service.name(),
methods, methods,
unary_unary.inlines, inlines,
unary_stream.inlines, events,
stream_unary.inlines, _MultiMethodImplementation(adaptations, control, pool),
stream_stream.inlines,
unary_unary.events,
unary_stream.events,
stream_unary.events,
stream_stream.events,
_MultiMethod(adaptations, control, pool),
unary_unary.messages, unary_unary.messages,
unary_stream.messages, unary_stream.messages,
stream_unary.messages, stream_unary.messages,

9
src/python/src/grpc/framework/face/testing/event_invocation_synchronous_event_service_test_case.py
Unescape View File

@ -60,14 +60,9 @@ class EventInvocationSynchronousEventServiceTestCase(
self.digest = digest.digest( self.digest = digest.digest(
stock_service.STOCK_TEST_SERVICE, self.control, None) stock_service.STOCK_TEST_SERVICE, self.control, None)
self.server, self.stub, self.memo = self.set_up_implementation( self.stub, self.memo = self.set_up_implementation(
self.digest.name, self.digest.methods, self.digest.name, self.digest.methods,
{}, {}, {}, {}, self.digest.event_method_implementations, None)
self.digest.event_unary_unary_methods,
self.digest.event_unary_stream_methods,
self.digest.event_stream_unary_methods,
self.digest.event_stream_stream_methods,
None)
def tearDown(self): def tearDown(self):
"""See unittest.TestCase.tearDown for full specification. """See unittest.TestCase.tearDown for full specification.

17
src/python/src/grpc/framework/face/testing/future_invocation_asynchronous_event_service_test_case.py
Unescape View File

@ -91,14 +91,9 @@ class FutureInvocationAsynchronousEventServiceTestCase(
self.digest = digest.digest( self.digest = digest.digest(
stock_service.STOCK_TEST_SERVICE, self.control, self.digest_pool) stock_service.STOCK_TEST_SERVICE, self.control, self.digest_pool)
self.server, self.stub, self.memo = self.set_up_implementation( self.stub, self.memo = self.set_up_implementation(
self.digest.name, self.digest.methods, self.digest.name, self.digest.methods,
{}, {}, {}, {}, self.digest.event_method_implementations, None)
self.digest.event_unary_unary_methods,
self.digest.event_unary_stream_methods,
self.digest.event_stream_unary_methods,
self.digest.event_stream_stream_methods,
None)
def tearDown(self): def tearDown(self):
"""See unittest.TestCase.tearDown for full specification. """See unittest.TestCase.tearDown for full specification.
@ -190,8 +185,8 @@ class FutureInvocationAsynchronousEventServiceTestCase(
request = test_messages.request() request = test_messages.request()
with self.control.pause(): with self.control.pause():
sync_async = self.stub.unary_unary_sync_async(name) multi_callable = self.stub.unary_unary_multi_callable(name)
response_future = sync_async.async(request, _TIMEOUT) response_future = multi_callable.future(request, _TIMEOUT)
self.assertIsInstance( self.assertIsInstance(
response_future.exception(), exceptions.ExpirationError) response_future.exception(), exceptions.ExpirationError)
with self.assertRaises(exceptions.ExpirationError): with self.assertRaises(exceptions.ExpirationError):
@ -216,8 +211,8 @@ class FutureInvocationAsynchronousEventServiceTestCase(
requests = test_messages.requests() requests = test_messages.requests()
with self.control.pause(): with self.control.pause():
sync_async = self.stub.stream_unary_sync_async(name) multi_callable = self.stub.stream_unary_multi_callable(name)
response_future = sync_async.async(iter(requests), _TIMEOUT) response_future = multi_callable.future(iter(requests), _TIMEOUT)
self.assertIsInstance( self.assertIsInstance(
response_future.exception(), exceptions.ExpirationError) response_future.exception(), exceptions.ExpirationError)
with self.assertRaises(exceptions.ExpirationError): with self.assertRaises(exceptions.ExpirationError):

38
src/python/src/grpc/framework/face/testing/service.py
Unescape View File

@ -36,8 +36,8 @@ from grpc.framework.face import interfaces as face_interfaces # pylint: disable
from grpc.framework.face.testing import interfaces from grpc.framework.face.testing import interfaces
class UnaryUnaryTestMethod(interfaces.Method): class UnaryUnaryTestMethodImplementation(interfaces.Method):
"""Like face_interfaces.EventValueInValueOutMethod but with a control.""" """A controllable implementation of a unary-unary RPC method."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@ -93,8 +93,8 @@ class UnaryUnaryTestMessages(object):
raise NotImplementedError() raise NotImplementedError()
class UnaryStreamTestMethod(interfaces.Method): class UnaryStreamTestMethodImplementation(interfaces.Method):
"""Like face_interfaces.EventValueInStreamOutMethod but with a control.""" """A controllable implementation of a unary-stream RPC method."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@ -106,7 +106,7 @@ class UnaryStreamTestMethod(interfaces.Method):
request: The single request message for the RPC. request: The single request message for the RPC.
response_consumer: A stream.Consumer to be called to accept the response response_consumer: A stream.Consumer to be called to accept the response
messages of the RPC. messages of the RPC.
context: An RpcContext object. context: A face_interfaces.RpcContext object.
control: A test_control.Control to control execution of this method. control: A test_control.Control to control execution of this method.
Raises: Raises:
@ -150,8 +150,8 @@ class UnaryStreamTestMessages(object):
raise NotImplementedError() raise NotImplementedError()
class StreamUnaryTestMethod(interfaces.Method): class StreamUnaryTestMethodImplementation(interfaces.Method):
"""Like face_interfaces.EventStreamInValueOutMethod but with a control.""" """A controllable implementation of a stream-unary RPC method."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@ -162,7 +162,7 @@ class StreamUnaryTestMethod(interfaces.Method):
Args: Args:
response_callback: A callback to be called to accept the response message response_callback: A callback to be called to accept the response message
of the RPC. of the RPC.
context: An RpcContext object. context: A face_interfaces.RpcContext object.
control: A test_control.Control to control execution of this method. control: A test_control.Control to control execution of this method.
Returns: Returns:
@ -214,8 +214,8 @@ class StreamUnaryTestMessages(object):
raise NotImplementedError() raise NotImplementedError()
class StreamStreamTestMethod(interfaces.Method): class StreamStreamTestMethodImplementation(interfaces.Method):
"""Like face_interfaces.EventStreamInStreamOutMethod but with a control.""" """A controllable implementation of a stream-stream RPC method."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@ -226,7 +226,7 @@ class StreamStreamTestMethod(interfaces.Method):
Args: Args:
response_consumer: A stream.Consumer to be called to accept the response response_consumer: A stream.Consumer to be called to accept the response
messages of the RPC. messages of the RPC.
context: An RpcContext object. context: A face_interfaces.RpcContext object.
control: A test_control.Control to control execution of this method. control: A test_control.Control to control execution of this method.
Returns: Returns:
@ -298,8 +298,8 @@ class TestService(object):
Returns: Returns:
A dict from method name to pair. The first element of the pair A dict from method name to pair. The first element of the pair
is a UnaryUnaryTestMethod object and the second element is a sequence is a UnaryUnaryTestMethodImplementation object and the second element
of UnaryUnaryTestMethodMessages objects. is a sequence of UnaryUnaryTestMethodMessages objects.
""" """
raise NotImplementedError() raise NotImplementedError()
@ -309,8 +309,8 @@ class TestService(object):
Returns: Returns:
A dict from method name to pair. The first element of the pair is a A dict from method name to pair. The first element of the pair is a
UnaryStreamTestMethod object and the second element is a sequence of UnaryStreamTestMethodImplementation object and the second element is a
UnaryStreamTestMethodMessages objects. sequence of UnaryStreamTestMethodMessages objects.
""" """
raise NotImplementedError() raise NotImplementedError()
@ -320,8 +320,8 @@ class TestService(object):
Returns: Returns:
A dict from method name to pair. The first element of the pair is a A dict from method name to pair. The first element of the pair is a
StreamUnaryTestMethod object and the second element is a sequence of StreamUnaryTestMethodImplementation object and the second element is a
StreamUnaryTestMethodMessages objects. sequence of StreamUnaryTestMethodMessages objects.
""" """
raise NotImplementedError() raise NotImplementedError()
@ -331,7 +331,7 @@ class TestService(object):
Returns: Returns:
A dict from method name to pair. The first element of the pair is a A dict from method name to pair. The first element of the pair is a
StreamStreamTestMethod object and the second element is a sequence of StreamStreamTestMethodImplementation object and the second element is a
StreamStreamTestMethodMessages objects. sequence of StreamStreamTestMethodMessages objects.
""" """
raise NotImplementedError() raise NotImplementedError()

8
src/python/src/grpc/framework/face/testing/stock_service.py
Unescape View File

@ -139,7 +139,7 @@ def _get_highest_trade_price(stock_reply_callback, control, active):
return StockRequestConsumer() return StockRequestConsumer()
class GetLastTradePrice(service.UnaryUnaryTestMethod): class GetLastTradePrice(service.UnaryUnaryTestMethodImplementation):
"""GetLastTradePrice for use in tests.""" """GetLastTradePrice for use in tests."""
def name(self): def name(self):
@ -186,7 +186,7 @@ class GetLastTradePriceMessages(service.UnaryUnaryTestMessages):
test_case.assertEqual(_price(request.symbol), response.price) test_case.assertEqual(_price(request.symbol), response.price)
class GetLastTradePriceMultiple(service.StreamStreamTestMethod): class GetLastTradePriceMultiple(service.StreamStreamTestMethodImplementation):
"""GetLastTradePriceMultiple for use in tests.""" """GetLastTradePriceMultiple for use in tests."""
def name(self): def name(self):
@ -238,7 +238,7 @@ class GetLastTradePriceMultipleMessages(service.StreamStreamTestMessages):
test_case.assertEqual(_price(stock_request.symbol), stock_reply.price) test_case.assertEqual(_price(stock_request.symbol), stock_reply.price)
class WatchFutureTrades(service.UnaryStreamTestMethod): class WatchFutureTrades(service.UnaryStreamTestMethodImplementation):
"""WatchFutureTrades for use in tests.""" """WatchFutureTrades for use in tests."""
def name(self): def name(self):
@ -288,7 +288,7 @@ class WatchFutureTradesMessages(service.UnaryStreamTestMessages):
test_case.assertEqual(base_price + index, response.price) test_case.assertEqual(base_price + index, response.price)
class GetHighestTradePrice(service.StreamUnaryTestMethod): class GetHighestTradePrice(service.StreamUnaryTestMethodImplementation):
"""GetHighestTradePrice for use in tests.""" """GetHighestTradePrice for use in tests."""
def name(self): def name(self):

53
src/python/src/grpc/framework/face/testing/test_case.py
Unescape View File

@ -46,55 +46,24 @@ class FaceTestCase(object):
@abc.abstractmethod @abc.abstractmethod
def set_up_implementation( def set_up_implementation(
self, self, name, methods, method_implementations,
name, multi_method_implementation):
methods,
inline_value_in_value_out_methods,
inline_value_in_stream_out_methods,
inline_stream_in_value_out_methods,
inline_stream_in_stream_out_methods,
event_value_in_value_out_methods,
event_value_in_stream_out_methods,
event_stream_in_value_out_methods,
event_stream_in_stream_out_methods,
multi_method):
"""Instantiates the Face Layer implementation under test. """Instantiates the Face Layer implementation under test.
Args: Args:
name: The service name to be used in the test. name: The service name to be used in the test.
methods: A sequence of interfaces.Method objects describing the RPC methods: A sequence of interfaces.Method objects describing the RPC
methods that will be called during the test. methods that will be called during the test.
inline_value_in_value_out_methods: A dictionary from string method names method_implementations: A dictionary from string RPC method name to
to face_interfaces.InlineValueInValueOutMethod implementations of those face_interfaces.MethodImplementation object specifying
methods. implementation of an RPC method.
inline_value_in_stream_out_methods: A dictionary from string method names multi_method_implementation: An face_interfaces.MultiMethodImplementation
to face_interfaces.InlineValueInStreamOutMethod implementations of those or None.
methods.
inline_stream_in_value_out_methods: A dictionary from string method names
to face_interfaces.InlineStreamInValueOutMethod implementations of those
methods.
inline_stream_in_stream_out_methods: A dictionary from string method names
to face_interfaces.InlineStreamInStreamOutMethod implementations of
those methods.
event_value_in_value_out_methods: A dictionary from string method names
to face_interfaces.EventValueInValueOutMethod implementations of those
methods.
event_value_in_stream_out_methods: A dictionary from string method names
to face_interfaces.EventValueInStreamOutMethod implementations of those
methods.
event_stream_in_value_out_methods: A dictionary from string method names
to face_interfaces.EventStreamInValueOutMethod implementations of those
methods.
event_stream_in_stream_out_methods: A dictionary from string method names
to face_interfaces.EventStreamInStreamOutMethod implementations of those
methods.
multi_method: An face_interfaces.MultiMethod, or None.
Returns: Returns:
A sequence of length three the first element of which is a A sequence of length two the first element of which is a
face_interfaces.Server, the second element of which is a face_interfaces.GenericStub (backed by the given method
face_interfaces.Stub, (both of which are backed by the given method implementations), and the second element of which is an arbitrary memo
implementations), and the third element of which is an arbitrary memo
object to be kept and passed to tearDownImplementation at the conclusion object to be kept and passed to tearDownImplementation at the conclusion
of the test. of the test.
""" """
@ -105,7 +74,7 @@ class FaceTestCase(object):
"""Destroys the Face layer implementation under test. """Destroys the Face layer implementation under test.
Args: Args:
memo: The object from the third position of the return value of memo: The object from the second position of the return value of
set_up_implementation. set_up_implementation.
""" """
raise NotImplementedError() raise NotImplementedError()

174
src/python/src/grpc/framework/face/utilities.py
Unescape View File

@ -27,101 +27,44 @@
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""Utilities for the face layer of RPC Framework.""" """Utilities for RPC framework's face layer."""
# stream is referenced from specification in this module. import collections
from grpc.framework.face import interfaces
from grpc.framework.foundation import stream # pylint: disable=unused-import
class _InlineUnaryUnaryMethod(interfaces.InlineValueInValueOutMethod):
def __init__(self, behavior):
self._behavior = behavior
def service(self, request, context):
return self._behavior(request, context)
class _InlineUnaryStreamMethod(interfaces.InlineValueInStreamOutMethod):
def __init__(self, behavior):
self._behavior = behavior
def service(self, request, context):
return self._behavior(request, context)
class _InlineStreamUnaryMethod(interfaces.InlineStreamInValueOutMethod):
def __init__(self, behavior):
self._behavior = behavior
def service(self, request_iterator, context):
return self._behavior(request_iterator, context)
class _InlineStreamStreamMethod(interfaces.InlineStreamInStreamOutMethod):
def __init__(self, behavior):
self._behavior = behavior
def service(self, request_iterator, context):
return self._behavior(request_iterator, context)
from grpc.framework.common import cardinality
from grpc.framework.common import style
from grpc.framework.face import interfaces
from grpc.framework.foundation import stream
class _EventUnaryUnaryMethod(interfaces.EventValueInValueOutMethod):
def __init__(self, behavior):
self._behavior = behavior
def service(self, request, response_callback, context):
return self._behavior(request, response_callback, context)
class _EventUnaryStreamMethod(interfaces.EventValueInStreamOutMethod):
def __init__(self, behavior):
self._behavior = behavior
def service(self, request, response_consumer, context):
return self._behavior(request, response_consumer, context)
class _EventStreamUnaryMethod(interfaces.EventStreamInValueOutMethod):
def __init__(self, behavior):
self._behavior = behavior
def service(self, response_callback, context):
return self._behavior(response_callback, context)
class _EventStreamStreamMethod(interfaces.EventStreamInStreamOutMethod):
def __init__(self, behavior):
self._behavior = behavior
def service(self, response_consumer, context): class _MethodImplementation(
return self._behavior(response_consumer, context) interfaces.MethodImplementation,
collections.namedtuple(
'_MethodImplementation',
['cardinality', 'style', 'unary_unary_inline', 'unary_stream_inline',
'stream_unary_inline', 'stream_stream_inline', 'unary_unary_event',
'unary_stream_event', 'stream_unary_event', 'stream_stream_event',])):
pass
def inline_unary_unary_method(behavior): def unary_unary_inline(behavior):
"""Creates an interfaces.InlineValueInValueOutMethod from a behavior. """Creates an interfaces.MethodImplementation for the given behavior.
Args: Args:
behavior: The implementation of a unary-unary RPC method as a callable behavior: The implementation of a unary-unary RPC method as a callable value
value that takes a request value and an interfaces.RpcContext object and that takes a request value and an interfaces.RpcContext object and
returns a response value. returns a response value.
Returns: Returns:
An interfaces.InlineValueInValueOutMethod derived from the given behavior. An interfaces.MethodImplementation derived from the given behavior.
""" """
return _InlineUnaryUnaryMethod(behavior) return _MethodImplementation(
cardinality.Cardinality.UNARY_UNARY, style.Service.INLINE, behavior,
None, None, None, None, None, None, None)
def inline_unary_stream_method(behavior): def unary_stream_inline(behavior):
"""Creates an interfaces.InlineValueInStreamOutMethod from a behavior. """Creates an interfaces.MethodImplementation for the given behavior.
Args: Args:
behavior: The implementation of a unary-stream RPC method as a callable behavior: The implementation of a unary-stream RPC method as a callable
@ -129,13 +72,15 @@ def inline_unary_stream_method(behavior):
returns an iterator of response values. returns an iterator of response values.
Returns: Returns:
An interfaces.InlineValueInStreamOutMethod derived from the given behavior. An interfaces.MethodImplementation derived from the given behavior.
""" """
return _InlineUnaryStreamMethod(behavior) return _MethodImplementation(
cardinality.Cardinality.UNARY_STREAM, style.Service.INLINE, None,
behavior, None, None, None, None, None, None)
def inline_stream_unary_method(behavior): def stream_unary_inline(behavior):
"""Creates an interfaces.InlineStreamInValueOutMethod from a behavior. """Creates an interfaces.MethodImplementation for the given behavior.
Args: Args:
behavior: The implementation of a stream-unary RPC method as a callable behavior: The implementation of a stream-unary RPC method as a callable
@ -143,13 +88,15 @@ def inline_stream_unary_method(behavior):
interfaces.RpcContext object and returns a response value. interfaces.RpcContext object and returns a response value.
Returns: Returns:
An interfaces.InlineStreamInValueOutMethod derived from the given behavior. An interfaces.MethodImplementation derived from the given behavior.
""" """
return _InlineStreamUnaryMethod(behavior) return _MethodImplementation(
cardinality.Cardinality.STREAM_UNARY, style.Service.INLINE, None, None,
behavior, None, None, None, None, None)
def inline_stream_stream_method(behavior): def stream_stream_inline(behavior):
"""Creates an interfaces.InlineStreamInStreamOutMethod from a behavior. """Creates an interfaces.MethodImplementation for the given behavior.
Args: Args:
behavior: The implementation of a stream-stream RPC method as a callable behavior: The implementation of a stream-stream RPC method as a callable
@ -157,14 +104,15 @@ def inline_stream_stream_method(behavior):
interfaces.RpcContext object and returns an iterator of response values. interfaces.RpcContext object and returns an iterator of response values.
Returns: Returns:
An interfaces.InlineStreamInStreamOutMethod derived from the given An interfaces.MethodImplementation derived from the given behavior.
behavior.
""" """
return _InlineStreamStreamMethod(behavior) return _MethodImplementation(
cardinality.Cardinality.STREAM_STREAM, style.Service.INLINE, None, None,
None, behavior, None, None, None, None)
def event_unary_unary_method(behavior): def unary_unary_event(behavior):
"""Creates an interfaces.EventValueInValueOutMethod from a behavior. """Creates an interfaces.MethodImplementation for the given behavior.
Args: Args:
behavior: The implementation of a unary-unary RPC method as a callable behavior: The implementation of a unary-unary RPC method as a callable
@ -172,27 +120,31 @@ def event_unary_unary_method(behavior):
the response value of the RPC, and an interfaces.RpcContext. the response value of the RPC, and an interfaces.RpcContext.
Returns: Returns:
An interfaces.EventValueInValueOutMethod derived from the given behavior. An interfaces.MethodImplementation derived from the given behavior.
""" """
return _EventUnaryUnaryMethod(behavior) return _MethodImplementation(
cardinality.Cardinality.UNARY_UNARY, style.Service.EVENT, None, None,
None, None, behavior, None, None, None)
def event_unary_stream_method(behavior): def unary_stream_event(behavior):
"""Creates an interfaces.EventValueInStreamOutMethod from a behavior. """Creates an interfaces.MethodImplementation for the given behavior.
Args: Args:
behavior: The implementation of a unary-stream RPC method as a callable behavior: The implementation of a unary-stream RPC method as a callable
value that takes a request value, a stream.Consumer to which to pass the value that takes a request value, a stream.Consumer to which to pass the
response values of the RPC, and an interfaces.RpcContext. the response values of the RPC, and an interfaces.RpcContext.
Returns: Returns:
An interfaces.EventValueInStreamOutMethod derived from the given behavior. An interfaces.MethodImplementation derived from the given behavior.
""" """
return _EventUnaryStreamMethod(behavior) return _MethodImplementation(
cardinality.Cardinality.UNARY_STREAM, style.Service.EVENT, None, None,
None, None, None, behavior, None, None)
def event_stream_unary_method(behavior): def stream_unary_event(behavior):
"""Creates an interfaces.EventStreamInValueOutMethod from a behavior. """Creates an interfaces.MethodImplementation for the given behavior.
Args: Args:
behavior: The implementation of a stream-unary RPC method as a callable behavior: The implementation of a stream-unary RPC method as a callable
@ -201,13 +153,15 @@ def event_stream_unary_method(behavior):
which the request values of the RPC should be passed. which the request values of the RPC should be passed.
Returns: Returns:
An interfaces.EventStreamInValueOutMethod derived from the given behavior. An interfaces.MethodImplementation derived from the given behavior.
""" """
return _EventStreamUnaryMethod(behavior) return _MethodImplementation(
cardinality.Cardinality.STREAM_UNARY, style.Service.EVENT, None, None,
None, None, None, None, behavior, None)
def event_stream_stream_method(behavior): def stream_stream_event(behavior):
"""Creates an interfaces.EventStreamInStreamOutMethod from a behavior. """Creates an interfaces.MethodImplementation for the given behavior.
Args: Args:
behavior: The implementation of a stream-stream RPC method as a callable behavior: The implementation of a stream-stream RPC method as a callable
@ -216,6 +170,8 @@ def event_stream_stream_method(behavior):
which the request values of the RPC should be passed. which the request values of the RPC should be passed.
Returns: Returns:
An interfaces.EventStreamInStreamOutMethod derived from the given behavior. An interfaces.MethodImplementation derived from the given behavior.
""" """
return _EventStreamStreamMethod(behavior) return _MethodImplementation(
cardinality.Cardinality.STREAM_STREAM, style.Service.EVENT, None, None,
None, None, None, None, None, behavior)

Write Preview
Loading…
Cancel
Save