aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/python/grpcio/grpc/_utilities.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/python/grpcio/grpc/_utilities.py')
-rw-r--r--src/python/grpcio/grpc/_utilities.py235
1 files changed, 119 insertions, 116 deletions
diff --git a/src/python/grpcio/grpc/_utilities.py b/src/python/grpcio/grpc/_utilities.py
index a375896e6e..7c602eb37e 100644
--- a/src/python/grpcio/grpc/_utilities.py
+++ b/src/python/grpcio/grpc/_utilities.py
@@ -26,7 +26,6 @@
# 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.
-
"""Internal utilities for gRPC Python."""
import collections
@@ -44,132 +43,136 @@ _DONE_CALLBACK_EXCEPTION_LOG_MESSAGE = (
class RpcMethodHandler(
- collections.namedtuple(
- '_RpcMethodHandler',
- ('request_streaming', 'response_streaming', 'request_deserializer',
- 'response_serializer', 'unary_unary', 'unary_stream', 'stream_unary',
- 'stream_stream',)),
- grpc.RpcMethodHandler):
- pass
+ collections.namedtuple('_RpcMethodHandler', (
+ 'request_streaming',
+ 'response_streaming',
+ 'request_deserializer',
+ 'response_serializer',
+ 'unary_unary',
+ 'unary_stream',
+ 'stream_unary',
+ 'stream_stream',)), grpc.RpcMethodHandler):
+ pass
class DictionaryGenericHandler(grpc.ServiceRpcHandler):
- def __init__(self, service, method_handlers):
- self._name = service
- self._method_handlers = {
- _common.fully_qualified_method(service, method): method_handler
- for method, method_handler in six.iteritems(method_handlers)}
+ def __init__(self, service, method_handlers):
+ self._name = service
+ self._method_handlers = {
+ _common.fully_qualified_method(service, method): method_handler
+ for method, method_handler in six.iteritems(method_handlers)
+ }
- def service_name(self):
- return self._name
+ def service_name(self):
+ return self._name
- def service(self, handler_call_details):
- return self._method_handlers.get(handler_call_details.method)
+ def service(self, handler_call_details):
+ return self._method_handlers.get(handler_call_details.method)
class _ChannelReadyFuture(grpc.Future):
- def __init__(self, channel):
- self._condition = threading.Condition()
- self._channel = channel
-
- self._matured = False
- self._cancelled = False
- self._done_callbacks = []
-
- def _block(self, timeout):
- until = None if timeout is None else time.time() + timeout
- with self._condition:
- while True:
- if self._cancelled:
- raise grpc.FutureCancelledError()
- elif self._matured:
- return
- else:
- if until is None:
- self._condition.wait()
- else:
- remaining = until - time.time()
- if remaining < 0:
- raise grpc.FutureTimeoutError()
+ def __init__(self, channel):
+ self._condition = threading.Condition()
+ self._channel = channel
+
+ self._matured = False
+ self._cancelled = False
+ self._done_callbacks = []
+
+ def _block(self, timeout):
+ until = None if timeout is None else time.time() + timeout
+ with self._condition:
+ while True:
+ if self._cancelled:
+ raise grpc.FutureCancelledError()
+ elif self._matured:
+ return
+ else:
+ if until is None:
+ self._condition.wait()
+ else:
+ remaining = until - time.time()
+ if remaining < 0:
+ raise grpc.FutureTimeoutError()
+ else:
+ self._condition.wait(timeout=remaining)
+
+ def _update(self, connectivity):
+ with self._condition:
+ if (not self._cancelled and
+ connectivity is grpc.ChannelConnectivity.READY):
+ self._matured = True
+ self._channel.unsubscribe(self._update)
+ self._condition.notify_all()
+ done_callbacks = tuple(self._done_callbacks)
+ self._done_callbacks = None
+ else:
+ return
+
+ for done_callback in done_callbacks:
+ callable_util.call_logging_exceptions(
+ done_callback, _DONE_CALLBACK_EXCEPTION_LOG_MESSAGE, self)
+
+ def cancel(self):
+ with self._condition:
+ if not self._matured:
+ self._cancelled = True
+ self._channel.unsubscribe(self._update)
+ self._condition.notify_all()
+ done_callbacks = tuple(self._done_callbacks)
+ self._done_callbacks = None
else:
- self._condition.wait(timeout=remaining)
-
- def _update(self, connectivity):
- with self._condition:
- if (not self._cancelled and
- connectivity is grpc.ChannelConnectivity.READY):
- self._matured = True
- self._channel.unsubscribe(self._update)
- self._condition.notify_all()
- done_callbacks = tuple(self._done_callbacks)
- self._done_callbacks = None
- else:
- return
-
- for done_callback in done_callbacks:
- callable_util.call_logging_exceptions(
- done_callback, _DONE_CALLBACK_EXCEPTION_LOG_MESSAGE, self)
-
- def cancel(self):
- with self._condition:
- if not self._matured:
- self._cancelled = True
- self._channel.unsubscribe(self._update)
- self._condition.notify_all()
- done_callbacks = tuple(self._done_callbacks)
- self._done_callbacks = None
- else:
- return False
-
- for done_callback in done_callbacks:
- callable_util.call_logging_exceptions(
- done_callback, _DONE_CALLBACK_EXCEPTION_LOG_MESSAGE, self)
-
- def cancelled(self):
- with self._condition:
- return self._cancelled
-
- def running(self):
- with self._condition:
- return not self._cancelled and not self._matured
-
- def done(self):
- with self._condition:
- return self._cancelled or self._matured
-
- def result(self, timeout=None):
- self._block(timeout)
- return None
-
- def exception(self, timeout=None):
- self._block(timeout)
- return None
-
- def traceback(self, timeout=None):
- self._block(timeout)
- return None
-
- def add_done_callback(self, fn):
- with self._condition:
- if not self._cancelled and not self._matured:
- self._done_callbacks.append(fn)
- return
-
- fn(self)
-
- def start(self):
- with self._condition:
- self._channel.subscribe(self._update, try_to_connect=True)
-
- def __del__(self):
- with self._condition:
- if not self._cancelled and not self._matured:
- self._channel.unsubscribe(self._update)
+ return False
+
+ for done_callback in done_callbacks:
+ callable_util.call_logging_exceptions(
+ done_callback, _DONE_CALLBACK_EXCEPTION_LOG_MESSAGE, self)
+
+ def cancelled(self):
+ with self._condition:
+ return self._cancelled
+
+ def running(self):
+ with self._condition:
+ return not self._cancelled and not self._matured
+
+ def done(self):
+ with self._condition:
+ return self._cancelled or self._matured
+
+ def result(self, timeout=None):
+ self._block(timeout)
+ return None
+
+ def exception(self, timeout=None):
+ self._block(timeout)
+ return None
+
+ def traceback(self, timeout=None):
+ self._block(timeout)
+ return None
+
+ def add_done_callback(self, fn):
+ with self._condition:
+ if not self._cancelled and not self._matured:
+ self._done_callbacks.append(fn)
+ return
+
+ fn(self)
+
+ def start(self):
+ with self._condition:
+ self._channel.subscribe(self._update, try_to_connect=True)
+
+ def __del__(self):
+ with self._condition:
+ if not self._cancelled and not self._matured:
+ self._channel.unsubscribe(self._update)
def channel_ready_future(channel):
- ready_future = _ChannelReadyFuture(channel)
- ready_future.start()
- return ready_future
+ ready_future = _ChannelReadyFuture(channel)
+ ready_future.start()
+ return ready_future