class PayloadSigningServerInterceptor(grpc.ServerInterceptor):
def __init__(identity):
self._identity = identity
def intercept_service(self, continuation, handler_call_details):
handler = continuation(handler_call_details)
if handler is None:
return None
if handler.request_streaming and handler.response_streaming:
return grpc.stream_stream_rpc_method_handler(handler.stream_stream,
request_deserializer=_verifying_deserializer(self._identity, handler.request_deserializer),
response_serializer=_signing_serializer(self._identity, handler._response_serializer))
elif handler.request_streaming and not handler.response_streaming:
return grpc.stream_unary_rpc_method_handler(handler.stream_unary,
request_deserializer=_verifying_deserializer(self._identity, handler.request_deserializer),
response_serializer=_signing_serializer(self._identity, handler._response_serializer))
elif not handler.request_streaming and handler.response_streaming:
return grpc.unary_stream_rpc_method_handler(handler.unary_stream,
request_deserializer=_verifying_deserializer(self._identity, handler.request_deserializer),
response_serializer=_signing_serializer(self._identity, handler._response_serializer))
else:
return grpc.unary_unary_rpc_method_handler(handler.unary_unary,
request_deserializer=_verifying_deserializer(self._identity, handler.request_deserializer),
response_serializer=_signing_serializer(self._identity, handler._response_serializer))