Add enqueuing timeout option in Netty configuration

acogoluegnes · acogoluegnes · commit f3c9e9465145 · 2025-08-07T09:34:09.000+02:00
Part of the back-pressure mechanism. References #1663
diff --git a/src/main/java/com/rabbitmq/client/ConnectionFactory.java b/src/main/java/com/rabbitmq/client/ConnectionFactory.java
@@ -35,6 +35,7 @@
 import java.net.URLDecoder;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
 import java.util.*;
 import java.util.Map.Entry;
 import java.util.concurrent.*;
@@ -1097,6 +1098,7 @@ protected synchronized FrameHandlerFactory createFrameHandlerFactory() throws IO
                 this.nettyConf.channelCustomizer,
                 this.nettyConf.bootstrapCustomizer,
                 this.nettyConf.sslContextFactory,
+                this.nettyConf.enqueuingTimeout,
                 connectionTimeout,
                 socketConf,
                 maxInboundMessageBodySize);
@@ -1917,6 +1919,7 @@ public static final class NettyConfiguration {
     private Consumer<io.netty.channel.Channel> channelCustomizer = ch -> {};
     private Consumer<Bootstrap> bootstrapCustomizer = b -> {};
     private Function<String, SslContext> sslContextFactory;
+    private Duration enqueuingTimeout = Duration.ofSeconds(10);
 
     private NettyConfiguration(ConnectionFactory cf) {
       this.cf = cf;
@@ -1990,6 +1993,19 @@ public NettyConfiguration sslContextFactory(Function<String, SslContext> sslCont
       return this;
     }
 
+    /**
+     * Set the timeout to enqueue outbound frames.
+     *
+     * <p>Default is 10 seconds.
+     *
+     * @param enqueuingTimeout the enqueuing timeout
+     * @return this configuration instance
+     */
+    public NettyConfiguration enqueuingTimeout(Duration enqueuingTimeout) {
+      this.enqueuingTimeout = enqueuingTimeout;
+      return this;
+    }
+
     /**
      * Go back to the connection factory.
      *
diff --git a/src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java b/src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java
@@ -70,12 +70,14 @@ public final class NettyFrameHandlerFactory extends AbstractFrameHandlerFactory
   private final Function<String, SslContext> sslContextFactory;
   private final Consumer<Channel> channelCustomizer;
   private final Consumer<Bootstrap> bootstrapCustomizer;
+  private final Duration enqueuingTimeout;
 
   public NettyFrameHandlerFactory(
       EventLoopGroup eventLoopGroup,
       Consumer<Channel> channelCustomizer,
       Consumer<Bootstrap> bootstrapCustomizer,
       Function<String, SslContext> sslContextFactory,
+      Duration enqueuingTimeout,
       int connectionTimeout,
       SocketConfigurator configurator,
       int maxInboundMessageBodySize) {
@@ -85,6 +87,7 @@ public NettyFrameHandlerFactory(
     this.channelCustomizer = channelCustomizer == null ? Utils.noOpConsumer() : channelCustomizer;
     this.bootstrapCustomizer =
         bootstrapCustomizer == null ? Utils.noOpConsumer() : bootstrapCustomizer;
+    this.enqueuingTimeout = enqueuingTimeout;
   }
 
   private static void closeNettyState(Channel channel, EventLoopGroup eventLoopGroup) {
@@ -127,6 +130,7 @@ public FrameHandler create(Address addr, String connectionName) throws IOExcepti
         addr,
         sslContext,
         this.eventLoopGroup,
+        this.enqueuingTimeout,
         this.channelCustomizer,
         this.bootstrapCustomizer);
   }
@@ -146,6 +150,7 @@ private static final class NettyFrameHandler implements FrameHandler {
           'A', 'M', 'Q', 'P', 0, AMQP.PROTOCOL.MAJOR, AMQP.PROTOCOL.MINOR, AMQP.PROTOCOL.REVISION
         };
     private final EventLoopGroup eventLoopGroup;
+    private final Duration enqueuingTimeout;
     private final Channel channel;
     private final AmqpHandler handler;
     private final AtomicBoolean closed = new AtomicBoolean(false);
@@ -155,9 +160,11 @@ private NettyFrameHandler(
         Address addr,
         SslContext sslContext,
         EventLoopGroup elg,
+        Duration enqueuingTimeout,
         Consumer<Channel> channelCustomizer,
         Consumer<Bootstrap> bootstrapCustomizer)
         throws IOException {
+      this.enqueuingTimeout = enqueuingTimeout;
       Bootstrap b = new Bootstrap();
       bootstrapCustomizer.accept(b);
       if (b.config().group() == null) {
@@ -310,7 +317,8 @@ public void writeFrame(Frame frame) throws IOException {
           this.doWriteFrame(frame);
         } else {
           try {
-            boolean canWriteNow = this.handler.writableLatch().await(10, SECONDS);
+            boolean canWriteNow =
+                this.handler.writableLatch().await(enqueuingTimeout.toMillis(), MILLISECONDS);
             if (canWriteNow) {
               this.doWriteFrame(frame);
             } else {
