diff --git a/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java b/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java index f23c5404eb..bd89bd6461 100644 --- a/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java +++ b/transport/src/main/java/io/netty/channel/pool/FixedChannelPool.java @@ -117,7 +117,7 @@ public final class FixedChannelPool extends SimpleChannelPool { * be failed. */ public FixedChannelPool(Bootstrap bootstrap, - ChannelPoolHandler handler, + ChannelPoolHandler handler, ChannelHealthChecker healthCheck, AcquireTimeoutAction action, final long acquireTimeoutMillis, int maxConnections, int maxPendingAcquires) { @@ -153,7 +153,7 @@ public final class FixedChannelPool extends SimpleChannelPool { public void onTimeout(AcquireTask task) { // Increment the acquire count and delegate to super to actually acquire a Channel which will // create a new connetion. - ++acquiredChannelCount; + task.acquired(); FixedChannelPool.super.acquire(task.promise); } @@ -191,14 +191,14 @@ public final class FixedChannelPool extends SimpleChannelPool { assert executor.inEventLoop(); if (acquiredChannelCount < maxConnections) { - ++acquiredChannelCount; - - assert acquiredChannelCount > 0; + assert acquiredChannelCount >= 0; // We need to create a new promise as we need to ensure the AcquireListener runs in the correct // EventLoop Promise p = executor.newPromise(); - p.addListener(new AcquireListener(promise)); + AcquireListener l = new AcquireListener(promise); + l.acquired(); + p.addListener(l); super.acquire(p); } else { if (pendingAcquireCount >= maxPendingAcquires) { @@ -271,10 +271,8 @@ public final class FixedChannelPool extends SimpleChannelPool { timeoutFuture.cancel(false); } - task.acquired(); - --pendingAcquireCount; - ++acquiredChannelCount; + task.acquired(); super.acquire(task.promise); } @@ -291,7 +289,7 @@ public final class FixedChannelPool extends SimpleChannelPool { ScheduledFuture timeoutFuture; public AcquireTask(Promise promise) { - super(promise, false); + super(promise); // We need to create a new promise as we need to ensure the AcquireListener runs in the correct // EventLoop. this.promise = executor.newPromise().addListener(this); @@ -327,12 +325,7 @@ public final class FixedChannelPool extends SimpleChannelPool { protected boolean acquired; AcquireListener(Promise originalPromise) { - this(originalPromise, true); - } - - protected AcquireListener(Promise originalPromise, boolean acquired) { this.originalPromise = originalPromise; - this.acquired = acquired; } @Override @@ -353,6 +346,10 @@ public final class FixedChannelPool extends SimpleChannelPool { } public void acquired() { + if (acquired) { + return; + } + acquiredChannelCount++; acquired = true; } }