|  | @@ -0,0 +1,545 @@
 | 
	
		
			
				|  |  | +# 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.
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +"""State and behavior for translating between sync and async control flow."""
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +import collections
 | 
	
		
			
				|  |  | +import enum
 | 
	
		
			
				|  |  | +import sys
 | 
	
		
			
				|  |  | +import threading
 | 
	
		
			
				|  |  | +import time
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +from grpc.framework.foundation import abandonment
 | 
	
		
			
				|  |  | +from grpc.framework.foundation import callable_util
 | 
	
		
			
				|  |  | +from grpc.framework.foundation import future
 | 
	
		
			
				|  |  | +from grpc.framework.foundation import stream
 | 
	
		
			
				|  |  | +from grpc.framework.interfaces.base import base
 | 
	
		
			
				|  |  | +from grpc.framework.interfaces.base import utilities
 | 
	
		
			
				|  |  | +from grpc.framework.interfaces.face import face
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +_DONE_CALLBACK_LOG_MESSAGE = 'Exception calling Future "done" callback!'
 | 
	
		
			
				|  |  | +_INTERNAL_ERROR_LOG_MESSAGE = ':-( RPC Framework (Crust) Internal Error! )-:'
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +_CANNOT_SET_INITIAL_METADATA = (
 | 
	
		
			
				|  |  | +    'Could not set initial metadata - has it already been set, or has a ' +
 | 
	
		
			
				|  |  | +    'payload already been sent?')
 | 
	
		
			
				|  |  | +_CANNOT_SET_TERMINAL_METADATA = (
 | 
	
		
			
				|  |  | +    'Could not set terminal metadata - has it already been set, or has RPC ' +
 | 
	
		
			
				|  |  | +    'completion already been indicated?')
 | 
	
		
			
				|  |  | +_CANNOT_SET_CODE = (
 | 
	
		
			
				|  |  | +    'Could not set code - has it already been set, or has RPC completion ' +
 | 
	
		
			
				|  |  | +    'already been indicated?')
 | 
	
		
			
				|  |  | +_CANNOT_SET_DETAILS = (
 | 
	
		
			
				|  |  | +    'Could not set details - has it already been set, or has RPC completion ' +
 | 
	
		
			
				|  |  | +    'already been indicated?')
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +class _DummyOperator(base.Operator):
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def advance(
 | 
	
		
			
				|  |  | +      self, initial_metadata=None, payload=None, completion=None,
 | 
	
		
			
				|  |  | +      allowance=None):
 | 
	
		
			
				|  |  | +    pass
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +_DUMMY_OPERATOR = _DummyOperator()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +class _Awaited(
 | 
	
		
			
				|  |  | +    collections.namedtuple('_Awaited', ('kind', 'value',))):
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  @enum.unique
 | 
	
		
			
				|  |  | +  class Kind(enum.Enum):
 | 
	
		
			
				|  |  | +    NOT_YET_ARRIVED = 'not yet arrived'
 | 
	
		
			
				|  |  | +    ARRIVED = 'arrived'
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +_NOT_YET_ARRIVED = _Awaited(_Awaited.Kind.NOT_YET_ARRIVED, None)
 | 
	
		
			
				|  |  | +_ARRIVED_AND_NONE = _Awaited(_Awaited.Kind.ARRIVED, None)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +class _Transitory(
 | 
	
		
			
				|  |  | +    collections.namedtuple('_Transitory', ('kind', 'value',))):
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  @enum.unique
 | 
	
		
			
				|  |  | +  class Kind(enum.Enum):
 | 
	
		
			
				|  |  | +    NOT_YET_SEEN = 'not yet seen'
 | 
	
		
			
				|  |  | +    PRESENT = 'present'
 | 
	
		
			
				|  |  | +    GONE = 'gone'
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +_NOT_YET_SEEN = _Transitory(_Transitory.Kind.NOT_YET_SEEN, None)
 | 
	
		
			
				|  |  | +_GONE = _Transitory(_Transitory.Kind.GONE, None)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +class _Termination(
 | 
	
		
			
				|  |  | +    collections.namedtuple(
 | 
	
		
			
				|  |  | +        '_Termination', ('terminated', 'abortion', 'abortion_error',))):
 | 
	
		
			
				|  |  | +  """Values indicating whether and how an RPC has terminated.
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  Attributes:
 | 
	
		
			
				|  |  | +    terminated: A boolean indicating whether or not the RPC has terminated.
 | 
	
		
			
				|  |  | +    abortion: A face.Abortion value describing the RPC's abortion or None if the
 | 
	
		
			
				|  |  | +      RPC did not abort.
 | 
	
		
			
				|  |  | +    abortion_error: A face.AbortionError describing the RPC's abortion or None
 | 
	
		
			
				|  |  | +      if the RPC did not abort.
 | 
	
		
			
				|  |  | +  """
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +_NOT_TERMINATED = _Termination(False, None, None)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +_OPERATION_OUTCOME_TO_TERMINATION_CONSTRUCTOR = {
 | 
	
		
			
				|  |  | +    base.Outcome.COMPLETED: lambda *unused_args: _Termination(True, None, None),
 | 
	
		
			
				|  |  | +    base.Outcome.CANCELLED: lambda *args: _Termination(
 | 
	
		
			
				|  |  | +        True, face.Abortion(face.Abortion.Kind.CANCELLED, *args),
 | 
	
		
			
				|  |  | +        face.CancellationError(*args)),
 | 
	
		
			
				|  |  | +    base.Outcome.EXPIRED: lambda *args: _Termination(
 | 
	
		
			
				|  |  | +        True, face.Abortion(face.Abortion.Kind.EXPIRED, *args),
 | 
	
		
			
				|  |  | +        face.ExpirationError(*args)),
 | 
	
		
			
				|  |  | +    base.Outcome.LOCAL_SHUTDOWN: lambda *args: _Termination(
 | 
	
		
			
				|  |  | +        True, face.Abortion(face.Abortion.Kind.LOCAL_SHUTDOWN, *args),
 | 
	
		
			
				|  |  | +        face.LocalShutdownError(*args)),
 | 
	
		
			
				|  |  | +    base.Outcome.REMOTE_SHUTDOWN: lambda *args: _Termination(
 | 
	
		
			
				|  |  | +        True, face.Abortion(face.Abortion.Kind.REMOTE_SHUTDOWN, *args),
 | 
	
		
			
				|  |  | +        face.RemoteShutdownError(*args)),
 | 
	
		
			
				|  |  | +    base.Outcome.RECEPTION_FAILURE: lambda *args: _Termination(
 | 
	
		
			
				|  |  | +        True, face.Abortion(face.Abortion.Kind.NETWORK_FAILURE, *args),
 | 
	
		
			
				|  |  | +        face.NetworkError(*args)),
 | 
	
		
			
				|  |  | +    base.Outcome.TRANSMISSION_FAILURE: lambda *args: _Termination(
 | 
	
		
			
				|  |  | +        True, face.Abortion(face.Abortion.Kind.NETWORK_FAILURE, *args),
 | 
	
		
			
				|  |  | +        face.NetworkError(*args)),
 | 
	
		
			
				|  |  | +    base.Outcome.LOCAL_FAILURE: lambda *args: _Termination(
 | 
	
		
			
				|  |  | +        True, face.Abortion(face.Abortion.Kind.LOCAL_FAILURE, *args),
 | 
	
		
			
				|  |  | +        face.LocalError(*args)),
 | 
	
		
			
				|  |  | +    base.Outcome.REMOTE_FAILURE: lambda *args: _Termination(
 | 
	
		
			
				|  |  | +        True, face.Abortion(face.Abortion.Kind.REMOTE_FAILURE, *args),
 | 
	
		
			
				|  |  | +        face.RemoteError(*args)),
 | 
	
		
			
				|  |  | +}
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +def _wait_once_until(condition, until):
 | 
	
		
			
				|  |  | +  if until is None:
 | 
	
		
			
				|  |  | +    condition.wait()
 | 
	
		
			
				|  |  | +  else:
 | 
	
		
			
				|  |  | +    remaining = until - time.time()
 | 
	
		
			
				|  |  | +    if remaining < 0:
 | 
	
		
			
				|  |  | +      raise future.TimeoutError()
 | 
	
		
			
				|  |  | +    else:
 | 
	
		
			
				|  |  | +      condition.wait(timeout=remaining)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +def _done_callback_as_operation_termination_callback(
 | 
	
		
			
				|  |  | +    done_callback, rendezvous):
 | 
	
		
			
				|  |  | +  def operation_termination_callback(operation_outcome):
 | 
	
		
			
				|  |  | +    rendezvous.set_outcome(operation_outcome)
 | 
	
		
			
				|  |  | +    done_callback(rendezvous)
 | 
	
		
			
				|  |  | +  return operation_termination_callback
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +def _abortion_callback_as_operation_termination_callback(
 | 
	
		
			
				|  |  | +    rpc_abortion_callback, rendezvous_set_outcome):
 | 
	
		
			
				|  |  | +  def operation_termination_callback(operation_outcome):
 | 
	
		
			
				|  |  | +    termination = rendezvous_set_outcome(operation_outcome)
 | 
	
		
			
				|  |  | +    if termination.abortion is not None:
 | 
	
		
			
				|  |  | +      rpc_abortion_callback(termination.abortion)
 | 
	
		
			
				|  |  | +  return operation_termination_callback
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +class Rendezvous(base.Operator, future.Future, stream.Consumer, face.Call):
 | 
	
		
			
				|  |  | +  """A rendez-vous for the threads of an operation.
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  Instances of this object present iterator and stream.Consumer interfaces for
 | 
	
		
			
				|  |  | +  interacting with application code and present a base.Operator interface and
 | 
	
		
			
				|  |  | +  maintain a base.Operator internally for interacting with base interface code.
 | 
	
		
			
				|  |  | +  """
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def __init__(self, operator, operation_context):
 | 
	
		
			
				|  |  | +    self._condition = threading.Condition()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    self._operator = operator
 | 
	
		
			
				|  |  | +    self._operation_context = operation_context
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    self._up_initial_metadata = _NOT_YET_ARRIVED
 | 
	
		
			
				|  |  | +    self._up_payload = None
 | 
	
		
			
				|  |  | +    self._up_allowance = 1
 | 
	
		
			
				|  |  | +    self._up_completion = _NOT_YET_ARRIVED
 | 
	
		
			
				|  |  | +    self._down_initial_metadata = _NOT_YET_SEEN
 | 
	
		
			
				|  |  | +    self._down_payload = None
 | 
	
		
			
				|  |  | +    self._down_allowance = 1
 | 
	
		
			
				|  |  | +    self._down_terminal_metadata = _NOT_YET_SEEN
 | 
	
		
			
				|  |  | +    self._down_code = _NOT_YET_SEEN
 | 
	
		
			
				|  |  | +    self._down_details = _NOT_YET_SEEN
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    self._termination = _NOT_TERMINATED
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    # The semantics of future.Future.cancel and future.Future.cancelled are
 | 
	
		
			
				|  |  | +    # slightly wonky, so they have to be tracked separately from the rest of the
 | 
	
		
			
				|  |  | +    # result of the RPC. This field tracks whether cancellation was requested
 | 
	
		
			
				|  |  | +    # prior to termination of the RPC
 | 
	
		
			
				|  |  | +    self._cancelled = False
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def set_operator_and_context(self, operator, operation_context):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      self._operator = operator
 | 
	
		
			
				|  |  | +      self._operation_context = operation_context
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def _down_completion(self):
 | 
	
		
			
				|  |  | +    if self._down_terminal_metadata.kind is _Transitory.Kind.NOT_YET_SEEN:
 | 
	
		
			
				|  |  | +      terminal_metadata = None
 | 
	
		
			
				|  |  | +      self._down_terminal_metadata = _GONE
 | 
	
		
			
				|  |  | +    elif self._down_terminal_metadata.kind is _Transitory.Kind.PRESENT:
 | 
	
		
			
				|  |  | +      terminal_metadata = self._down_terminal_metadata.value
 | 
	
		
			
				|  |  | +      self._down_terminal_metadata = _GONE
 | 
	
		
			
				|  |  | +    else:
 | 
	
		
			
				|  |  | +      terminal_metadata = None
 | 
	
		
			
				|  |  | +    if self._down_code.kind is _Transitory.Kind.NOT_YET_SEEN:
 | 
	
		
			
				|  |  | +      code = None
 | 
	
		
			
				|  |  | +      self._down_code = _GONE
 | 
	
		
			
				|  |  | +    elif self._down_code.kind is _Transitory.Kind.PRESENT:
 | 
	
		
			
				|  |  | +      code = self._down_code.value
 | 
	
		
			
				|  |  | +      self._down_code = _GONE
 | 
	
		
			
				|  |  | +    else:
 | 
	
		
			
				|  |  | +      code = None
 | 
	
		
			
				|  |  | +    if self._down_details.kind is _Transitory.Kind.NOT_YET_SEEN:
 | 
	
		
			
				|  |  | +      details = None
 | 
	
		
			
				|  |  | +      self._down_details = _GONE
 | 
	
		
			
				|  |  | +    elif self._down_details.kind is _Transitory.Kind.PRESENT:
 | 
	
		
			
				|  |  | +      details = self._down_details.value
 | 
	
		
			
				|  |  | +      self._down_details = _GONE
 | 
	
		
			
				|  |  | +    else:
 | 
	
		
			
				|  |  | +      details = None
 | 
	
		
			
				|  |  | +    return utilities.completion(terminal_metadata, code, details)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def _set_outcome(self, outcome):
 | 
	
		
			
				|  |  | +    if not self._termination.terminated:
 | 
	
		
			
				|  |  | +      self._operator = _DUMMY_OPERATOR
 | 
	
		
			
				|  |  | +      self._operation_context = None
 | 
	
		
			
				|  |  | +      self._down_initial_metadata = _GONE
 | 
	
		
			
				|  |  | +      self._down_payload = None
 | 
	
		
			
				|  |  | +      self._down_terminal_metadata = _GONE
 | 
	
		
			
				|  |  | +      self._down_code = _GONE
 | 
	
		
			
				|  |  | +      self._down_details = _GONE
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +      if self._up_initial_metadata.kind is _Awaited.Kind.NOT_YET_ARRIVED:
 | 
	
		
			
				|  |  | +        initial_metadata = None
 | 
	
		
			
				|  |  | +      else:
 | 
	
		
			
				|  |  | +        initial_metadata = self._up_initial_metadata.value
 | 
	
		
			
				|  |  | +      if self._up_completion.kind is _Awaited.Kind.NOT_YET_ARRIVED:
 | 
	
		
			
				|  |  | +        terminal_metadata, code, details = None, None, None
 | 
	
		
			
				|  |  | +      else:
 | 
	
		
			
				|  |  | +        terminal_metadata = self._up_completion.value.terminal_metadata
 | 
	
		
			
				|  |  | +        code = self._up_completion.value.code
 | 
	
		
			
				|  |  | +        details = self._up_completion.value.message
 | 
	
		
			
				|  |  | +      self._termination = _OPERATION_OUTCOME_TO_TERMINATION_CONSTRUCTOR[
 | 
	
		
			
				|  |  | +          outcome](initial_metadata, terminal_metadata, code, details)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +      self._condition.notify_all()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    return self._termination
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def advance(
 | 
	
		
			
				|  |  | +      self, initial_metadata=None, payload=None, completion=None,
 | 
	
		
			
				|  |  | +      allowance=None):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      if initial_metadata is not None:
 | 
	
		
			
				|  |  | +        self._up_initial_metadata = _Awaited(
 | 
	
		
			
				|  |  | +            _Awaited.Kind.ARRIVED, initial_metadata)
 | 
	
		
			
				|  |  | +      if payload is not None:
 | 
	
		
			
				|  |  | +        if self._up_initial_metadata.kind is _Awaited.Kind.NOT_YET_ARRIVED:
 | 
	
		
			
				|  |  | +          self._up_initial_metadata = _ARRIVED_AND_NONE
 | 
	
		
			
				|  |  | +        self._up_payload = payload
 | 
	
		
			
				|  |  | +        self._up_allowance -= 1
 | 
	
		
			
				|  |  | +      if completion is not None:
 | 
	
		
			
				|  |  | +        if self._up_initial_metadata.kind is _Awaited.Kind.NOT_YET_ARRIVED:
 | 
	
		
			
				|  |  | +          self._up_initial_metadata = _ARRIVED_AND_NONE
 | 
	
		
			
				|  |  | +        self._up_completion = _Awaited(
 | 
	
		
			
				|  |  | +            _Awaited.Kind.ARRIVED, completion)
 | 
	
		
			
				|  |  | +      if allowance is not None:
 | 
	
		
			
				|  |  | +        if self._down_payload is not None:
 | 
	
		
			
				|  |  | +          self._operator.advance(payload=self._down_payload)
 | 
	
		
			
				|  |  | +          self._down_payload = None
 | 
	
		
			
				|  |  | +          self._down_allowance += allowance - 1
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          self._down_allowance += allowance
 | 
	
		
			
				|  |  | +      self._condition.notify_all()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def cancel(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      if self._operation_context is not None:
 | 
	
		
			
				|  |  | +        self._operation_context.cancel()
 | 
	
		
			
				|  |  | +        self._cancelled = True
 | 
	
		
			
				|  |  | +      return False
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def cancelled(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      return self._cancelled
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def running(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      return not self._termination.terminated
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def done(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      return self._termination.terminated
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def result(self, timeout=None):
 | 
	
		
			
				|  |  | +    until = None if timeout is None else time.time() + timeout
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._termination.terminated:
 | 
	
		
			
				|  |  | +          if self._termination.abortion is None:
 | 
	
		
			
				|  |  | +            return self._up_payload
 | 
	
		
			
				|  |  | +          elif self._termination.abortion.kind is face.Abortion.Kind.CANCELLED:
 | 
	
		
			
				|  |  | +            raise future.CancelledError()
 | 
	
		
			
				|  |  | +          else:
 | 
	
		
			
				|  |  | +            raise self._termination.abortion_error  # pylint: disable=raising-bad-type
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          _wait_once_until(self._condition, until)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def exception(self, timeout=None):
 | 
	
		
			
				|  |  | +    until = None if timeout is None else time.time() + timeout
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._termination.terminated:
 | 
	
		
			
				|  |  | +          if self._termination.abortion is None:
 | 
	
		
			
				|  |  | +            return None
 | 
	
		
			
				|  |  | +          else:
 | 
	
		
			
				|  |  | +            return self._termination.abortion_error
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          _wait_once_until(self._condition, until)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def traceback(self, timeout=None):
 | 
	
		
			
				|  |  | +    until = None if timeout is None else time.time() + timeout
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._termination.terminated:
 | 
	
		
			
				|  |  | +          if self._termination.abortion_error is None:
 | 
	
		
			
				|  |  | +            return None
 | 
	
		
			
				|  |  | +          else:
 | 
	
		
			
				|  |  | +            abortion_error = self._termination.abortion_error
 | 
	
		
			
				|  |  | +            break
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          _wait_once_until(self._condition, until)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    try:
 | 
	
		
			
				|  |  | +      raise abortion_error
 | 
	
		
			
				|  |  | +    except face.AbortionError:
 | 
	
		
			
				|  |  | +      return sys.exc_info()[2]
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def add_done_callback(self, fn):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      if self._operation_context is not None:
 | 
	
		
			
				|  |  | +        outcome = self._operation_context.add_termination_callback(
 | 
	
		
			
				|  |  | +            _done_callback_as_operation_termination_callback(fn, self))
 | 
	
		
			
				|  |  | +        if outcome is None:
 | 
	
		
			
				|  |  | +          return
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          self._set_outcome(outcome)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    fn(self)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def consume(self, value):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._termination.terminated:
 | 
	
		
			
				|  |  | +          return
 | 
	
		
			
				|  |  | +        elif 0 < self._down_allowance:
 | 
	
		
			
				|  |  | +          self._operator.advance(payload=value)
 | 
	
		
			
				|  |  | +          self._down_allowance -= 1
 | 
	
		
			
				|  |  | +          return
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          self._condition.wait()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def terminate(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      if self._termination.terminated:
 | 
	
		
			
				|  |  | +        return
 | 
	
		
			
				|  |  | +      elif self._down_code.kind is _Transitory.Kind.GONE:
 | 
	
		
			
				|  |  | +        # Conform to specified idempotence of terminate by ignoring extra calls.
 | 
	
		
			
				|  |  | +        return
 | 
	
		
			
				|  |  | +      else:
 | 
	
		
			
				|  |  | +        completion = self._down_completion()
 | 
	
		
			
				|  |  | +        self._operator.advance(completion=completion)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def consume_and_terminate(self, value):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._termination.terminated:
 | 
	
		
			
				|  |  | +          return
 | 
	
		
			
				|  |  | +        elif 0 < self._down_allowance:
 | 
	
		
			
				|  |  | +          completion = self._down_completion()
 | 
	
		
			
				|  |  | +          self._operator.advance(payload=value, completion=completion)
 | 
	
		
			
				|  |  | +          return
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          self._condition.wait()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def __iter__(self):
 | 
	
		
			
				|  |  | +    return self
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def next(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._termination.abortion_error is not None:
 | 
	
		
			
				|  |  | +          raise self._termination.abortion_error
 | 
	
		
			
				|  |  | +        elif self._up_payload is not None:
 | 
	
		
			
				|  |  | +          payload = self._up_payload
 | 
	
		
			
				|  |  | +          self._up_payload = None
 | 
	
		
			
				|  |  | +          if self._up_completion.kind is _Awaited.Kind.NOT_YET_ARRIVED:
 | 
	
		
			
				|  |  | +            self._operator.advance(allowance=1)
 | 
	
		
			
				|  |  | +          return payload
 | 
	
		
			
				|  |  | +        elif self._up_completion.kind is _Awaited.Kind.ARRIVED:
 | 
	
		
			
				|  |  | +          raise StopIteration()
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          self._condition.wait()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def is_active(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      return not self._termination.terminated
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def time_remaining(self):
 | 
	
		
			
				|  |  | +    if self._operation_context is None:
 | 
	
		
			
				|  |  | +      return 0
 | 
	
		
			
				|  |  | +    else:
 | 
	
		
			
				|  |  | +      return self._operation_context.time_remaining()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def add_abortion_callback(self, abortion_callback):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      if self._operation_context is None:
 | 
	
		
			
				|  |  | +        return self._termination.abortion
 | 
	
		
			
				|  |  | +      else:
 | 
	
		
			
				|  |  | +        outcome = self._operation_context.add_termination_callback(
 | 
	
		
			
				|  |  | +            _abortion_callback_as_operation_termination_callback(
 | 
	
		
			
				|  |  | +                abortion_callback, self.set_outcome))
 | 
	
		
			
				|  |  | +        if outcome is not None:
 | 
	
		
			
				|  |  | +          return self._set_outcome(outcome).abortion
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          return self._termination.abortion
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def initial_metadata(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._up_initial_metadata.kind is _Awaited.Kind.ARRIVED:
 | 
	
		
			
				|  |  | +          return self._up_initial_metadata.value
 | 
	
		
			
				|  |  | +        elif self._termination.terminated:
 | 
	
		
			
				|  |  | +          return None
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          self._condition.wait()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def terminal_metadata(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._up_completion.kind is _Awaited.Kind.ARRIVED:
 | 
	
		
			
				|  |  | +          return self._up_completion.value.terminal_metadata
 | 
	
		
			
				|  |  | +        elif self._termination.terminated:
 | 
	
		
			
				|  |  | +          return None
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          self._condition.wait()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def code(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._up_completion.kind is _Awaited.Kind.ARRIVED:
 | 
	
		
			
				|  |  | +          return self._up_completion.value.code
 | 
	
		
			
				|  |  | +        elif self._termination.terminated:
 | 
	
		
			
				|  |  | +          return None
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          self._condition.wait()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def details(self):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      while True:
 | 
	
		
			
				|  |  | +        if self._up_completion.kind is _Awaited.Kind.ARRIVED:
 | 
	
		
			
				|  |  | +          return self._up_completion.value.message
 | 
	
		
			
				|  |  | +        elif self._termination.terminated:
 | 
	
		
			
				|  |  | +          return None
 | 
	
		
			
				|  |  | +        else:
 | 
	
		
			
				|  |  | +          self._condition.wait()
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def set_initial_metadata(self, initial_metadata):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      if (self._down_initial_metadata.kind is not
 | 
	
		
			
				|  |  | +          _Transitory.Kind.NOT_YET_SEEN):
 | 
	
		
			
				|  |  | +        raise ValueError(_CANNOT_SET_INITIAL_METADATA)
 | 
	
		
			
				|  |  | +      else:
 | 
	
		
			
				|  |  | +        self._down_initial_metadata = _GONE
 | 
	
		
			
				|  |  | +        self._operator.advance(initial_metadata=initial_metadata)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def set_terminal_metadata(self, terminal_metadata):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      if (self._down_terminal_metadata.kind is not
 | 
	
		
			
				|  |  | +          _Transitory.Kind.NOT_YET_SEEN):
 | 
	
		
			
				|  |  | +        raise ValueError(_CANNOT_SET_TERMINAL_METADATA)
 | 
	
		
			
				|  |  | +      else:
 | 
	
		
			
				|  |  | +        self._down_terminal_metadata = _Transitory(
 | 
	
		
			
				|  |  | +            _Transitory.Kind.PRESENT, terminal_metadata)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def set_code(self, code):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      if self._down_code.kind is not _Transitory.Kind.NOT_YET_SEEN:
 | 
	
		
			
				|  |  | +        raise ValueError(_CANNOT_SET_CODE)
 | 
	
		
			
				|  |  | +      else:
 | 
	
		
			
				|  |  | +        self._down_code = _Transitory(_Transitory.Kind.PRESENT, code)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def set_details(self, details):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      if self._down_details.kind is not _Transitory.Kind.NOT_YET_SEEN:
 | 
	
		
			
				|  |  | +        raise ValueError(_CANNOT_SET_DETAILS)
 | 
	
		
			
				|  |  | +      else:
 | 
	
		
			
				|  |  | +        self._down_details = _Transitory(_Transitory.Kind.PRESENT, details)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  def set_outcome(self, outcome):
 | 
	
		
			
				|  |  | +    with self._condition:
 | 
	
		
			
				|  |  | +      return self._set_outcome(outcome)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +def pool_wrap(behavior, operation_context):
 | 
	
		
			
				|  |  | +  """Wraps an operation-related behavior so that it may be called in a pool.
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  Args:
 | 
	
		
			
				|  |  | +    behavior: A callable related to carrying out an operation.
 | 
	
		
			
				|  |  | +    operation_context: A base_interfaces.OperationContext for the operation.
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +  Returns:
 | 
	
		
			
				|  |  | +    A callable that when called carries out the behavior of the given callable
 | 
	
		
			
				|  |  | +      and handles whatever exceptions it raises appropriately.
 | 
	
		
			
				|  |  | +  """
 | 
	
		
			
				|  |  | +  def translation(*args):
 | 
	
		
			
				|  |  | +    try:
 | 
	
		
			
				|  |  | +      behavior(*args)
 | 
	
		
			
				|  |  | +    except (
 | 
	
		
			
				|  |  | +        abandonment.Abandoned,
 | 
	
		
			
				|  |  | +        face.CancellationError,
 | 
	
		
			
				|  |  | +        face.ExpirationError,
 | 
	
		
			
				|  |  | +        face.LocalShutdownError,
 | 
	
		
			
				|  |  | +        face.RemoteShutdownError,
 | 
	
		
			
				|  |  | +        face.NetworkError,
 | 
	
		
			
				|  |  | +        face.RemoteError,
 | 
	
		
			
				|  |  | +    ) as e:
 | 
	
		
			
				|  |  | +      if operation_context.outcome() is None:
 | 
	
		
			
				|  |  | +        operation_context.fail(e)
 | 
	
		
			
				|  |  | +    except Exception as e:
 | 
	
		
			
				|  |  | +      operation_context.fail(e)
 | 
	
		
			
				|  |  | +  return callable_util.with_exceptions_logged(
 | 
	
		
			
				|  |  | +      translation, _INTERNAL_ERROR_LOG_MESSAGE)
 |