Add code and message to base.NoSuchMethodError

This is part of support for applications being able to respond to RPCs
with unrecognized names with specific codes and messages.

10 files changed, 86 insertions, 36 deletions

diff --git a/src/python/grpcio/grpc/framework/core/_context.py b/src/python/grpcio/grpc/framework/core/_context.py
index 24a12b612e..76b3534530 100644
--- a/src/python/grpcio/grpc/framework/core/_context.py
+++ b/src/python/grpcio/grpc/framework/core/_context.py
@@ -60,7 +60,7 @@ class OperationContext(base.OperationContext):
     with self._lock:
       if self._termination_manager.outcome is None:
         self._termination_manager.abort(outcome)
-        self._transmission_manager.abort(outcome)
+        self._transmission_manager.abort(outcome, None, None)
         self._expiration_manager.terminate()
 
   def outcome(self):
diff --git a/src/python/grpcio/grpc/framework/core/_emission.py b/src/python/grpcio/grpc/framework/core/_emission.py
index 7c702ab2ce..2d7b2e2f10 100644
--- a/src/python/grpcio/grpc/framework/core/_emission.py
+++ b/src/python/grpcio/grpc/framework/core/_emission.py
@@ -82,7 +82,8 @@ class EmissionManager(_interfaces.EmissionManager):
             completion_present and self._completion_seen or
             allowance_present and allowance <= 0):
           self._termination_manager.abort(base.Outcome.LOCAL_FAILURE)
-          self._transmission_manager.abort(base.Outcome.LOCAL_FAILURE)
+          self._transmission_manager.abort(
+              base.Outcome.LOCAL_FAILURE, None, None)
           self._expiration_manager.terminate()
         else:
           self._initial_metadata_seen |= initial_metadata_present
diff --git a/src/python/grpcio/grpc/framework/core/_expiration.py b/src/python/grpcio/grpc/framework/core/_expiration.py
index d94bdf2d2b..d8690b3a02 100644
--- a/src/python/grpcio/grpc/framework/core/_expiration.py
+++ b/src/python/grpcio/grpc/framework/core/_expiration.py
@@ -73,7 +73,7 @@ class _ExpirationManager(_interfaces.ExpirationManager):
         if self._future is not None and index == self._index:
           self._future = None
           self._termination_manager.expire()
-          self._transmission_manager.abort(base.Outcome.EXPIRED)
+          self._transmission_manager.abort(base.Outcome.EXPIRED, None, None)
     return expire
 
   def start(self):
diff --git a/src/python/grpcio/grpc/framework/core/_ingestion.py b/src/python/grpcio/grpc/framework/core/_ingestion.py
index 59f7f8adc8..7b8127f3fc 100644
--- a/src/python/grpcio/grpc/framework/core/_ingestion.py
+++ b/src/python/grpcio/grpc/framework/core/_ingestion.py
@@ -31,6 +31,7 @@
 
 import abc
 import collections
+import enum
 
 from grpc.framework.core import _constants
 from grpc.framework.core import _interfaces
@@ -42,21 +43,31 @@ _CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE = 'Exception initializing ingestion!'
 _INGESTION_EXCEPTION_LOG_MESSAGE = 'Exception during ingestion!'
 
 
-class _SubscriptionCreation(collections.namedtuple(
-    '_SubscriptionCreation', ('subscription', 'remote_error', 'abandoned'))):
+class _SubscriptionCreation(
+    collections.namedtuple(
+        '_SubscriptionCreation',
+        ('kind', 'subscription', 'code', 'message',))):
   """A sum type for the outcome of ingestion initialization.
 
-  Either subscription will be non-None, remote_error will be True, or abandoned
-  will be True.
-
   Attributes:
-    subscription: A base.Subscription describing the customer's interest in
-      operation values from the other side.
-    remote_error: A boolean indicating that the subscription could not be
-      created due to an error on the remote side of the operation.
-    abandoned: A boolean indicating that subscription creation was abandoned.
+    kind: A Kind value coarsely indicating how subscription creation completed.
+    subscription: The created subscription. Only present if kind is
+      Kind.SUBSCRIPTION.
+    code: A code value to be sent to the other side of the operation along with
+      an indication that the operation is being aborted due to an error on the
+      remote side of the operation. Only present if kind is Kind.REMOTE_ERROR.
+    message: A message value to be sent to the other side of the operation
+      along with an indication that the operation is being aborted due to an
+      error on the remote side of the operation. Only present if kind is
+      Kind.REMOTE_ERROR.
   """
 
+  @enum.unique
+  class Kind(enum.Enum):
+    SUBSCRIPTION = 'subscription'
+    REMOTE_ERROR = 'remote error'
+    ABANDONED = 'abandoned'
+
 
 class _SubscriptionCreator(object):
   """Common specification of subscription-creating behavior."""
@@ -101,12 +112,15 @@ class _ServiceSubscriptionCreator(_SubscriptionCreator):
     try:
       subscription = self._servicer.service(
           group, method, self._operation_context, self._output_operator)
-    except base.NoSuchMethodError:
-      return _SubscriptionCreation(None, True, False)
+    except base.NoSuchMethodError as e:
+      return _SubscriptionCreation(
+          _SubscriptionCreation.Kind.REMOTE_ERROR, None, e.code, e.message)
     except abandonment.Abandoned:
-      return _SubscriptionCreation(None, False, True)
+      return _SubscriptionCreation(
+          _SubscriptionCreation.Kind.ABANDONED, None, None, None)
     else:
-      return _SubscriptionCreation(subscription, False, False)
+      return _SubscriptionCreation(
+          _SubscriptionCreation.Kind.SUBSCRIPTION, subscription, None, None)
 
 
 def _wrap(behavior):
@@ -176,10 +190,10 @@ class _IngestionManager(_interfaces.IngestionManager):
     self._pending_payloads = None
     self._pending_completion = None
 
-  def _abort_and_notify(self, outcome):
+  def _abort_and_notify(self, outcome, code, message):
     self._abort_internal_only()
     self._termination_manager.abort(outcome)
-    self._transmission_manager.abort(outcome)
+    self._transmission_manager.abort(outcome, code, message)
     self._expiration_manager.terminate()
 
   def _operator_next(self):
@@ -236,12 +250,12 @@ class _IngestionManager(_interfaces.IngestionManager):
         else:
           with self._lock:
             if self._termination_manager.outcome is None:
-              self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+              self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
             return
       else:
         with self._lock:
           if self._termination_manager.outcome is None:
-            self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+            self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
           return
 
   def _operator_post_create(self, subscription):
@@ -260,20 +274,22 @@ class _IngestionManager(_interfaces.IngestionManager):
 
   def _create(self, subscription_creator, group, name):
     outcome = callable_util.call_logging_exceptions(
-        subscription_creator.create, _CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE,
-        group, name)
+        subscription_creator.create,
+        _CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE, group, name)
     if outcome.return_value is None:
       with self._lock:
         if self._termination_manager.outcome is None:
-          self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
-    elif outcome.return_value.abandoned:
+          self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
+    elif outcome.return_value.kind is _SubscriptionCreation.Kind.ABANDONED:
       with self._lock:
         if self._termination_manager.outcome is None:
-          self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
-    elif outcome.return_value.remote_error:
+          self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
+    elif outcome.return_value.kind is _SubscriptionCreation.Kind.REMOTE_ERROR:
+      code = outcome.return_value.code
+      message = outcome.return_value.message
       with self._lock:
         if self._termination_manager.outcome is None:
-          self._abort_and_notify(base.Outcome.REMOTE_FAILURE)
+          self._abort_and_notify(base.Outcome.REMOTE_FAILURE, code, message)
     elif outcome.return_value.subscription.kind is base.Subscription.Kind.FULL:
       self._operator_post_create(outcome.return_value.subscription)
     else:
diff --git a/src/python/grpcio/grpc/framework/core/_interfaces.py b/src/python/grpcio/grpc/framework/core/_interfaces.py
index a626b9f767..deb5f34f9b 100644
--- a/src/python/grpcio/grpc/framework/core/_interfaces.py
+++ b/src/python/grpcio/grpc/framework/core/_interfaces.py
@@ -155,13 +155,19 @@ class TransmissionManager(object):
     raise NotImplementedError()
 
   @abc.abstractmethod
-  def abort(self, outcome):
+  def abort(self, outcome, code, message):
     """Indicates that the operation has aborted.
 
     Args:
       outcome: An interfaces.Outcome for the operation. If None, indicates that
         the operation abortion should not be communicated to the other side of
         the operation.
+      code: A code value to communicate to the other side of the operation
+        along with indication of operation abortion. May be None, and has no
+        effect if outcome is None.
+      message: A message value to communicate to the other side of the
+        operation along with indication of operation abortion. May be None, and
+        has no effect if outcome is None.
     """
     raise NotImplementedError()
 
diff --git a/src/python/grpcio/grpc/framework/core/_operation.py b/src/python/grpcio/grpc/framework/core/_operation.py
index d20e40a53d..cc873c03f9 100644
--- a/src/python/grpcio/grpc/framework/core/_operation.py
+++ b/src/python/grpcio/grpc/framework/core/_operation.py
@@ -79,7 +79,7 @@ class _EasyOperation(_interfaces.Operation):
     with self._lock:
       if self._termination_manager.outcome is None:
         self._termination_manager.abort(outcome)
-        self._transmission_manager.abort(outcome)
+        self._transmission_manager.abort(outcome, None, None)
         self._expiration_manager.terminate()
 
 
diff --git a/src/python/grpcio/grpc/framework/core/_reception.py b/src/python/grpcio/grpc/framework/core/_reception.py
index 0858f64ff6..1cebe3874b 100644
--- a/src/python/grpcio/grpc/framework/core/_reception.py
+++ b/src/python/grpcio/grpc/framework/core/_reception.py
@@ -73,7 +73,7 @@ class ReceptionManager(_interfaces.ReceptionManager):
     self._aborted = True
     if self._termination_manager.outcome is None:
       self._termination_manager.abort(outcome)
-      self._transmission_manager.abort(None)
+      self._transmission_manager.abort(None, None, None)
       self._expiration_manager.terminate()
 
   def _sequence_failure(self, ticket):
diff --git a/src/python/grpcio/grpc/framework/core/_transmission.py b/src/python/grpcio/grpc/framework/core/_transmission.py
index 03644f4d49..efef87dd4c 100644
--- a/src/python/grpcio/grpc/framework/core/_transmission.py
+++ b/src/python/grpcio/grpc/framework/core/_transmission.py
@@ -104,9 +104,13 @@ class TransmissionManager(_interfaces.TransmissionManager):
           return None
         else:
           self._abortion_outcome = None
+          if self._completion is None:
+            code, message = None, None
+          else:
+            code, message = self._completion.code, self._completion.message
           return links.Ticket(
               self._operation_id, self._lowest_unused_sequence_number, None,
-              None, None, None, None, None, None, None, None, None,
+              None, None, None, None, None, None, None, code, message,
               termination, None)
 
     action = False
@@ -277,7 +281,7 @@ class TransmissionManager(_interfaces.TransmissionManager):
     self._remote_complete = True
     self._local_allowance = 0
 
-  def abort(self, outcome):
+  def abort(self, outcome, code, message):
     """See _interfaces.TransmissionManager.abort for specification."""
     if self._transmitting:
       self._aborted, self._abortion_outcome = True, outcome
@@ -287,8 +291,12 @@ class TransmissionManager(_interfaces.TransmissionManager):
         termination = _constants.ABORTION_OUTCOME_TO_TICKET_TERMINATION[
             outcome]
         if termination is not None:
+          if self._completion is None:
+            code, message = None, None
+          else:
+            code, message = self._completion.code, self._completion.message
           ticket = links.Ticket(
               self._operation_id, self._lowest_unused_sequence_number, None,
-              None, None, None, None, None, None, None, None, None,
+              None, None, None, None, None, None, None, code, message,
               termination, None)
           self._transmit(ticket)
diff --git a/src/python/grpcio/grpc/framework/interfaces/base/base.py b/src/python/grpcio/grpc/framework/interfaces/base/base.py
index 76e0a5bdae..bc52efb4c5 100644
--- a/src/python/grpcio/grpc/framework/interfaces/base/base.py
+++ b/src/python/grpcio/grpc/framework/interfaces/base/base.py
@@ -47,7 +47,26 @@ from grpc.framework.foundation import abandonment  # pylint: disable=unused-impo
 
 
 class NoSuchMethodError(Exception):
-  """Indicates that an unrecognized operation has been called."""
+  """Indicates that an unrecognized operation has been called.
+
+  Attributes:
+    code: A code value to communicate to the other side of the operation along
+      with indication of operation termination. May be None.
+    details: A details value to communicate to the other side of the operation
+      along with indication of operation termination. May be None.
+  """
+
+  def __init__(self, code, details):
+    """Constructor.
+
+    Args:
+      code: A code value to communicate to the other side of the operation
+        along with indication of operation termination. May be None.
+      details: A details value to communicate to the other side of the
+        operation along with indication of operation termination. May be None.
+    """
+    self.code = code
+    self.details = details
 
 
 @enum.unique
diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
index 5c8b176da4..87332cf612 100644
--- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
+++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py
@@ -134,7 +134,7 @@ class _Servicer(base.Servicer):
       if group != self._group or method != self._method:
         controller.fail(
             '%s != %s or %s != %s' % (group, self._group, method, self._method))
-        raise base.NoSuchMethodError()
+        raise base.NoSuchMethodError(None, None)
       else:
         operator = _Operator(
             controller, controller.on_service_advance, self._pool,