api: Implement custom events framework in gRPC-Java server

api/src/main/java/io/grpc/Contexts.java

@@ -118,6 +118,16 @@ public void onReady() {
         context.detach(previous);
       }
     }
+
+    @Override
+    public void onEvent(Object event) {
+      Context previous = context.attach();
+      try {
+        super.onEvent(event);
+      } finally {
+        context.detach(previous);
+      }
+    }
   }
 
   /**

api/src/main/java/io/grpc/PartialForwardingServerCall.java

@@ -87,6 +87,11 @@ public SecurityLevel getSecurityLevel() {
     return delegate().getSecurityLevel();
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    delegate().triggerEvent(event);
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this).add("delegate", delegate()).toString();

api/src/main/java/io/grpc/PartialForwardingServerCallListener.java

@@ -50,6 +50,11 @@ public void onReady() {
     delegate().onReady();
   }
 
+  @Override
+  public void onEvent(Object event) {
+    delegate().onEvent(event);
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this).add("delegate", delegate()).toString();

api/src/main/java/io/grpc/ServerCall.java

@@ -100,6 +100,20 @@ public void onComplete() {}
      * <em>another</em> {@code onReady()} callback.
      */
     public void onReady() {}
+
+    /**
+     * A custom event has been triggered by the call.
+     *
+     * <p>This callback is guaranteed to run on the call's executor, serialized with other
+     * callbacks (like {@link #onMessage}, {@link #onHalfClose}). This means the implementation
+     * does not need internal synchronization to access call-specific state.
+     *
+     * @param event the triggered event.
+     */
+    @ExperimentalApi("https://github.com/grpc/grpc-java/issues/12979")
+    public void onEvent(Object event) {
+      // Default no-op
+    }
   }
 
   /**
@@ -262,6 +276,20 @@ public String getAuthority() {
     return null;
   }
 
+  /**
+   * Triggers a custom event to be processed by the listener.
+   * The event will be delivered to {@link Listener#onEvent(Object)} on the call's executor.
+   *
+   * <p>This method is thread-safe and can be called from any thread. No events will be delivered
+   * after the RPC is cancelled or completed.
+   *
+   * @param event the event to trigger.
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/12979")
+  public void triggerEvent(Object event) {
+    // Default no-op
+  }
+
   /**
    * The {@link MethodDescriptor} for the call.
    */

api/src/test/java/io/grpc/ContextsTest.java

@@ -82,6 +82,11 @@ public void interceptCall_basic() {
         assertSame(uniqueContext, Context.current());
         methodCalls.add(5);
       }
+
+      @Override public void onEvent(Object event) {
+        assertSame(uniqueContext, Context.current());
+        methodCalls.add(6);
+      }
     };
     ServerCall.Listener<Object> wrapped = interceptCall(uniqueContext, call, headers,
         new ServerCallHandler<Object, Object>() {
@@ -101,7 +106,8 @@ public ServerCall.Listener<Object> startCall(
     wrapped.onCancel();
     wrapped.onComplete();
     wrapped.onReady();
-    assertEquals(Arrays.asList(1, 2, 3, 4, 5), methodCalls);
+    wrapped.onEvent(new Object());
+    assertEquals(Arrays.asList(1, 2, 3, 4, 5, 6), methodCalls);
     assertSame(origContext, Context.current());
   }
 
@@ -145,6 +151,10 @@ public void interceptCall_restoresIfListenerThrows() {
       @Override public void onReady() {
         throw new RuntimeException();
       }
+
+      @Override public void onEvent(Object event) {
+        throw new RuntimeException();
+      }
     };
     ServerCall.Listener<Object> wrapped = interceptCall(uniqueContext, call, headers,
         new ServerCallHandler<Object, Object>() {
@@ -180,6 +190,11 @@ public ServerCall.Listener<Object> startCall(
       fail("Exception expected");
     } catch (RuntimeException expected) {
     }
+    try {
+      wrapped.onEvent(new Object());
+      fail("Exception expected");
+    } catch (RuntimeException expected) {
+    }
     assertSame(origContext, Context.current());
   }
 

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

@@ -668,6 +668,17 @@ protected void deliverCloseAbnormal(Status status) {
       listener.closed(status);
     }
 
+    void triggerEvent(Object event) {
+      synchronized (this) {
+        if (isClosed()) {
+          return;
+        }
+        if (listener != null) {
+          listener.triggerEvent(event);
+        }
+      }
+    }
+
     @GuardedBy("this")
     void onCloseSent(Status status) {
       if (!isClosed()) {

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

@@ -175,6 +175,11 @@ public void setDecompressor(Decompressor decompressor) {
     // Ignore.
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    inbound.triggerEvent(event);
+  }
+
   @Override
   public void optimizeForDirectExecutor() {
     // Ignore.

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/main/java/io/grpc/binder/internal/SingleMessageServerStream.java

@@ -167,6 +167,11 @@ public void setDecompressor(Decompressor decompressor) {
     // Ignore.
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    inbound.triggerEvent(event);
+  }
+
   @Override
   public void optimizeForDirectExecutor() {
     // Ignore.

binder/src/test/java/io/grpc/binder/AsyncSecurityPoliciesTest.java

@@ -289,6 +289,7 @@ public ListenableFuture<Status> checkAuthorizationAsync(int uid) {
     ListenableFuture<Status> authFuture = asyncPolicy.checkAuthorizationAsync(SOME_UID);
     assertThat(awaitResult(settableUid)).isEqualTo(SOME_UID);
     authFuture.cancel(false);
+    executor.submit(() -> {}).get(10, TimeUnit.SECONDS);
 
     assertThat(delegateAuthFuture.isCancelled()).isTrue();
   }

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();

core/src/main/java/io/grpc/internal/AbstractServerStream.java

@@ -173,6 +173,16 @@ public final void setListener(ServerStreamListener serverStreamListener) {
     transportState().setListener(serverStreamListener);
   }
 
+  @Override
+  public final void triggerEvent(final Object event) {
+    transportState().runOnTransportThread(new Runnable() {
+      @Override
+      public void run() {
+        transportState().triggerEvent(event);
+      }
+    });
+  }
+
   @Override
   public StatsTraceContext statsTraceContext() {
     return statsTraceCtx;
@@ -259,6 +269,13 @@ public void deframerClosed(boolean hasPartialMessage) {
 
 
 
+    public final void triggerEvent(Object event) {
+      if (listenerClosed) {
+        return;
+      }
+      listener().triggerEvent(event);
+    }
+
     @Override
     protected ServerStreamListener listener() {
       return listener;

core/src/main/java/io/grpc/internal/ServerCallImpl.java

@@ -254,6 +254,14 @@ public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
     return method;
   }
 
+  @Override
+  public void triggerEvent(Object event) {
+    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.triggerEvent")) {
+      PerfMark.attachTag(tag);
+      stream.triggerEvent(event);
+    }
+  }
+
   @Override
   public SecurityLevel getSecurityLevel() {
     final Attributes attributes = getAttributes();
@@ -395,5 +403,16 @@ public void onReady() {
         listener.onReady();
       }
     }
+
+    @Override
+    public void triggerEvent(Object event) {
+      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.triggerEvent")) {
+        PerfMark.attachTag(call.tag);
+        if (call.cancelled) {
+          return;
+        }
+        listener.onEvent(event);
+      }
+    }
   }
 }

core/src/main/java/io/grpc/internal/ServerImpl.java

@@ -781,6 +781,9 @@ public void closed(Status status) {}
 
     @Override
     public void onReady() {}
+
+    @Override
+    public void triggerEvent(Object event) {}
   }
 
   /**
@@ -960,6 +963,34 @@ public void runInContext() {
         callExecutor.execute(new OnReady());
       }
     }
+
+    @Override
+    public void triggerEvent(final Object event) {
+      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.triggerEvent")) {
+        PerfMark.attachTag(tag);
+        final Link link = PerfMark.linkOut();
+
+        final class TriggerEvent extends ContextRunnable {
+          TriggerEvent() {
+            super(context);
+          }
+
+          @Override
+          public void runInContext() {
+            try (TaskCloseable ignore = PerfMark.traceTask("ServerCallListener(app).onEvent")) {
+              PerfMark.attachTag(tag);
+              PerfMark.linkIn(link);
+              getListener().triggerEvent(event);
+            } catch (Throwable t) {
+              internalClose(t);
+              throw t;
+            }
+          }
+        }
+
+        callExecutor.execute(new TriggerEvent());
+      }
+    }
   }
 
   @VisibleForTesting

core/src/main/java/io/grpc/internal/ServerStream.java

@@ -87,6 +87,12 @@ public interface ServerStream extends Stream {
    */
   void setListener(ServerStreamListener serverStreamListener);
 
+  /**
+   * Triggers a custom event. Implementations must ensure this is propagated to the
+   * listener on the transport thread.
+   */
+  void triggerEvent(Object event);
+
   /**
    * The context for recording stats and traces for this stream.
    */

core/src/main/java/io/grpc/internal/ServerStreamListener.java

@@ -42,4 +42,9 @@ public interface ServerStreamListener extends StreamListener {
    * @param status details about the remote closure
    */
   void closed(Status status);
+
+  /**
+   * Propagates a custom event to the listener. Must be called on the transport thread.
+   */
+  void triggerEvent(Object event);
 }

core/src/test/java/io/grpc/internal/AbstractServerStreamTest.java

@@ -361,6 +361,31 @@ public void close_sendTrailersClearsReservedFields() {
     assertEquals("bad", metadataCaptor.getValue().get(InternalStatus.MESSAGE_KEY));
   }
 
+  @Test
+  public void triggerEvent_propagatesToListener() {
+    ServerStreamListener listener = mock(ServerStreamListener.class);
+    stream.transportState().setListener(listener);
+
+    Object event = new Object();
+    stream.triggerEvent(event);
+
+    verify(listener).triggerEvent(event);
+  }
+
+  @Test
+  public void triggerEvent_ignoredAfterClose() {
+    ServerStreamListener listener = mock(ServerStreamListener.class);
+    stream.transportState().setListener(listener);
+
+    stream.close(Status.OK, new Metadata());
+    stream.transportState().complete();
+
+    Object event = new Object();
+    stream.triggerEvent(event);
+
+    verify(listener, never()).triggerEvent(any());
+  }
+
   @Test
   public void changeOnReadyThreshold() {
     stream.setListener(new ServerStreamListenerBase());
@@ -391,6 +416,9 @@ public void halfClosed() {}
 
     @Override
     public void closed(Status status) {}
+
+    @Override
+    public void triggerEvent(Object event) {}
   }
 
   private static class AbstractServerStreamBase extends AbstractServerStream {