• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2017 gRPC authors.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""Base class for interceptors that operate on all RPC types."""
15
16import grpc
17
18
19class _GenericClientInterceptor(grpc.UnaryUnaryClientInterceptor,
20                                grpc.UnaryStreamClientInterceptor,
21                                grpc.StreamUnaryClientInterceptor,
22                                grpc.StreamStreamClientInterceptor):
23
24    def __init__(self, interceptor_function):
25        self._fn = interceptor_function
26
27    def intercept_unary_unary(self, continuation, client_call_details, request):
28        new_details, new_request_iterator, postprocess = self._fn(
29            client_call_details, iter((request,)), False, False)
30        response = continuation(new_details, next(new_request_iterator))
31        return postprocess(response) if postprocess else response
32
33    def intercept_unary_stream(self, continuation, client_call_details,
34                               request):
35        new_details, new_request_iterator, postprocess = self._fn(
36            client_call_details, iter((request,)), False, True)
37        response_it = continuation(new_details, next(new_request_iterator))
38        return postprocess(response_it) if postprocess else response_it
39
40    def intercept_stream_unary(self, continuation, client_call_details,
41                               request_iterator):
42        new_details, new_request_iterator, postprocess = self._fn(
43            client_call_details, request_iterator, True, False)
44        response = continuation(new_details, new_request_iterator)
45        return postprocess(response) if postprocess else response
46
47    def intercept_stream_stream(self, continuation, client_call_details,
48                                request_iterator):
49        new_details, new_request_iterator, postprocess = self._fn(
50            client_call_details, request_iterator, True, True)
51        response_it = continuation(new_details, new_request_iterator)
52        return postprocess(response_it) if postprocess else response_it
53
54
55def create(intercept_call):
56    return _GenericClientInterceptor(intercept_call)
57