Add support for anonymous in-process servers.

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

@@ -0,0 +1,58 @@
+/*
+ * 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 io.grpc.ExperimentalApi;
+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.
+ */
+@ExperimentalApi("https://github.com/grpc/grpc-java/issues/8626")
+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;
+  }
+}

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 AnonymousInProcessSocketAddress) {
+      return ((AnonymousInProcessSocketAddress) addr).getServer();
+    } else if (addr instanceof InProcessSocketAddress) {
+      return registry.get(((InProcessSocketAddress) addr).getName());
+    }
+    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 AnonymousInProcessSocketAddress) {
+      ((AnonymousInProcessSocketAddress) listenAddress).setServer(this);
+    } else if (listenAddress instanceof InProcessSocketAddress) {
+      String name = ((InProcessSocketAddress) listenAddress).getName();
+      if (registry.putIfAbsent(name, this) != null) {
+        throw new IOException("name already registered: " + name);
+      }
     }
   }
 
   @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 AnonymousInProcessSocketAddress) {
+      ((AnonymousInProcessSocketAddress) listenAddress).clearServer(this);
+    } else if (listenAddress instanceof InProcessSocketAddress) {
+      String name = ((InProcessSocketAddress) listenAddress).getName();
+      if (!registry.remove(name, this)) {
+        throw new AssertionError();
+      }
+    }
+  }
+
   @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

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

@@ -59,6 +59,7 @@
 import io.grpc.internal.StatsTraceContext;
 import io.grpc.internal.StreamListener;
 import java.io.InputStream;
+import java.net.SocketAddress;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -80,7 +81,7 @@ final class InProcessTransport implements ServerTransport, ConnectionClientTrans
   private static final Logger log = Logger.getLogger(InProcessTransport.class.getName());
 
   private final InternalLogId logId;
-  private final String name;
+  private final SocketAddress address;
   private final int clientMaxInboundMetadataSize;
   private final String authority;
   private final String userAgent;
@@ -119,29 +120,29 @@ protected void handleNotInUse() {
         }
       };
 
-  private InProcessTransport(String name, int maxInboundMetadataSize, String authority,
+  private InProcessTransport(SocketAddress address, int maxInboundMetadataSize, String authority,
       String userAgent, Attributes eagAttrs,
       Optional<ServerListener> optionalServerListener, boolean includeCauseWithStatus) {
-    this.name = name;
+    this.address = address;
     this.clientMaxInboundMetadataSize = maxInboundMetadataSize;
     this.authority = authority;
     this.userAgent = GrpcUtil.getGrpcUserAgent("inprocess", userAgent);
     checkNotNull(eagAttrs, "eagAttrs");
     this.attributes = Attributes.newBuilder()
         .set(GrpcAttributes.ATTR_SECURITY_LEVEL, SecurityLevel.PRIVACY_AND_INTEGRITY)
         .set(GrpcAttributes.ATTR_CLIENT_EAG_ATTRS, eagAttrs)
-        .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InProcessSocketAddress(name))
-        .set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, new InProcessSocketAddress(name))
+        .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, address)
+        .set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, address)
         .build();
     this.optionalServerListener = optionalServerListener;
-    logId = InternalLogId.allocate(getClass(), name);
+    logId = InternalLogId.allocate(getClass(), address.toString());
     this.includeCauseWithStatus = includeCauseWithStatus;
   }
 
   public InProcessTransport(
-      String name, int maxInboundMetadataSize, String authority, String userAgent,
+      SocketAddress address, int maxInboundMetadataSize, String authority, String userAgent,
       Attributes eagAttrs, boolean includeCauseWithStatus) {
-    this(name, maxInboundMetadataSize, authority, userAgent, eagAttrs,
+    this(address, maxInboundMetadataSize, authority, userAgent, eagAttrs,
         Optional.<ServerListener>absent(), includeCauseWithStatus);
   }
 
@@ -150,7 +151,7 @@ public InProcessTransport(
       Attributes eagAttrs, ObjectPool<ScheduledExecutorService> serverSchedulerPool,
       List<ServerStreamTracer.Factory> serverStreamTracerFactories,
       ServerListener serverListener) {
-    this(name, maxInboundMetadataSize, authority, userAgent, eagAttrs,
+    this(new InProcessSocketAddress(name), maxInboundMetadataSize, authority, userAgent, eagAttrs,
         Optional.of(serverListener), false);
     this.serverMaxInboundMetadataSize = maxInboundMetadataSize;
     this.serverSchedulerPool = serverSchedulerPool;
@@ -165,7 +166,7 @@ public synchronized Runnable start(ManagedClientTransport.Listener listener) {
       serverScheduler = serverSchedulerPool.getObject();
       serverTransportListener = optionalServerListener.get().transportCreated(this);
     } else {
-      InProcessServer server = InProcessServer.findServer(name);
+      InProcessServer server = InProcessServer.findServer(address);
       if (server != null) {
         serverMaxInboundMetadataSize = server.getMaxInboundMetadataSize();
         serverSchedulerPool = server.getScheduledExecutorServicePool();
@@ -176,7 +177,7 @@ public synchronized Runnable start(ManagedClientTransport.Listener listener) {
       }
     }
     if (serverTransportListener == null) {
-      shutdownStatus = Status.UNAVAILABLE.withDescription("Could not find server: " + name);
+      shutdownStatus = Status.UNAVAILABLE.withDescription("Could not find server: " + address);
       final Status localShutdownStatus = shutdownStatus;
       return new Runnable() {
         @Override
@@ -194,8 +195,8 @@ public void run() {
       public void run() {
         synchronized (InProcessTransport.this) {
           Attributes serverTransportAttrs = Attributes.newBuilder()
-              .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, new InProcessSocketAddress(name))
-              .set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, new InProcessSocketAddress(name))
+              .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, address)
+              .set(Grpc.TRANSPORT_ATTR_LOCAL_ADDR, address)
               .build();
           serverStreamAttributes = serverTransportListener.transportReady(serverTransportAttrs);
           clientTransportListener.transportReady();
@@ -307,7 +308,7 @@ public void shutdownNow(Status reason) {
   public String toString() {
     return MoreObjects.toStringHelper(this)
         .add("logId", logId.getId())
-        .add("name", name)
+        .add("address", address)
         .toString();
   }