api,core: Add ServerInterceptor2 interface

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

@@ -77,6 +77,12 @@ public T intercept(ServerInterceptor interceptor) {
     return thisT();
   }
 
+  @Override
+  public T intercept(ServerInterceptor2 interceptor) {
+    delegate().intercept(interceptor);
+    return thisT();
+  }
+
   @Override
   public T addTransportFilter(ServerTransportFilter filter) {
     delegate().addTransportFilter(filter);

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

@@ -90,10 +90,11 @@ public static ServerBuilder<?> forPort(int port) {
   public abstract T addService(BindableService bindableService);
 
   /**
-   * Adds a {@link ServerInterceptor} that is run for all services on the server.  Interceptors
-   * added through this method always run before per-service interceptors added through {@link
-   * ServerInterceptors}.  Interceptors run in the reverse order in which they are added, just as
-   * with consecutive calls to {@code ServerInterceptors.intercept()}.
+   * Adds a {@link ServerInterceptor} that is run for all services on the server. Interceptors added
+   * through this method always run before per-service interceptors added through {@link
+   * ServerInterceptors}. Interceptors run in the reverse order in which they are added (either
+   * through this method or {@link #intercept(ServerInterceptor2)}), just as with consecutive calls
+   * to {@code ServerInterceptors.intercept()}.
    *
    * @param interceptor the all-service interceptor
    * @return this
@@ -104,6 +105,22 @@ public T intercept(ServerInterceptor interceptor) {
     throw new UnsupportedOperationException();
   }
 
+  /**
+   * Adds a {@link ServerInterceptor2} that is run for all services on the server. Interceptors
+   * added through this method always run before per-service interceptors added through {@link
+   * ServerInterceptors}. Interceptors run in the reverse order in which they are added (either
+   * through this method or {@link #intercept(ServerInterceptor)}), just as with consecutive calls
+   * to {@code ServerInterceptors.intercept()}.
+   *
+   * @param interceptor the all-service interceptor
+   * @return this
+   * @since 1.36.0
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/????")
+  public T intercept(ServerInterceptor2 interceptor) {
+    throw new UnsupportedOperationException();
+  }
+
   /**
    * Adds a {@link ServerTransportFilter}. The order of filters being added is the order they will
    * be executed.

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2020 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+/**
+ * Interface for intercepting server calls by modifying the {@link ServerMethodDefinition} after
+ * method lookup and before call dispatch. This provides a superset of the functionality offered by
+ * {@link ServerInterceptor}.
+ */
+@ExperimentalApi("https://github.com/grpc/grpc-java/issues/????")
+@ThreadSafe
+public interface ServerInterceptor2 {
+  /**
+   * Intercept and modify the {@link ServerMethodDefinition}.
+   *
+   * <p>The returned ServerMethodDefinition cannot be null and implementations must not throw.
+   */
+  <ReqT, RespT> ServerMethodDefinition<ReqT, RespT> interceptMethodDefinition(
+      ServerMethodDefinition<ReqT, RespT> method);
+}

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

@@ -303,4 +303,31 @@ public void onMessage(WReqT message) {
       }
     };
   }
+
+  /** Wrapper to convert ServerInterceptor to ServerInterceptor2. */
+  private static final class ServerInterceptorConverter implements ServerInterceptor2 {
+    private final ServerInterceptor wrappedInterceptor;
+
+    ServerInterceptorConverter(ServerInterceptor interceptor) {
+      wrappedInterceptor = interceptor;
+    }
+
+    @Override
+    public <ReqT, RespT> ServerMethodDefinition<ReqT, RespT> interceptMethodDefinition(
+        ServerMethodDefinition<ReqT, RespT> method) {
+      return method.withServerCallHandler(
+          InternalServerInterceptors.interceptCallHandler(
+              wrappedInterceptor, method.getServerCallHandler()));
+    }
+  }
+
+  /**
+   * Returns a {@link ServerInterceptor2} instance that wraps the given {@code ServerInterceptor}.
+   *
+   * @since 1.36.0
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/????")
+  public static ServerInterceptor2 serverInterceptorConverter(ServerInterceptor interceptor) {
+    return new ServerInterceptorConverter(interceptor);
+  }
 }

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

@@ -16,6 +16,9 @@
 
 package io.grpc;
 
+import java.util.Collections;
+import java.util.List;
+
 /**
  * Definition of a method exposed by a {@link Server}.
  *
@@ -24,11 +27,15 @@
 public final class ServerMethodDefinition<ReqT, RespT> {
   private final MethodDescriptor<ReqT, RespT> method;
   private final ServerCallHandler<ReqT, RespT> handler;
+  private final List<ServerStreamTracer.Factory> tracerFactories;
 
-  private ServerMethodDefinition(MethodDescriptor<ReqT, RespT> method,
-      ServerCallHandler<ReqT, RespT> handler) {
+  private ServerMethodDefinition(
+      MethodDescriptor<ReqT, RespT> method,
+      ServerCallHandler<ReqT, RespT> handler,
+      List<ServerStreamTracer.Factory> tracerFactories) {
     this.method = method;
     this.handler = handler;
+    this.tracerFactories = Collections.unmodifiableList(tracerFactories);
   }
 
   /**
@@ -39,9 +46,9 @@ private ServerMethodDefinition(MethodDescriptor<ReqT, RespT> method,
    * @return a new instance.
    */
   public static <ReqT, RespT> ServerMethodDefinition<ReqT, RespT> create(
-      MethodDescriptor<ReqT, RespT> method,
-      ServerCallHandler<ReqT, RespT> handler) {
-    return new ServerMethodDefinition<>(method, handler);
+      MethodDescriptor<ReqT, RespT> method, ServerCallHandler<ReqT, RespT> handler) {
+    return new ServerMethodDefinition<>(
+        method, handler, Collections.<ServerStreamTracer.Factory>emptyList());
   }
 
   /** The {@code MethodDescriptor} for this method. */
@@ -62,6 +69,29 @@ public ServerCallHandler<ReqT, RespT> getServerCallHandler() {
    */
   public ServerMethodDefinition<ReqT, RespT> withServerCallHandler(
       ServerCallHandler<ReqT, RespT> handler) {
-    return new ServerMethodDefinition<>(method, handler);
+    return new ServerMethodDefinition<>(method, handler, tracerFactories);
+  }
+
+  /**
+   * Create a new method definition with a different list of tracer factories.
+   *
+   * @param tracerFactories to bind to a cloned instance of this
+   * @return a cloned instance of this with an unmodifiable view of the new tracer factories
+   * @since 1.36.0
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/????")
+  public ServerMethodDefinition<ReqT, RespT> withStreamTracerFactories(
+      List<ServerStreamTracer.Factory> tracerFactories) {
+    return new ServerMethodDefinition<>(method, handler, tracerFactories);
+  }
+
+  /**
+   * Returns the stream tracer factories bound to this definition.
+   *
+   * @since 1.36.0
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/????")
+  public List<ServerStreamTracer.Factory> getStreamTracerFactories() {
+    return tracerFactories;
   }
 }

core/src/main/java/io/grpc/inprocess/InProcessTransport.java

@@ -417,13 +417,47 @@ private class InProcessServerStream implements ServerStream {
       @GuardedBy("this")
       private int outboundSeqNo;
 
+      private final class CallStartedListener
+          implements StatsTraceContext.ServerCallStartedListener {
+        private final ArrayList<Runnable> queuedStatsEvents = new ArrayList<Runnable>();
+        private boolean isReady;
+
+        @Override
+        public void serverCallStarted() {
+          synchronized (this) {
+            for (Runnable queuedEvent : queuedStatsEvents) {
+              queuedEvent.run();
+            }
+            queuedStatsEvents.clear();
+            isReady = true;
+          }
+        }
+
+        private void queueEvent(Runnable event) {
+          synchronized (this) {
+            if (isReady) {
+              event.run();
+            } else {
+              queuedStatsEvents.add(event);
+            }
+          }
+        }
+      }
+
+      private final CallStartedListener serverCallStartedListener = new CallStartedListener();
+
       InProcessServerStream(MethodDescriptor<?, ?> method, Metadata headers) {
-        statsTraceCtx = StatsTraceContext.newServerContext(
-            serverStreamTracerFactories, method.getFullMethodName(), headers);
+        statsTraceCtx =
+            StatsTraceContext.newServerContext(
+                serverStreamTracerFactories,
+                method.getFullMethodName(),
+                headers,
+                serverCallStartedListener);
       }
 
-      private synchronized void setListener(ClientStreamListener listener) {
-        clientStreamListener = listener;
+      private synchronized void setListener(
+          ClientStreamListener statsTraceCtxServerCallStartedListener) {
+        clientStreamListener = statsTraceCtxServerCallStartedListener;
       }
 
       @Override
@@ -725,8 +759,21 @@ public synchronized void writeMessage(InputStream message) {
         }
         statsTraceCtx.outboundMessage(outboundSeqNo);
         statsTraceCtx.outboundMessageSent(outboundSeqNo, -1, -1);
-        serverStream.statsTraceCtx.inboundMessage(outboundSeqNo);
-        serverStream.statsTraceCtx.inboundMessageRead(outboundSeqNo, -1, -1);
+        final int outboundSeqNumCopy = outboundSeqNo;
+        serverStream.serverCallStartedListener.queueEvent(
+            new Runnable() {
+              @Override
+              public void run() {
+                serverStream.statsTraceCtx.inboundMessage(outboundSeqNumCopy);
+              }
+            });
+        serverStream.serverCallStartedListener.queueEvent(
+            new Runnable() {
+              @Override
+              public void run() {
+                serverStream.statsTraceCtx.inboundMessageRead(outboundSeqNumCopy, -1, -1);
+              }
+            });
         outboundSeqNo++;
         StreamListener.MessageProducer producer = new SingleMessageProducer(message);
         if (serverRequested > 0) {

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

@@ -25,6 +25,7 @@
 import io.grpc.Server;
 import io.grpc.ServerBuilder;
 import io.grpc.ServerInterceptor;
+import io.grpc.ServerInterceptor2;
 import io.grpc.ServerServiceDefinition;
 import io.grpc.ServerStreamTracer;
 import io.grpc.ServerTransportFilter;
@@ -91,6 +92,12 @@ public T intercept(ServerInterceptor interceptor) {
     return thisT();
   }
 
+  @Override
+  public T intercept(ServerInterceptor2 interceptor) {
+    delegate().intercept(interceptor);
+    return thisT();
+  }
+
   @Override
   public T addTransportFilter(ServerTransportFilter filter) {
     delegate().addTransportFilter(filter);