Propagate custom events in PendingAuthListener, TransmitStatusRuntimeExceptionInterceptor, and OpenTelemetryTracingModule.

- binder: Implement onEvent in PendingAuthListener to buffer and replay
  custom events to the delegate once auth completes, preventing events
  from being dropped.
- util: Handle onEvent in TransmitStatusRuntimeExceptionInterceptor
  listener wrapper to catch StatusRuntimeException and close the call.
  Serialize triggerEvent on SerializingServerCall's executor.
- opentelemetry: Implement onEvent in ContextServerCallListener to
  attach OpenTelemetry trace context and scope during delegate invocation.
- Add unit tests for all updated implementations.

TAG=agy
CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be

Author: kannanjgithub

binder/src/main/java/io/grpc/binder/internal/PendingAuthListener.java

@@ -88,6 +88,12 @@ public void onReady() {
     maybeRunPendingSteps();
   }
 
+  @Override
+  public void onEvent(Object event) {
+    pendingSteps.offer(delegate -> delegate.onEvent(event));
+    maybeRunPendingSteps();
+  }
+
   /**
    * Similar to Java8's {@link java.util.function.Consumer}, but redeclared in order to support
    * Android SDK 21.

binder/src/test/java/io/grpc/binder/internal/PendingAuthListenerTest.java

@@ -45,6 +45,7 @@ public void setUp() {
   public void onCallbacks_noOpBeforeStartCall() {
     listener.onReady();
     listener.onMessage("foo");
+    listener.onEvent("bar");
     listener.onHalfClose();
     listener.onComplete();
 
@@ -54,16 +55,19 @@ public void onCallbacks_noOpBeforeStartCall() {
   @Test
   public void onCallbacks_runsPendingCallbacksAfterStartCall() {
     String message = "foo";
+    String event = "bar";
 
     // Act 1
     listener.onReady();
     listener.onMessage(message);
+    listener.onEvent(event);
     listener.startCall(call, headers, next);
 
     // Assert 1
     InOrder order = Mockito.inOrder(delegate);
     order.verify(delegate).onReady();
     order.verify(delegate).onMessage(message);
+    order.verify(delegate).onEvent(event);
 
     // Act 2
     listener.onHalfClose();

opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryTracingModule.java

@@ -452,6 +452,13 @@ public void onReady() {
         delegate().onReady();
       }
     }
+
+    @Override
+    public void onEvent(Object event) {
+      try (Scope scope = context.makeCurrent()) {
+        delegate().onEvent(event);
+      }
+    }
   }
 
   @VisibleForTesting

opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryTracingModuleTest.java

@@ -948,6 +948,11 @@ public void onCancel() {
       public void onComplete() {
         callbackSpan.set(Span.fromContext(Context.current()));
       }
+
+      @Override
+      public void onEvent(Object event) {
+        callbackSpan.set(Span.fromContext(Context.current()));
+      }
     };
     ServerInterceptor interceptor = tracingModule.getServerSpanPropagationInterceptor();
     @SuppressWarnings("unchecked")
@@ -967,6 +972,8 @@ public void onComplete() {
     assertEquals(callbackSpan.get(), Span.getInvalid());
     listener.onComplete();
     assertEquals(callbackSpan.get(), Span.getInvalid());
+    listener.onEvent(new Object());
+    assertEquals(callbackSpan.get(), Span.getInvalid());
 
     Span parentSpan = tracerRule.spanBuilder("parent-span").startSpan();
     io.grpc.Context context = io.grpc.Context.current().withValue(
@@ -990,6 +997,9 @@ public void onComplete() {
       listener.onComplete();
       assertEquals(callbackSpan.get().getSpanContext().getTraceId(),
           parentSpan.getSpanContext().getTraceId());
+      listener.onEvent(new Object());
+      assertEquals(callbackSpan.get().getSpanContext().getTraceId(),
+          parentSpan.getSpanContext().getTraceId());
     } finally {
       context.detach(previous);
     }

util/src/main/java/io/grpc/util/TransmitStatusRuntimeExceptionInterceptor.java

@@ -104,6 +104,15 @@ public void onReady() {
         }
       }
 
+      @Override
+      public void onEvent(Object event) {
+        try {
+          super.onEvent(event);
+        } catch (StatusRuntimeException e) {
+          closeWithException(e);
+        }
+      }
+
       private void closeWithException(StatusRuntimeException t) {
         Metadata metadata = t.getTrailers();
         if (metadata == null) {
@@ -276,5 +285,15 @@ public void run() {
         throw new RuntimeException(ERROR_MSG, e);
       }
     }
+
+    @Override
+    public void triggerEvent(final Object event) {
+      serializingExecutor.execute(new Runnable() {
+        @Override
+        public void run() {
+          SerializingServerCall.super.triggerEvent(event);
+        }
+      });
+    }
   }
 }

util/src/test/java/io/grpc/util/UtilServerInterceptorsTest.java

@@ -104,6 +104,11 @@ public void onComplete() {
       public void onReady() {
         throw exception;
       }
+
+      @Override
+      public void onEvent(Object event) {
+        throw exception;
+      }
     };
 
     ServerServiceDefinition intercepted = ServerInterceptors.intercept(
@@ -116,7 +121,36 @@ public void onReady() {
     getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onComplete();
     getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onHalfClose();
     getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onReady();
-    assertEquals(5, call.numCloses);
+    getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers)
+        .onEvent(new Object());
+    assertEquals(6, call.numCloses);
+  }
+
+  @Test
+  public void statusRuntimeExceptionTransmitter_serializingServerCall_triggerEvent() {
+    final java.util.concurrent.atomic.AtomicReference<Object> eventRef =
+        new java.util.concurrent.atomic.AtomicReference<>();
+    FakeServerCall<Void, Void> call = new FakeServerCall<Void, Void>(Status.OK, new Metadata()) {
+      @Override
+      public void triggerEvent(Object event) {
+        eventRef.set(event);
+      }
+    };
+    final java.util.concurrent.atomic.AtomicReference<ServerCall<Void, Void>> interceptedCall =
+        new java.util.concurrent.atomic.AtomicReference<>();
+    listener = new VoidCallListener() {
+      @Override
+      public void onCall(ServerCall<Void, Void> call, Metadata headers) {
+        interceptedCall.set(call);
+      }
+    };
+    ServerServiceDefinition intercepted = ServerInterceptors.intercept(
+        serviceDefinition,
+        Arrays.asList(TransmitStatusRuntimeExceptionInterceptor.instance()));
+    getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers);
+    Object testEvent = new Object();
+    interceptedCall.get().triggerEvent(testEvent);
+    assertEquals(testEvent, eventRef.get());
   }
 
   @Test