reorganized gRPC channel logic

src/main/java/emissary/grpc/GrpcConnectionPlace.java

@@ -1,7 +1,7 @@
 package emissary.grpc;
 
 import emissary.config.Configurator;
-import emissary.grpc.pool.ConnectionFactory;
+import emissary.grpc.channel.ChannelManager;
 import emissary.grpc.retry.RetryHandler;
 
 import com.google.common.util.concurrent.ListenableFuture;
@@ -27,7 +27,7 @@
  * <ul>
  * <li>{@code GRPC_HOST} - gRPC service hostname or DNS target, <i>required</i></li>
  * <li>{@code GRPC_PORT} - gRPC service port, <i>required</i></li>
- * <li>See {@link ConnectionFactory} for supported pooling and gRPC channel configuration keys and defaults.</li>
+ * <li>See {@link ChannelManager} for supported gRPC channel configuration keys and defaults.</li>
  * <li>See {@link RetryHandler} for supported retry configuration keys and defaults.</li>
  * </ul>
  */

src/main/java/emissary/grpc/GrpcRoutingPlace.java

@@ -1,9 +1,9 @@
 package emissary.grpc;
 
 import emissary.config.Configurator;
+import emissary.grpc.channel.ChannelManager;
+import emissary.grpc.channel.ChannelPoolFactory.PoolException;
 import emissary.grpc.invoker.GrpcInvoker;
-import emissary.grpc.pool.ConnectionFactory;
-import emissary.grpc.pool.PoolException;
 import emissary.grpc.retry.RetryHandler;
 import emissary.place.ServiceProviderPlace;
 
@@ -15,7 +15,6 @@
 import io.grpc.stub.AbstractBlockingStub;
 import io.grpc.stub.AbstractFutureStub;
 import jakarta.annotation.Nullable;
-import org.apache.commons.pool2.ObjectPool;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -38,7 +37,7 @@
  * identifier for the given host:port</li>
  * <li>{@code GRPC_PORT_{Target-ID}} - gRPC service port, where {@code Target-ID} is the unique identifier for the given
  * host:port</li>
- * <li>See {@link ConnectionFactory} for supported pooling and gRPC channel configuration keys and defaults.</li>
+ * <li>See {@link ChannelManager} for supported gRPC channel configuration keys and defaults.</li>
  * <li>See {@link RetryHandler} for supported retry configuration keys and defaults.</li>
  * </ul>
  */
@@ -51,11 +50,7 @@ public abstract class GrpcRoutingPlace extends ServiceProviderPlace implements I
     public static final String GRPC_HOST = "GRPC_HOST_";
     public static final String GRPC_PORT = "GRPC_PORT_";
 
-    protected GrpcInvoker grpcInvoker;
-
-    protected final Map<String, String> hostnameTable = new HashMap<>();
-    protected final Map<String, Integer> portNumberTable = new HashMap<>();
-    protected final Map<String, ObjectPool<ManagedChannel>> channelPoolTable = new HashMap<>();
+    protected final Map<String, GrpcInvoker> invokerTable = new HashMap<>();
 
     protected GrpcRoutingPlace() throws IOException {
         super();
@@ -98,28 +93,27 @@ protected GrpcRoutingPlace(@Nullable Configurator configs) throws IOException {
     }
 
     private void configureGrpc() {
-        if (configG == null) {
-            throw new IllegalStateException("gRPC configurations not found for " + this.getPlaceName());
-        }
+        Objects.requireNonNull(configG);
 
-        hostnameTable.putAll(getHostnameConfigs());
-        portNumberTable.putAll(getPortNumberConfigs());
+        Map<String, String> hosts = getHostnameConfigs();
+        Map<String, Integer> ports = getPortNumberConfigs();
 
-        if (!hostnameTable.keySet().equals(portNumberTable.keySet())) {
+        if (!hosts.keySet().equals(ports.keySet())) {
             throw new IllegalArgumentException("gRPC hostname target-IDs do not match gRPC port number target-IDs");
         }
 
-        if (hostnameTable.isEmpty()) {
+        Set<String> targetIds = hosts.keySet();
+        if (targetIds.isEmpty()) {
             throw new NullPointerException(String.format(
                     "Missing required arguments: %s${Target-ID} and %s${Target-ID}", GRPC_HOST, GRPC_PORT));
         }
 
-        Set<String> targetIds = hostnameTable.keySet();
+        RetryHandler retryHandler = new RetryHandler(configG, this.getPlaceName(), this::retryOnException);
         for (String id : targetIds) {
-            channelPoolTable.put(id, newConnectionPool(id));
+            ChannelManager channelManager = new ChannelManager(hosts.get(id), ports.get(id), configG);
+            GrpcInvoker grpcInvoker = new GrpcInvoker(channelManager, retryHandler);
+            invokerTable.put(id, grpcInvoker);
         }
-
-        grpcInvoker = new GrpcInvoker(new RetryHandler(configG, this.getPlaceName(), this::retryOnException));
     }
 
     protected Map<String, String> getHostnameConfigs() {
@@ -147,17 +141,9 @@ protected boolean retryOnException(Throwable t) {
         return t instanceof PoolException;
     }
 
-    private ObjectPool<ManagedChannel> newConnectionPool(String id) {
-        return newConnectionFactory(id).newConnectionPool();
-    }
-
-    private ConnectionFactory newConnectionFactory(String id) {
-        return new ConnectionFactory(hostnameTable.get(id), portNumberTable.get(id), Objects.requireNonNull(configG));
-    }
-
     /**
-     * Wrapper method for {@link GrpcInvoker#invoke(ObjectPool, Function, BiFunction, Message)} that executes a unary gRPC
-     * call to a given endpoint.
+     * Wrapper method for {@link GrpcInvoker#invoke(Function, BiFunction, Message)} that executes a unary gRPC call to a
+     * given endpoint.
      *
      * @param targetId the identifier used in the configs for the given gRPC endpoint
      * @param stubFactory function that creates the appropriate gRPC stub from a {@link ManagedChannel}
@@ -170,12 +156,12 @@ private ConnectionFactory newConnectionFactory(String id) {
      */
     protected <Q extends Message, R extends Message, S extends AbstractBlockingStub<S>> R invokeGrpc(
             String targetId, Function<ManagedChannel, S> stubFactory, BiFunction<S, Q, R> callLogic, Q request) {
-        return grpcInvoker.invoke(channelPoolLookup(targetId), stubFactory, callLogic, request);
+        return getInvoker(targetId).invoke(stubFactory, callLogic, request);
     }
 
     /**
-     * Wrapper method for {@link GrpcInvoker#invokeAsync(ObjectPool, Function, BiFunction, Message)} that executes a unary
-     * gRPC call to a given endpoint and returns a {@link CompletableFuture future}.
+     * Wrapper method for {@link GrpcInvoker#invokeAsync(Function, BiFunction, Message)} that executes a unary gRPC call to
+     * a given endpoint and returns a {@link CompletableFuture future}.
      *
      * @param targetId the identifier used in the configs for the given gRPC endpoint
      * @param stubFactory function that creates the appropriate gRPC stub from a {@link ManagedChannel}
@@ -188,25 +174,27 @@ protected <Q extends Message, R extends Message, S extends AbstractBlockingStub<
      */
     protected <Q extends Message, R extends Message, S extends AbstractFutureStub<S>> CompletableFuture<R> invokeGrpcAsync(
             String targetId, Function<ManagedChannel, S> stubFactory, BiFunction<S, Q, ListenableFuture<R>> callLogic, Q request) {
-        return grpcInvoker.invokeAsync(channelPoolLookup(targetId), stubFactory, callLogic, request);
-    }
-
-    private ObjectPool<ManagedChannel> channelPoolLookup(String targetId) {
-        return tableLookup(channelPoolTable, targetId);
+        return getInvoker(targetId).invokeAsync(stubFactory, callLogic, request);
     }
 
     public String getHostname(String targetId) {
-        return tableLookup(hostnameTable, targetId);
+        return getInvoker(targetId).getHost();
     }
 
     public int getPortNumber(String targetId) {
-        return tableLookup(portNumberTable, targetId);
+        return getInvoker(targetId).getPort();
     }
 
-    protected <T> T tableLookup(Map<String, T> table, String targetId) {
-        if (table.containsKey(targetId)) {
-            return table.get(targetId);
+    private GrpcInvoker getInvoker(String targetId) {
+        if (invokerTable.containsKey(targetId)) {
+            return invokerTable.get(targetId);
         }
         throw new IllegalArgumentException(String.format("Target-ID %s was never configured", targetId));
     }
+
+    @Override
+    public void shutDown() {
+        super.shutDown();
+        invokerTable.values().forEach(GrpcInvoker::close);
+    }
 }

src/main/java/emissary/grpc/channel/ChannelManager.java

@@ -0,0 +1,46 @@
+package emissary.grpc.channel;
+
+import emissary.config.Configurator;
+
+import io.grpc.ManagedChannel;
+import org.apache.commons.pool2.ObjectPool;
+
+/**
+ * Wrapper for an {@link ObjectPool} created by a {@link ChannelPoolFactory}.
+ */
+public class ChannelManager implements AutoCloseable {
+    private final ObjectPool<ManagedChannel> channelPool;
+    private final String host;
+    private final int port;
+
+    public ChannelManager(String host, int port, Configurator configG) {
+        this.channelPool = new ChannelPoolFactory(host, port, configG).newConnectionPool();
+        this.host = host;
+        this.port = port;
+    }
+
+    public ManagedChannel acquire() {
+        return ChannelPoolFactory.acquireChannel(channelPool);
+    }
+
+    public void release(ManagedChannel channel) {
+        ChannelPoolFactory.returnChannel(channel, channelPool);
+    }
+
+    public void shutdown(ManagedChannel channel) {
+        ChannelPoolFactory.invalidateChannel(channel, channelPool);
+    }
+
+    @Override
+    public void close() {
+        channelPool.close();
+    }
+
+    public String getHost() {
+        return host;
+    }
+
+    public int getPort() {
+        return port;
+    }
+}

src/main/java/emissary/grpc/pool/ConnectionFactory.java → src/main/java/emissary/grpc/channel/ChannelPoolFactory.java

@@ -1,4 +1,4 @@
-package emissary.grpc.pool;
+package emissary.grpc.channel;
 
 import emissary.config.Configurator;
 
@@ -45,7 +45,7 @@
  * <a href="https://docs.microsoft.com/en-us/aspnet/core/grpc/performance?view=aspnetcore-5.0">Source</a> for default
  * gRPC configurations.
  */
-public class ConnectionFactory extends BasePooledObjectFactory<ManagedChannel> {
+public class ChannelPoolFactory extends BasePooledObjectFactory<ManagedChannel> {
     public static final String GRPC_KEEP_ALIVE_MILLIS = "GRPC_KEEP_ALIVE_MILLIS";
     public static final String GRPC_KEEP_ALIVE_TIMEOUT_MILLIS = "GRPC_KEEP_ALIVE_TIMEOUT_MILLIS";
     public static final String GRPC_KEEP_ALIVE_WITHOUT_CALLS = "GRPC_KEEP_ALIVE_WITHOUT_CALLS";
@@ -62,7 +62,7 @@ public class ConnectionFactory extends BasePooledObjectFactory<ManagedChannel> {
 
     private static final int MAX_PORT_NUMBER = 0xFFFF;
 
-    protected static final Logger logger = LoggerFactory.getLogger(ConnectionFactory.class);
+    protected static final Logger logger = LoggerFactory.getLogger(ChannelPoolFactory.class);
 
     private final GenericObjectPoolConfig<ManagedChannel> poolConfig = new GenericObjectPoolConfig<>();
 
@@ -81,13 +81,13 @@ public class ConnectionFactory extends BasePooledObjectFactory<ManagedChannel> {
      * Constructs a new gRPC connection factory using the provided host, port, and configuration. Initializes pool settings
      * and gRPC channel properties from the given configuration source.
      * <p>
-     * See {@link ConnectionFactory} for supported configuration keys and defaults.
+     * See {@link ChannelPoolFactory} for supported configuration keys and defaults.
      * 
      * @param host gRPC service hostname or DNS target
      * @param port gRPC service port
      * @param configG configuration provider for channel and pool parameters
      */
-    public ConnectionFactory(String host, int port, Configurator configG) {
+    public ChannelPoolFactory(String host, int port, Configurator configG) {
         this.host = host;
         this.port = port;
         this.target = createTarget();
@@ -105,7 +105,7 @@ public ConnectionFactory(String host, int port, Configurator configG) {
         // Specifies how the client chooses between multiple backend addresses
         // e.g. "pick_first" uses the first address only, "round_robin" cycles through all of them for client-side balancing
         this.loadBalancingPolicy = configG.findObjectEntry(
-                GRPC_LOAD_BALANCING_POLICY, LoadBalancingPolicy::valueOf, LoadBalancingPolicy.ROUND_ROBIN).formattedName();
+                GRPC_LOAD_BALANCING_POLICY, LoadBalancingPolicy::valueOf, LoadBalancingPolicy.ROUND_ROBIN).toString();
 
         // Max size (in bytes) for incoming messages and message metadata from the server
         this.maxInboundMessageByteSize = configG.findIntEntry(GRPC_MAX_INBOUND_MESSAGE_BYTE_SIZE, 4 << 20); // 4 MiB
@@ -331,4 +331,35 @@ public boolean getPoolIsLifo() {
     public boolean getPoolIsFifo() {
         return !this.poolConfig.getLifo();
     }
+
+    /**
+     * Exception type for failures with handling the gRPC connection pool, such as failed borrows.
+     */
+    public static class PoolException extends RuntimeException {
+
+        private static final long serialVersionUID = 1495483102825486040L;
+
+        public PoolException(String errorMessage, Throwable err) {
+            super(errorMessage, err);
+        }
+    }
+
+    public enum PoolRetrievalOrdering {
+        LIFO, FIFO;
+    }
+
+    public enum LoadBalancingPolicy {
+        ROUND_ROBIN("round_robin"), PICK_FIRST("pick_first");
+
+        private final String policy;
+
+        LoadBalancingPolicy(String policy) {
+            this.policy = policy;
+        }
+
+        @Override
+        public String toString() {
+            return policy;
+        }
+    }
 }

src/main/java/emissary/grpc/exceptions/GrpcExceptionUtils.java

@@ -58,7 +58,7 @@ private static String getStatusCodeDescription(Status status) {
 
     private static String getStatusCodeDescription(Status status, String message) {
         String description = status.getDescription();
-        if (description != null) {
+        if (description != null && !description.equals(message)) {
             return String.format("%s (%s)", description, message);
         }
         return message;