Add support for anonymous in-process servers.

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2021 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.inprocess;
+
+import static com.google.common.base.Preconditions.checkState;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+/**
+ * Custom SocketAddress class for {@link InProcessTransport}, for 
+ * a server which can only be referenced via this address instance.
+ */
+public final class AnonymousInProcessSocketAddress extends SocketAddress {
+  private static final long serialVersionUID = -8567592561863414695L;
+
+  @Nullable
+  @GuardedBy("this")
+  private InProcessServer server;
+
+  /** Creates a new AnonymousInProcessSocketAddress. */
+  public AnonymousInProcessSocketAddress() { }
+
+  @Nullable
+  synchronized InProcessServer getServer() {
+    return server;
+  }
+
+  synchronized void setServer(InProcessServer server) throws IOException {
+    if (this.server != null) {
+      throw new IOException("Server instance already registered");
+    }
+    this.server = server;
+  }
+
+  synchronized void clearServer(InProcessServer server) {
+    checkState(this.server == server);
+    this.server = null;
+  }
+
+  @Override
+  public synchronized int hashCode() {
+    return server == null ? 0 : server.hashCode();
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (!(obj instanceof AnonymousInProcessSocketAddress)) {
+      return false;
+    }
+    InProcessServer otherServer = ((AnonymousInProcessSocketAddress) obj).getServer();
+    synchronized (this) {
+      return otherServer == server;
+    }
+  }
+}

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

@@ -55,15 +55,28 @@ public final class InProcessChannelBuilder extends
    * @return a new builder
    */
   public static InProcessChannelBuilder forName(String name) {
-    return new InProcessChannelBuilder(name);
+    return forAddress(new InProcessSocketAddress(checkNotNull(name, "name")));
   }
 
   /**
-   * Always fails.  Call {@link #forName} instead.
+   * Create a channel builder that will connect to the server referenced by the given target URI.
+   * Only intended for use with a custom name resolver.
+   *
+   * @param target the identity of the server to connect to
+   * @return a new builder
    */
-  @DoNotCall("Unsupported. Use forName() instead")
   public static InProcessChannelBuilder forTarget(String target) {
-    throw new UnsupportedOperationException("call forName() instead");
+    return new InProcessChannelBuilder(null, checkNotNull(target, "target"));
+  }
+
+  /**
+   * Create a channel builder that will connect to the server referenced by the given address.
+   *
+   * @param address the address of the server to connect to
+   * @return a new builder
+   */
+  public static InProcessChannelBuilder forAddress(SocketAddress address) {
+    return new InProcessChannelBuilder(checkNotNull(address, "address"), null);
   }
 
   /**
@@ -75,13 +88,11 @@ public static InProcessChannelBuilder forAddress(String name, int port) {
   }
 
   private final ManagedChannelImplBuilder managedChannelImplBuilder;
-  private final String name;
   private ScheduledExecutorService scheduledExecutorService;
   private int maxInboundMetadataSize = Integer.MAX_VALUE;
   private boolean transportIncludeStatusCause = false;
 
-  private InProcessChannelBuilder(String name) {
-    this.name = checkNotNull(name, "name");
+  private InProcessChannelBuilder(@Nullable SocketAddress directAddress, @Nullable String target) {
 
     final class InProcessChannelTransportFactoryBuilder implements ClientTransportFactoryBuilder {
       @Override
@@ -90,8 +101,13 @@ public ClientTransportFactory buildClientTransportFactory() {
       }
     }
 
-    managedChannelImplBuilder = new ManagedChannelImplBuilder(new InProcessSocketAddress(name),
-        "localhost", new InProcessChannelTransportFactoryBuilder(), null);
+    if (directAddress != null) {
+      managedChannelImplBuilder = new ManagedChannelImplBuilder(directAddress, "localhost",
+          new InProcessChannelTransportFactoryBuilder(), null);
+    } else {
+      managedChannelImplBuilder = new ManagedChannelImplBuilder(target,
+          new InProcessChannelTransportFactoryBuilder(), null);
+    }
 
     // In-process transport should not record its traffic to the stats module.
     // https://github.com/grpc/grpc-java/issues/2284
@@ -204,7 +220,7 @@ public InProcessChannelBuilder propagateCauseWithStatus(boolean enable) {
 
   ClientTransportFactory buildTransportFactory() {
     return new InProcessClientTransportFactory(
-        name, scheduledExecutorService, maxInboundMetadataSize, transportIncludeStatusCause);
+        scheduledExecutorService, maxInboundMetadataSize, transportIncludeStatusCause);
   }
 
   void setStatsEnabled(boolean value) {
@@ -215,18 +231,15 @@ void setStatsEnabled(boolean value) {
    * Creates InProcess transports. Exposed for internal use, as it should be private.
    */
   static final class InProcessClientTransportFactory implements ClientTransportFactory {
-    private final String name;
     private final ScheduledExecutorService timerService;
     private final boolean useSharedTimer;
     private final int maxInboundMetadataSize;
     private boolean closed;
     private final boolean includeCauseWithStatus;
 
     private InProcessClientTransportFactory(
-        String name,
         @Nullable ScheduledExecutorService scheduledExecutorService,
         int maxInboundMetadataSize, boolean includeCauseWithStatus) {
-      this.name = name;
       useSharedTimer = scheduledExecutorService == null;
       timerService = useSharedTimer
           ? SharedResourceHolder.get(GrpcUtil.TIMER_SERVICE) : scheduledExecutorService;
@@ -242,7 +255,7 @@ public ConnectionClientTransport newClientTransport(
       }
       // TODO(carl-mastrangelo): Pass channelLogger in.
       return new InProcessTransport(
-          name, maxInboundMetadataSize, options.getAuthority(), options.getUserAgent(),
+          addr, maxInboundMetadataSize, options.getAuthority(), options.getUserAgent(),
           options.getEagAttributes(), includeCauseWithStatus);
     }
 

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

@@ -40,11 +40,16 @@ final class InProcessServer implements InternalServer {
   private static final ConcurrentMap<String, InProcessServer> registry
       = new ConcurrentHashMap<>();
 
-  static InProcessServer findServer(String name) {
-    return registry.get(name);
+  static InProcessServer findServer(SocketAddress addr) {
+    if (addr instanceof InProcessSocketAddress) {
+      return registry.get(((InProcessSocketAddress) addr).getName());
+    } else if (addr instanceof AnonymousInProcessSocketAddress) {
+      return ((AnonymousInProcessSocketAddress) addr).getServer();
+    }
+    return null;
   }
 
-  private final String name;
+  private final SocketAddress listenAddress;
   private final int maxInboundMetadataSize;
   private final List<ServerStreamTracer.Factory> streamTracerFactories;
   private ServerListener listener;
@@ -60,7 +65,7 @@ static InProcessServer findServer(String name) {
   InProcessServer(
       InProcessServerBuilder builder,
       List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
-    this.name = builder.name;
+    this.listenAddress = builder.listenAddress;
     this.schedulerPool = builder.schedulerPool;
     this.maxInboundMetadataSize = builder.maxInboundMetadataSize;
     this.streamTracerFactories =
@@ -72,14 +77,23 @@ public void start(ServerListener serverListener) throws IOException {
     this.listener = serverListener;
     this.scheduler = schedulerPool.getObject();
     // Must be last, as channels can start connecting after this point.
-    if (registry.putIfAbsent(name, this) != null) {
-      throw new IOException("name already registered: " + name);
+    registerInstance();
+  }
+
+  private void registerInstance() throws IOException {
+    if (listenAddress instanceof InProcessSocketAddress) {
+      String name = ((InProcessSocketAddress) listenAddress).getName();
+      if (registry.putIfAbsent(name, this) != null) {
+        throw new IOException("name already registered: " + name);
+      }
+    } else if (listenAddress instanceof AnonymousInProcessSocketAddress) {
+      ((AnonymousInProcessSocketAddress) listenAddress).setServer(this);
     }
   }
 
   @Override
   public SocketAddress getListenSocketAddress() {
-    return new InProcessSocketAddress(name);
+    return listenAddress;
   }
 
   @Override
@@ -99,19 +113,28 @@ public List<InternalInstrumented<SocketStats>> getListenSocketStatsList() {
 
   @Override
   public void shutdown() {
-    if (!registry.remove(name, this)) {
-      throw new AssertionError();
-    }
+    unregisterInstance();
     scheduler = schedulerPool.returnObject(scheduler);
     synchronized (this) {
       shutdown = true;
       listener.serverShutdown();
     }
   }
 
+  private void unregisterInstance() {
+    if (listenAddress instanceof InProcessSocketAddress) {
+      String name = ((InProcessSocketAddress) listenAddress).getName();
+      if (!registry.remove(name, this)) {
+        throw new AssertionError();
+      }
+    } else if (listenAddress instanceof AnonymousInProcessSocketAddress) {
+      ((AnonymousInProcessSocketAddress) listenAddress).clearServer(this);
+    }
+  }
+
   @Override
   public String toString() {
-    return MoreObjects.toStringHelper(this).add("name", name).toString();
+    return MoreObjects.toStringHelper(this).add("listenAddress", listenAddress).toString();
   }
 
   synchronized ServerTransportListener register(InProcessTransport transport) {

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

@@ -34,6 +34,7 @@
 import io.grpc.internal.ServerImplBuilder.ClientTransportServersBuilder;
 import io.grpc.internal.SharedResourcePool;
 import java.io.File;
+import java.net.SocketAddress;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ScheduledExecutorService;
@@ -81,7 +82,16 @@ public final class InProcessServerBuilder extends
    * @return a new builder
    */
   public static InProcessServerBuilder forName(String name) {
-    return new InProcessServerBuilder(name);
+    return forAddress(new InProcessSocketAddress(checkNotNull(name, "name")));
+  }
+
+  /**
+   * Create a server builder which listens on the given address.
+   * @param listenAddress The SocketAddress this server will listen on.
+   * @return a new builder
+   */
+  public static InProcessServerBuilder forAddress(SocketAddress listenAddress) {
+    return new InProcessServerBuilder(listenAddress);
   }
 
   /**
@@ -100,13 +110,13 @@ public static String generateName() {
   }
 
   private final ServerImplBuilder serverImplBuilder;
-  final String name;
+  final SocketAddress listenAddress;
   int maxInboundMetadataSize = Integer.MAX_VALUE;
   ObjectPool<ScheduledExecutorService> schedulerPool =
       SharedResourcePool.forResource(GrpcUtil.TIMER_SERVICE);
 
-  private InProcessServerBuilder(String name) {
-    this.name = Preconditions.checkNotNull(name, "name");
+  private InProcessServerBuilder(SocketAddress listenAddress) {
+    this.listenAddress = checkNotNull(listenAddress, "listenAddress");
 
     final class InProcessClientTransportServersBuilder implements ClientTransportServersBuilder {
       @Override