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

@@ -104,6 +104,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} and any interceptors added through {@link #intercept(ServerInterceptor)}.
+   * Interceptors run in the reverse order in which they are added, just as with consecutive calls
+   * to {@code ServerInterceptors.intercept()}.
+   *
+   * @param interceptor the all-service interceptor
+   * @return this
+   * @since 1.35.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/ServerMethodDefinition.java

@@ -16,6 +16,9 @@
 
 package io.grpc;
 
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * Definition of a method exposed by a {@link Server}.
  *
@@ -24,11 +27,13 @@
 public final class ServerMethodDefinition<ReqT, RespT> {
   private final MethodDescriptor<ReqT, RespT> method;
   private final ServerCallHandler<ReqT, RespT> handler;
+  private final List<ServerStreamTracer.Factory> streamTracerFactories;
 
   private ServerMethodDefinition(MethodDescriptor<ReqT, RespT> method,
       ServerCallHandler<ReqT, RespT> handler) {
     this.method = method;
     this.handler = handler;
+    this.streamTracerFactories = new ArrayList<>();
   }
 
   /**
@@ -64,4 +69,16 @@ public ServerMethodDefinition<ReqT, RespT> withServerCallHandler(
       ServerCallHandler<ReqT, RespT> handler) {
     return new ServerMethodDefinition<>(method, handler);
   }
+
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/????")
+  public ServerMethodDefinition<ReqT, RespT>
+      addStreamTracerFactory(ServerStreamTracer.Factory factory) {
+    streamTracerFactories.add(factory);
+    return this;
+  }
+
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/????")
+  public List<? extends ServerStreamTracer.Factory> getStreamTracerFactories() {
+    return streamTracerFactories;
+  }
 }

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