This is an automated email from the ASF dual-hosted git repository.
dakirily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-broker-j.git
The following commit(s) were added to refs/heads/main by this push:
new 438cc4b6f2 QPID-8788: [Broker-J] Connection close can exceed the
configured timeout under transport backpressure (#463)
438cc4b6f2 is described below
commit 438cc4b6f2c15398df4b99a6df3c7bf413b07420
Author: Kato Kaito <[email protected]>
AuthorDate: Mon Nov 2 13:36:58 2026 +0100
QPID-8788: [Broker-J] Connection close can exceed the configured timeout
under transport backpressure (#463)
---
.../org/apache/qpid/server/model/Connection.java | 4 +-
.../server/protocol/ConnectionClosingTicker.java | 34 +-
.../qpid/server/transport/AMQPConnection.java | 4 +
.../server/transport/AbstractAMQPConnection.java | 26 +
.../server/transport/NonBlockingConnection.java | 110 ++-
.../NonBlockingConnectionTLSDelegate.java | 32 +-
.../server/transport/ServerNetworkConnection.java | 10 +
.../protocol/ConnectionClosingTickerTest.java | 109 +++
.../transport/NativeCloseDeadlineTestBase.java | 279 ++++++++
.../NonBlockingConnectionCloseSchedulingTest.java | 197 ++++++
.../transport/NonBlockingConnectionCloseTest.java | 770 +++++++++++++++++++++
.../NonBlockingConnectionExceptionTest.java | 23 +-
.../NonBlockingConnectionIdleTimeoutTest.java | 31 +-
.../transport/NonBlockingConnectionTestHelper.java | 84 +++
.../server/protocol/v0_10/ServerConnection.java | 13 +-
.../v0_10/NativeCloseDeadline_0_10Test.java | 92 +++
.../protocol/v0_8/AMQPConnection_0_8Impl.java | 18 +-
.../protocol/v0_8/AMQPConnection_0_8Test.java | 6 +-
.../protocol/v0_8/NativeCloseDeadline_0_8Test.java | 66 ++
.../protocol/v1_0/AMQPConnection_1_0Impl.java | 27 +-
.../protocol/v1_0/NativeCloseDeadline_1_0Test.java | 55 ++
.../AMQPConnection_1_0CloseExistingTest.java | 298 ++++++++
22 files changed, 2166 insertions(+), 122 deletions(-)
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/model/Connection.java
b/broker-core/src/main/java/org/apache/qpid/server/model/Connection.java
index f72d0451c4..33230d3485 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/Connection.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/Connection.java
@@ -59,7 +59,9 @@ public interface Connection<X extends Connection<X>> extends
ConfiguredObject<X>
int DEFAULT_MAX_MESSAGE_DECOMPRESSION_SIZE = 100 * 1024 * 1024;
String CLOSE_RESPONSE_TIMEOUT = "connection.closeResponseTimeout";
- @ManagedContextDefault(name = CLOSE_RESPONSE_TIMEOUT)
+ @ManagedContextDefault(name = CLOSE_RESPONSE_TIMEOUT,
+ description = "Maximum time in milliseconds to send the connection
close and await the peer's response," +
+ " starting when closing is requested. A non-positive value
expires immediately.")
long DEFAULT_CLOSE_RESPONSE_TIMEOUT = 2000L;
String MAX_MESSAGE_SIZE = "qpid.max_message_size";
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/protocol/ConnectionClosingTicker.java
b/broker-core/src/main/java/org/apache/qpid/server/protocol/ConnectionClosingTicker.java
index 2a82329678..4d063df1c3 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/protocol/ConnectionClosingTicker.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/protocol/ConnectionClosingTicker.java
@@ -20,38 +20,54 @@
*/
package org.apache.qpid.server.protocol;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+
import org.apache.qpid.server.transport.ServerNetworkConnection;
import org.apache.qpid.server.transport.network.Ticker;
public class ConnectionClosingTicker implements Ticker
{
- private final long _timeoutTime;
+ private final long _startTimeNanos;
+ private final long _timeoutNanos;
+ private final LongSupplier _nanoTime;
private final ServerNetworkConnection _network;
- public ConnectionClosingTicker(final long timeoutTime, final
ServerNetworkConnection network)
+ public ConnectionClosingTicker(final long timeoutMillis, final
ServerNetworkConnection network)
{
- _timeoutTime = timeoutTime;
+ this(timeoutMillis, network, System::nanoTime);
+ }
+
+ ConnectionClosingTicker(final long timeoutMillis,
+ final ServerNetworkConnection network,
+ final LongSupplier nanoTime)
+ {
+ _nanoTime = nanoTime;
+ _startTimeNanos = nanoTime.getAsLong();
+ _timeoutNanos = TimeUnit.MILLISECONDS.toNanos(Math.max(0L,
timeoutMillis));
_network = network;
}
@Override
public int getTimeToNextTick(final long currentTime)
{
- if (_network.getScheduledTime() > 0)
+ final long elapsed = _nanoTime.getAsLong() - _startTimeNanos;
+ if (elapsed >= _timeoutNanos)
{
- return (int) (_timeoutTime - _network.getScheduledTime());
+ return 0;
}
-
- return (int) (_timeoutTime - currentTime);
+ final long remaining = _timeoutNanos - elapsed;
+ final long millis = TimeUnit.NANOSECONDS.toMillis(remaining);
+ return (int) Math.min(Integer.MAX_VALUE, millis + (remaining %
1_000_000L == 0 ? 0 : 1));
}
@Override
public int tick(final long currentTime)
{
- int nextTick = getTimeToNextTick(currentTime);
+ final int nextTick = getTimeToNextTick(currentTime);
if(nextTick <= 0)
{
- _network.close();
+ _network.abort();
}
return nextTick;
}
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
index 257bf7f430..e78fff1345 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
@@ -120,6 +120,10 @@ public interface AMQPConnection<C extends
AMQPConnection<C>>
void sendConnectionCloseAsync(CloseReason reason, String description);
+ void startCloseDeadline();
+
+ CompletableFuture<Void> getTransportClosedFuture();
+
boolean isIOThread();
CompletableFuture<Void> doOnIOThreadAsync(final Runnable task);
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/AbstractAMQPConnection.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/AbstractAMQPConnection.java
index 0195eb3692..1e65641110 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/AbstractAMQPConnection.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/AbstractAMQPConnection.java
@@ -64,6 +64,7 @@ import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.TaskExecutorProvider;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.protocol.ConnectionClosingTicker;
import org.apache.qpid.server.security.AccessDeniedException;
import org.apache.qpid.server.security.SubjectExecutionContext;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
@@ -114,6 +115,7 @@ public abstract class AbstractAMQPConnection<C extends
AbstractAMQPConnection<C,
private String _clientId;
private volatile boolean _stopped;
+ private boolean _closeDeadlineStarted;
private final AtomicLong _messagesIn = new AtomicLong();
private final AtomicLong _messagesOut = new AtomicLong();
@@ -325,6 +327,24 @@ public abstract class AbstractAMQPConnection<C extends
AbstractAMQPConnection<C,
_stopped = true;
}
+ @Override
+ public final void startCloseDeadline()
+ {
+ synchronized (this)
+ {
+ if (_closeDeadlineStarted || _transportClosedFuture.isDone())
+ {
+ return;
+ }
+ final long timeoutMillis = getContextValue(Long.class,
Connection.CLOSE_RESPONSE_TIMEOUT);
+ final ConnectionClosingTicker deadline = new
ConnectionClosingTicker(timeoutMillis, _network);
+ _network.setCloseDeadline(deadline);
+ _aggregateTicker.addTicker(deadline);
+ _closeDeadlineStarted = true;
+ }
+ notifyWork();
+ }
+
@Override
public boolean isConnectionStopped()
{
@@ -799,6 +819,12 @@ public abstract class AbstractAMQPConnection<C extends
AbstractAMQPConnection<C,
return getSessionModels().size();
}
+ @Override
+ public final CompletableFuture<Void> getTransportClosedFuture()
+ {
+ return _transportClosedFuture.copy();
+ }
+
protected void markTransportClosed()
{
_transportClosedFuture.complete(null);
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
index 1d0cc188fe..f721a99c81 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
@@ -44,6 +44,7 @@ import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.logging.messages.NetworkMessages;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.transport.network.Ticker;
import org.apache.qpid.server.transport.network.TransportEncryption;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.util.SystemUtils;
@@ -83,6 +84,8 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
private final AtomicBoolean _hasShutdown = new AtomicBoolean();
private volatile long _bufferedSize;
private String _selectedHost;
+ private volatile boolean _aborted;
+ private volatile Ticker _closeDeadline;
public NonBlockingConnection(final SocketChannel socketChannel,
final ProtocolEngine protocolEngine,
@@ -158,6 +161,22 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
return this;
}
+ @Override
+ public synchronized void setCloseDeadline(final Ticker deadline)
+ {
+ if (_closeDeadline == null)
+ {
+ _closeDeadline = deadline;
+ }
+ }
+
+ @Override
+ public void abort()
+ {
+ _aborted = true;
+ close();
+ }
+
@Override
public void close()
{
@@ -276,29 +295,50 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
final boolean processPendingComplete = processPending();
- if (processPendingComplete)
+ if (isCloseDeadlineExpired())
{
- _pendingIterator = null;
+ abort();
}
- if (processPendingComplete ||
_delegate.needsReadForWrite())
+ if (!_closed.get())
{
- final boolean dataRead = doRead();
- doWrite();
+ if (processPendingComplete)
+ {
+ _pendingIterator = null;
+ }
- if (_fullyWritten && (dataRead ||
(_delegate.needsWork() &&
- _delegate.getNetInputBuffer().position() !=
0)))
+ if (processPendingComplete ||
_delegate.needsReadForWrite())
{
- _protocolEngine.notifyWork();
+ final boolean dataRead = doRead();
+ // Processing incoming frames may close the
connection or start its close deadline.
+ if (isCloseDeadlineExpired())
+ {
+ abort();
+ }
+
+ if (!_closed.get())
+ {
+ doWrite();
+
+ if (_fullyWritten && (dataRead ||
(_delegate.needsWork() &&
+
_delegate.getNetInputBuffer().position() != 0)))
+ {
+ _protocolEngine.notifyWork();
+ }
+ }
}
- }
- if (!processPendingComplete && (_fullyWritten ||
getBufferedSize() < _port.getNetworkBufferSize()))
- {
- _protocolEngine.notifyWork();
- }
+ if (!_closed.get())
+ {
+ if (!processPendingComplete &&
+ (_fullyWritten || getBufferedSize() <
_port.getNetworkBufferSize()))
+ {
+ _protocolEngine.notifyWork();
+ }
-
_protocolEngine.setTransportBlockedForWriting(!_fullyWritten);
+
_protocolEngine.setTransportBlockedForWriting(!_fullyWritten);
+ }
+ }
}
}
@@ -358,7 +398,7 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
final int networkBufferSize = _port.getNetworkBufferSize();
- while(_pendingIterator.hasNext())
+ while (!_closed.get() && !isCloseDeadlineExpired() &&
_pendingIterator.hasNext())
{
long size = getBufferedSize();
if(size >= networkBufferSize)
@@ -377,6 +417,11 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
}
}
+ if (_closed.get() || isCloseDeadlineExpired())
+ {
+ return false;
+ }
+
boolean complete = !_pendingIterator.hasNext();
if (getBufferedSize() >= networkBufferSize)
{
@@ -400,10 +445,16 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
try
{
- shutdownInput();
- shutdownFinalWrite();
- _protocolEngine.closed();
- shutdownOutput();
+ try
+ {
+ shutdownInput();
+ shutdownFinalWrite();
+ _protocolEngine.closed();
+ }
+ finally
+ {
+ shutdownOutput();
+ }
}
finally
{
@@ -443,7 +494,7 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
{
final long startTime = System.currentTimeMillis();
int cnt = 0;
- while (!doWrite())
+ while (!checkForWriteAbort() && !doWrite())
{
if (cnt % _finalWriteThreshold == 0 &&
System.currentTimeMillis() - startTime > _finalWriteTimeout)
{
@@ -459,6 +510,21 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
}
}
+ private boolean isCloseDeadlineExpired()
+ {
+ final Ticker deadline = _closeDeadline;
+ return deadline != null &&
deadline.getTimeToNextTick(System.currentTimeMillis()) <= 0;
+ }
+
+ boolean checkForWriteAbort()
+ {
+ if (!_aborted && isCloseDeadlineExpired())
+ {
+ abort();
+ }
+ return _aborted;
+ }
+
private void shutdownOutput()
{
try
@@ -538,6 +604,10 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
long writeToTransport(final Collection<QpidByteBuffer> buffers) throws
IOException
{
+ if (checkForWriteAbort())
+ {
+ return 0L;
+ }
final long written = QpidByteBuffer.write(_socketChannel, buffers,
_maxGatheringWriteBuffers);
if (LOGGER.isDebugEnabled())
{
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java
index d860a8216c..8175552aac 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java
@@ -139,7 +139,7 @@ public class NonBlockingConnectionTLSDelegate implements
NonBlockingConnectionDe
LOGGER.debug("SSLEngine closed, discarded {} byte(s)",
remaining);
}
- tasksRun = runSSLEngineTasks(_status);
+ tasksRun = runSSLEngineTasks(_status, false);
if(_applicationBuffer.position() > oldAppBufPos)
{
readData = true;
@@ -175,11 +175,15 @@ public class NonBlockingConnectionTLSDelegate implements
NonBlockingConnectionDe
}
@Override
- public WriteResult doWrite(Collection<QpidByteBuffer> buffers) throws
IOException
+ public WriteResult doWrite(final Collection<QpidByteBuffer> buffers)
throws IOException
{
final int bufCount = buffers.size();
- int totalConsumed = wrapBufferArray(buffers);
+ final int totalConsumed = wrapBufferArray(buffers);
+ if (_parent.checkForWriteAbort())
+ {
+ return new WriteResult(false, totalConsumed);
+ }
boolean bufsSent = true;
final Iterator<QpidByteBuffer> itr = buffers.iterator();
@@ -257,12 +261,16 @@ public class NonBlockingConnectionTLSDelegate implements
NonBlockingConnectionDe
}
- private int wrapBufferArray(Collection<QpidByteBuffer> buffers) throws
SSLException
+ private int wrapBufferArray(final Collection<QpidByteBuffer> buffers)
throws SSLException
{
int totalConsumed = 0;
boolean encrypted;
do
{
+ if (_parent.checkForWriteAbort())
+ {
+ return totalConsumed;
+ }
if(_sslEngine.getHandshakeStatus() !=
SSLEngineResult.HandshakeStatus.NEED_UNWRAP)
{
if(_netOutputBuffer.remaining() <
_sslEngine.getSession().getPacketBufferSize())
@@ -297,7 +305,11 @@ public class NonBlockingConnectionTLSDelegate implements
NonBlockingConnectionDe
}
encrypted = _status.bytesProduced() > 0;
totalConsumed += _status.bytesConsumed();
- runSSLEngineTasks(_status);
+ runSSLEngineTasks(_status, true);
+ if (_parent.checkForWriteAbort())
+ {
+ return totalConsumed;
+ }
if(encrypted && _netOutputBuffer.remaining() <
_sslEngine.getSession().getPacketBufferSize())
{
_netOutputBuffer.flip();
@@ -352,12 +364,12 @@ public class NonBlockingConnectionTLSDelegate implements
NonBlockingConnectionDe
return totalConsumed;
}
- private boolean runSSLEngineTasks(final SSLEngineResult status)
+ private boolean runSSLEngineTasks(final SSLEngineResult status, final
boolean writing)
{
if(status.getHandshakeStatus() ==
SSLEngineResult.HandshakeStatus.NEED_TASK)
{
Runnable task;
- while((task = _sslEngine.getDelegatedTask()) != null)
+ while ((!writing || !_parent.checkForWriteAbort()) && (task =
_sslEngine.getDelegatedTask()) != null)
{
task.run();
}
@@ -461,7 +473,11 @@ public class NonBlockingConnectionTLSDelegate implements
NonBlockingConnectionDe
@Override
public void shutdownOutput()
{
-
+ for (final QpidByteBuffer buffer : _encryptedOutput)
+ {
+ buffer.dispose();
+ }
+ _encryptedOutput.clear();
if (_netOutputBuffer != null)
{
_netOutputBuffer.dispose();
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/ServerNetworkConnection.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/ServerNetworkConnection.java
index eab77cb883..ba75cc9739 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/ServerNetworkConnection.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/ServerNetworkConnection.java
@@ -23,9 +23,19 @@ import java.net.InetSocketAddress;
import java.net.SocketAddress;
import org.apache.qpid.server.transport.network.NetworkConnection;
+import org.apache.qpid.server.transport.network.Ticker;
public interface ServerNetworkConnection extends NetworkConnection
{
+ default void setCloseDeadline(final Ticker deadline)
+ {
+ }
+
+ default void abort()
+ {
+ close();
+ }
+
String getTransportInfo();
long getScheduledTime();
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/protocol/ConnectionClosingTickerTest.java
b/broker-core/src/test/java/org/apache/qpid/server/protocol/ConnectionClosingTickerTest.java
new file mode 100644
index 0000000000..c93711f7c5
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/protocol/ConnectionClosingTickerTest.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.qpid.server.protocol;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import org.apache.qpid.server.transport.ServerNetworkConnection;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class ConnectionClosingTickerTest extends UnitTestBase
+{
+ private final AtomicLong _nanoTime = new AtomicLong();
+ private final ServerNetworkConnection _network =
mock(ServerNetworkConnection.class);
+
+ @BeforeEach
+ void setUp()
+ {
+ _nanoTime.set(0);
+ reset(_network);
+ }
+
+ @Test
+ void expiresAtDeadlineAndRoundsUpUntilThen()
+ {
+ final ConnectionClosingTicker ticker = new
ConnectionClosingTicker(1000, _network, _nanoTime::get);
+ assertEquals(1000, ticker.tick(0));
+ _nanoTime.set(TimeUnit.MILLISECONDS.toNanos(999) + 1);
+ assertEquals(1, ticker.tick(0));
+ verify(_network, never()).abort();
+
+ _nanoTime.set(TimeUnit.MILLISECONDS.toNanos(1000));
+ assertEquals(0, ticker.tick(0));
+ verify(_network).abort();
+ verify(_network, never()).close();
+ }
+
+ @Test
+ void wallClockChangesAndWorkerSchedulingDoNotExtendDeadline()
+ {
+ when(_network.getScheduledTime()).thenReturn(1L);
+ final ConnectionClosingTicker ticker = new
ConnectionClosingTicker(1000, _network, _nanoTime::get);
+ _nanoTime.set(TimeUnit.MILLISECONDS.toNanos(500));
+ assertEquals(500, ticker.getTimeToNextTick(Long.MIN_VALUE));
+ assertEquals(500, ticker.getTimeToNextTick(Long.MAX_VALUE));
+ _nanoTime.set(TimeUnit.MILLISECONDS.toNanos(1000));
+ assertEquals(0, ticker.tick(0));
+ verify(_network).abort();
+ }
+
+ @Test
+ void nanoTimeWrapDoesNotExtendDeadline()
+ {
+ final long start = Long.MAX_VALUE - TimeUnit.MILLISECONDS.toNanos(500);
+ _nanoTime.set(start);
+ final ConnectionClosingTicker ticker = new
ConnectionClosingTicker(1000, _network, _nanoTime::get);
+ _nanoTime.set(start + TimeUnit.MILLISECONDS.toNanos(750));
+ assertEquals(250, ticker.tick(0));
+ _nanoTime.set(start + TimeUnit.MILLISECONDS.toNanos(1000));
+ assertEquals(0, ticker.tick(0));
+ verify(_network).abort();
+ }
+
+ @ParameterizedTest
+ @ValueSource(longs = {0, -1, Long.MIN_VALUE})
+ void nonPositiveTimeoutExpiresImmediately(final long timeout)
+ {
+ final ConnectionClosingTicker ticker = new
ConnectionClosingTicker(timeout, _network, _nanoTime::get);
+ assertEquals(0, ticker.tick(0));
+ verify(_network).abort();
+ }
+
+ @ParameterizedTest
+ @ValueSource(longs = {Integer.MAX_VALUE, (long) Integer.MAX_VALUE + 1,
Long.MAX_VALUE})
+ void largeTimeoutDoesNotOverflowToImmediateExpiry(final long timeout)
+ {
+ final ConnectionClosingTicker ticker = new
ConnectionClosingTicker(timeout, _network, _nanoTime::get);
+ assertEquals(Integer.MAX_VALUE, ticker.tick(0));
+ verify(_network, never()).abort();
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NativeCloseDeadlineTestBase.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NativeCloseDeadlineTestBase.java
new file mode 100644
index 0000000000..c9fe2e4253
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NativeCloseDeadlineTestBase.java
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.qpid.server.transport;
+
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createPort;
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createSocketChannel;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.configuration.updater.CurrentThreadTaskExecutor;
+import org.apache.qpid.server.configuration.updater.TaskExecutor;
+import org.apache.qpid.server.logging.EventLogger;
+import org.apache.qpid.server.model.AuthenticationProvider;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.BrokerModel;
+import org.apache.qpid.server.model.Connection;
+import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.Protocol;
+import org.apache.qpid.server.model.SystemConfig;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.plugin.ProtocolEngineCreator;
+import org.apache.qpid.server.transport.network.Ticker;
+import org.apache.qpid.server.transport.network.TransportEncryption;
+import org.apache.qpid.server.util.Action;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+public abstract class NativeCloseDeadlineTestBase extends UnitTestBase
+{
+ protected static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private final AtomicBoolean _writable = new AtomicBoolean();
+
+ protected Broker<?> _broker;
+ protected AmqpPort<?> _port;
+ protected NonBlockingConnection _network;
+ protected AggregateTicker _ticker;
+
+ private SocketChannel _socket;
+ private TaskExecutor _taskExecutor;
+ private MultiVersionProtocolEngine _engine;
+
+ @BeforeEach
+ void setUpNativeTransport() throws Exception
+ {
+ _writable.set(false);
+ _taskExecutor = CurrentThreadTaskExecutor.newStartedInstance();
+ final SystemConfig<?> systemConfig = mock(SystemConfig.class);
+ when(systemConfig.getEventLogger()).thenReturn(new EventLogger());
+ _broker = mock(Broker.class);
+ doReturn(systemConfig).when(_broker).getParent();
+ doReturn(Broker.class).when(_broker).getCategoryClass();
+ when(_broker.getModel()).thenReturn(BrokerModel.getInstance());
+ when(_broker.getTaskExecutor()).thenReturn(_taskExecutor);
+ when(_broker.getChildExecutor()).thenReturn(_taskExecutor);
+ when(_broker.getEventLogger()).thenReturn(new EventLogger());
+ when(_broker.getNetworkBufferSize()).thenReturn(NETWORK_BUFFER_SIZE);
+
+ _port = createPort(_broker, NETWORK_BUFFER_SIZE, 0L);
+ doReturn(Port.class).when(_port).getCategoryClass();
+ when(_port.getModel()).thenReturn(BrokerModel.getInstance());
+ when(_port.getTaskExecutor()).thenReturn(_taskExecutor);
+ when(_port.getChildExecutor()).thenReturn(_taskExecutor);
+ setCloseTimeout(30_000);
+ final AuthenticationProvider<?> authenticationProvider =
mock(AuthenticationProvider.class);
+
when(authenticationProvider.getAvailableMechanisms(anyBoolean())).thenReturn(List.of("PLAIN"));
+
doReturn(authenticationProvider).when(_port).getAuthenticationProvider();
+
+ _socket = createSocketChannel(_writable::get);
+ _engine = new MultiVersionProtocolEngine(_broker, Set.of(), null,
_port, Transport.TCP, 1L,
+ new ProtocolEngineCreator[0], null);
+ _ticker = _engine.getAggregateTicker();
+ _network = new NonBlockingConnection(_socket, _engine,
Set.of(TransportEncryption.NONE), () -> { },
+ mock(NetworkConnectionScheduler.class), _port);
+ _network.setSelectionTask(mock(SelectorThread.SelectionTask.class));
+ }
+
+ @AfterEach
+ void tearDownNativeTransport()
+ {
+ try
+ {
+ if (_network != null)
+ {
+ _writable.set(true);
+ _network.close();
+ _network.doWork();
+ }
+ }
+ finally
+ {
+ _taskExecutor.stop();
+ }
+ }
+
+ protected abstract AMQPConnection<?> createConnection(final Protocol
protocol) throws Exception;
+
+ protected abstract void receivePeerClose(final AMQPConnection<?>
connection) throws Exception;
+
+ protected abstract Stream<Protocol> protocols();
+
+ protected void setCloseTimeout(final long timeoutMillis)
+ {
+
when(_port.getContext()).thenReturn(Map.of(Connection.CLOSE_RESPONSE_TIMEOUT,
Long.toString(timeoutMillis)));
+ }
+
+ protected Stream<Arguments> protocolsAndSizes()
+ {
+ return protocols().flatMap(protocol -> Stream.of(NETWORK_BUFFER_SIZE -
1, NETWORK_BUFFER_SIZE,
+ 2 * NETWORK_BUFFER_SIZE).map(size -> Arguments.of(protocol,
size)));
+ }
+
+ protected void installProtocol(final ProtocolEngine protocol) throws
Exception
+ {
+ final Field delegate =
MultiVersionProtocolEngine.class.getDeclaredField("_delegate");
+ delegate.setAccessible(true);
+ delegate.set(_engine, protocol);
+ final Field listener =
MultiVersionProtocolEngine.class.getDeclaredField("_workListener");
+ listener.setAccessible(true);
+ protocol.setWorkListener(((AtomicReference<Action<ProtocolEngine>>)
listener.get(_engine)).get());
+ }
+
+ @ParameterizedTest
+ @MethodSource("protocolsAndSizes")
+ protected void testRequestedCloseHasDeadlineWithBlockedOutput(final
Protocol protocol, final int bufferedBytes)
+ throws Exception
+ {
+ setCloseTimeout(0);
+ final AMQPConnection<?> connection = createConnection(protocol);
+ fillOutput(bufferedBytes);
+
connection.sendConnectionCloseAsync(AMQPConnection.CloseReason.MANAGEMENT,
"close");
+
+ assertTrue(_network.doWork(), protocol + ": requested close remained
pending with " + bufferedBytes +
+ " buffered bytes; next ticker=" +
_ticker.getTimeToNextTick(System.currentTimeMillis()));
+ verify(_socket).close();
+ }
+
+ @ParameterizedTest
+ @MethodSource("protocols")
+ protected void testCloseResumesAfterSocketDrains(final Protocol protocol)
throws Exception
+ {
+ final AMQPConnection<?> connection = createConnection(protocol);
+ fillOutput(NETWORK_BUFFER_SIZE);
+
connection.sendConnectionCloseAsync(AMQPConnection.CloseReason.MANAGEMENT,
"close");
+ assertFalse(_network.doWork());
+ verify(_socket, never()).close();
+
+ _writable.set(true);
+ assertFalse(_network.doWork());
+ assertTrue(_ticker.getTimeToNextTick(System.currentTimeMillis()) <
Integer.MAX_VALUE,
+ "Draining the socket must retain the deadline while awaiting
the peer's close response");
+ assertTrue(connection.isClosing(), "The protocol close must be sent
when output can drain");
+ verify(_socket, never()).close();
+ }
+
+ @ParameterizedTest
+ @MethodSource("protocols")
+ protected void testPeerCloseIsBoundedWithBlockedOutput(final Protocol
protocol) throws Exception
+ {
+ setCloseTimeout(0);
+ final AMQPConnection<?> connection = createConnection(protocol);
+ fillOutput(128);
+ _engine.setIOThread(Thread.currentThread());
+ try
+ {
+ receivePeerClose(connection);
+ }
+ finally
+ {
+ _engine.setIOThread(null);
+ }
+ assertEquals(1, getTickers().size(), "Peer close must start the same
bounded closing phase");
+ assertTrue(_network.doWork(), protocol + ": peer close did not reach
native shutdown while output was paused");
+ verify(_socket).close();
+ }
+
+ @ParameterizedTest
+ @MethodSource("protocols")
+ protected void testPeerCloseDuringReadSkipsExpiredWrite(final Protocol
protocol) throws Exception
+ {
+ setCloseTimeout(0);
+ final AMQPConnection<?> connection = createConnection(protocol);
+ _writable.set(true);
+ when(_socket.read(any(ByteBuffer.class))).thenAnswer(invocation ->
+ {
+ // Invoke the protocol handler within doRead so its deadline
starts after the initial expiry check.
+ receivePeerClose(connection);
+ return 0;
+ });
+
+ assertTrue(_network.doWork(), protocol + ": close received during read
must finish in this work cycle");
+
+ verify(_socket).read(any(ByteBuffer.class));
+ verify(_socket, never()).write(any(ByteBuffer[].class));
+ verify(_socket).close();
+ }
+
+ @ParameterizedTest
+ @MethodSource("protocols")
+ protected void testRepeatedCloseKeepsOriginalDeadline(final Protocol
protocol) throws Exception
+ {
+ final AMQPConnection<?> connection = createConnection(protocol);
+ fillOutput(NETWORK_BUFFER_SIZE);
+ IntStream.range(0, 32).parallel().forEach(i ->
+
connection.sendConnectionCloseAsync(AMQPConnection.CloseReason.MANAGEMENT,
"First close"));
+ final List<Ticker> tickers = getTickers();
+ assertEquals(1, tickers.size(), "Closing must arm a deadline before
any queued task runs");
+ final Ticker original = tickers.get(0);
+
+ IntStream.range(0, 32).parallel().forEach(i ->
+
connection.sendConnectionCloseAsync(AMQPConnection.CloseReason.MANAGEMENT,
"Repeated close"));
+ _writable.set(true);
+ _network.doWork();
+
+ assertEquals(1, tickers.size(), "Repeated requests and the close task
must not add another timer");
+ assertSame(original, tickers.get(0));
+ }
+
+ private List<Ticker> getTickers() throws ReflectiveOperationException
+ {
+ final Field tickers =
AggregateTicker.class.getDeclaredField("_tickers");
+ tickers.setAccessible(true);
+ return (List<Ticker>) tickers.get(_ticker);
+ }
+
+ private void fillOutput(final int size)
+ {
+ try (final QpidByteBuffer buffer = QpidByteBuffer.wrap(new byte[size]))
+ {
+ _network.send(buffer);
+ }
+ assertFalse(_network.doWork());
+ assertTrue(_network.wantsWrite(), "The native socket must retain
unwritten output");
+ assertEquals(Integer.MAX_VALUE,
_ticker.getTimeToNextTick(System.currentTimeMillis()));
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionCloseSchedulingTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionCloseSchedulingTest.java
new file mode 100644
index 0000000000..1d7e0e09f5
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionCloseSchedulingTest.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.qpid.server.transport;
+
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createPort;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.protocol.ConnectionClosingTicker;
+import org.apache.qpid.server.transport.network.TransportEncryption;
+import org.apache.qpid.server.util.Action;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class NonBlockingConnectionCloseSchedulingTest extends UnitTestBase
+{
+ private static final int TIMEOUT_MILLIS = 10000;
+ private final List<Thread> _threads = new CopyOnWriteArrayList<>();
+ private NetworkConnectionScheduler _scheduler;
+
+ @BeforeEach
+ void startScheduler()
+ {
+ _threads.clear();
+ _scheduler = new NetworkConnectionScheduler(getTestName(), 1, 2, 1L,
task ->
+ {
+ final Thread thread = new Thread(task, getTestName() + "-" +
_threads.size());
+ _threads.add(thread);
+ return thread;
+ });
+ _scheduler.start();
+ }
+
+ @AfterEach
+ void stopScheduler() throws InterruptedException
+ {
+ _scheduler.close();
+ try
+ {
+ for (final Thread thread : _threads)
+ {
+ thread.join(TIMEOUT_MILLIS);
+ assertFalse(thread.isAlive(), "Scheduler thread did not
terminate: " + thread.getName());
+ }
+ }
+ finally
+ {
+ for (final Thread thread : _threads)
+ {
+ thread.interrupt();
+ }
+ for (final Thread thread : _threads)
+ {
+ thread.join(TIMEOUT_MILLIS);
+ }
+ }
+ }
+
+ @Test
+ void deadlineClosesBlockedSocketWithoutPeerActivity() throws Exception
+ {
+ final AggregateTicker ticker = new AggregateTicker();
+ final ProtocolEngine engine = createProtocolEngine(ticker);
+ final CompletableFuture<Void> blocked = new CompletableFuture<>();
+ final CompletableFuture<Void> closed = new CompletableFuture<>();
+ try (final ServerSocketChannel listener = ServerSocketChannel.open();
+ final Socket client = new Socket();
+ final SocketChannel channel = acceptConnection(listener, client))
+ {
+ channel.configureBlocking(false);
+ channel.socket().setSendBufferSize(4096);
+ final NonBlockingConnection connection = createConnection(channel,
engine, blocked, closed);
+ final AtomicBoolean pendingTaskRan = new AtomicBoolean();
+ final List<Runnable> work = List.of(() ->
+ {
+ try (final QpidByteBuffer output = QpidByteBuffer.wrap(new
byte[2 * 1024 * 1024]))
+ {
+ connection.send(output);
+ }
+ }, () -> pendingTaskRan.set(true));
+ when(engine.processPendingIterator()).thenReturn(work.iterator());
+ _scheduler.addConnection(connection);
+ try
+ {
+ engine.notifyWork();
+ blocked.get(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+ ticker.addTicker(new ConnectionClosingTicker(100, connection));
+ engine.notifyWork();
+ closed.get(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+
+ assertFalse(pendingTaskRan.get(), "The socket must close while
protocol work is still blocked");
+ assertFalse(channel.isOpen());
+ }
+ finally
+ {
+ connection.abort();
+ closed.get(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+ }
+ }
+ }
+
+ private SocketChannel acceptConnection(final ServerSocketChannel listener,
final Socket client) throws IOException
+ {
+ listener.bind(new InetSocketAddress(InetAddress.getLoopbackAddress(),
0));
+ client.setReceiveBufferSize(4096);
+ client.connect(listener.getLocalAddress(), TIMEOUT_MILLIS);
+ return listener.accept();
+ }
+
+ private ProtocolEngine createProtocolEngine(final AggregateTicker ticker)
+ {
+ final ProtocolEngine engine = mock(ProtocolEngine.class);
+ final AtomicBoolean hasWork = new AtomicBoolean();
+ final AtomicReference<Action<ProtocolEngine>> workListener = new
AtomicReference<>();
+ when(engine.getAggregateTicker()).thenReturn(ticker);
+ when(engine.hasWork()).thenAnswer(invocation -> hasWork.get());
+ doAnswer(invocation ->
+ {
+ hasWork.set(false);
+ return null;
+ }).when(engine).clearWork();
+ doAnswer(invocation ->
+ {
+ workListener.set(invocation.getArgument(0));
+ return null;
+ }).when(engine).setWorkListener(any());
+ doAnswer(invocation ->
+ {
+ hasWork.set(true);
+ workListener.get().performAction(engine);
+ return null;
+ }).when(engine).notifyWork();
+ return engine;
+ }
+
+ private NonBlockingConnection createConnection(final SocketChannel channel,
+ final ProtocolEngine engine,
+ final
CompletableFuture<Void> blocked,
+ final
CompletableFuture<Void> closed)
+ {
+ return new NonBlockingConnection(channel, engine,
Set.of(TransportEncryption.NONE), () -> { },
+ _scheduler, createPort(1024, 1000L))
+ {
+ @Override
+ public boolean doWork()
+ {
+ final boolean result = super.doWork();
+ if (result)
+ {
+ closed.complete(null);
+ }
+ else if (wantsWrite())
+ {
+ blocked.complete(null);
+ }
+ return result;
+ }
+ };
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionCloseTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionCloseTest.java
new file mode 100644
index 0000000000..a753abb6c8
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionCloseTest.java
@@ -0,0 +1,770 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.qpid.server.transport;
+
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createPort;
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createSocketChannel;
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLSession;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.InOrder;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.protocol.ConnectionClosingTicker;
+import org.apache.qpid.server.transport.network.Ticker;
+import org.apache.qpid.server.transport.network.TransportEncryption;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class NonBlockingConnectionCloseTest extends UnitTestBase
+{
+ private static final int BUFFER_SIZE = 1024;
+ private final AtomicBoolean _writable = new AtomicBoolean();
+ private AggregateTicker _ticker;
+ private SocketChannel _socket;
+ private ProtocolEngine _engine;
+ private NetworkConnectionScheduler _scheduler;
+ private SelectorThread.SelectionTask _selectionTask;
+ private AmqpPort<?> _port;
+ private SSLEngine _sslEngine;
+ private NonBlockingConnection _connection;
+
+ @BeforeEach
+ void setUp() throws Exception
+ {
+ _ticker = new AggregateTicker();
+ _writable.set(false);
+ _connection = null;
+ _sslEngine = null;
+ QpidByteBuffer.deinitialisePool();
+ QpidByteBuffer.initialisePool(BUFFER_SIZE, 16, 0.5);
+ _socket = createSocketChannel(_writable::get);
+ _engine = mock(ProtocolEngine.class);
+ when(_engine.getAggregateTicker()).thenReturn(_ticker);
+ when(_engine.processPendingIterator()).thenAnswer(invocation ->
Collections.emptyIterator());
+ _scheduler = mock(NetworkConnectionScheduler.class);
+ _selectionTask = mock(SelectorThread.SelectionTask.class);
+ _port = createPort(BUFFER_SIZE, 1000L);
+ }
+
+ @AfterEach
+ void tearDown()
+ {
+ try
+ {
+ if (_connection != null)
+ {
+ _connection.abort();
+ _connection.doWork();
+ }
+ }
+ finally
+ {
+ QpidByteBuffer.deinitialisePool();
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void expiredDeadlineDiscardsOutputWithoutFinalWrite(final
TransportEncryption encryption) throws Exception
+ {
+ createConnection(encryption);
+ fillOutput();
+ _ticker.addTicker(new ConnectionClosingTicker(0, _connection));
+ clearInvocations(_socket, _engine, _selectionTask);
+
+ assertTrue(_connection.doWork());
+ assertTrue(_connection.doWork(), "Shutdown must be idempotent");
+
+ verify(_socket, never()).write(any(ByteBuffer[].class));
+ verify(_engine, never()).processPendingIterator();
+ verify(_selectionTask).wakeup();
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void deadlineExpiresDuringPendingWork(final TransportEncryption
encryption) throws Exception
+ {
+ createConnection(encryption);
+ _writable.set(true);
+ final AtomicBoolean expired = new AtomicBoolean();
+ installDeadline(expired);
+ final Runnable pending = mock(Runnable.class);
+ final List<Runnable> work = List.of(() ->
+ {
+ queueOutput();
+ expired.set(true);
+ }, pending);
+ when(_engine.processPendingIterator()).thenReturn(work.iterator());
+
+ assertTrue(_connection.doWork(), "Deadline expiry must stop the active
pending-work loop");
+
+ verify(pending, never()).run();
+ verify(_socket, never()).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void deadlineInstalledByLastPendingTaskDiscardsOutput(final
TransportEncryption encryption) throws Exception
+ {
+ createConnection(encryption);
+ _writable.set(true);
+ final List<Runnable> work = List.of(() ->
+ {
+ queueOutput();
+ _connection.setCloseDeadline(new ConnectionClosingTicker(0,
_connection));
+ });
+ when(_engine.processPendingIterator()).thenReturn(work.iterator());
+
+ assertTrue(_connection.doWork(), "A deadline installed during
processing must prevent the final pending write");
+
+ verify(_socket, never()).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @CsvSource({"NONE, true", "NONE, false", "TLS, true", "TLS, false"})
+ void deadlineExpiresDuringSuccessfulWrite(final TransportEncryption
encryption,
+ final boolean pendingWork)
throws Exception
+ {
+ createConnection(encryption);
+ final AtomicBoolean expired = new AtomicBoolean();
+ installDeadline(expired);
+ queueOutput();
+ final Runnable pending = mock(Runnable.class);
+ when(_engine.processPendingIterator()).thenReturn(pendingWork ?
+ List.of(pending).iterator() : Collections.emptyIterator());
+ when(_socket.write(any(ByteBuffer[].class))).thenAnswer(invocation ->
+ {
+ long written = 0;
+ for (final ByteBuffer buffer : (ByteBuffer[])
invocation.getArgument(0))
+ {
+ written += buffer.remaining();
+ buffer.position(buffer.limit());
+ }
+ expired.set(true);
+ return written;
+ });
+
+ assertTrue(_connection.doWork(), "Successful writes must not postpone
deadline enforcement");
+
+ verify(pending, never()).run();
+ verify(_socket).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void abortDuringPendingWorkStopsFurtherTasks(final TransportEncryption
encryption) throws Exception
+ {
+ createConnection(encryption);
+ _writable.set(true);
+ final Runnable pending = mock(Runnable.class);
+ final List<Runnable> work = List.of(() ->
+ {
+ queueOutput();
+ _connection.abort();
+ }, pending);
+ when(_engine.processPendingIterator()).thenReturn(work.iterator());
+
+ assertTrue(_connection.doWork());
+
+ verify(pending, never()).run();
+ verify(_socket, never()).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @Test
+ void deadlineExpiresAfterLastTlsWriteSkipsFurtherIo() throws Exception
+ {
+ createConnection(TransportEncryption.TLS);
+ final AtomicBoolean expired = new AtomicBoolean();
+ installDeadline(expired);
+ try (final QpidByteBuffer output =
QpidByteBuffer.allocateDirect(BUFFER_SIZE))
+ {
+ _connection.send(output);
+ }
+ when(_socket.write(any(ByteBuffer[].class))).thenAnswer(invocation ->
+ {
+ long written = 0;
+ for (final ByteBuffer buffer : (ByteBuffer[])
invocation.getArgument(0))
+ {
+ written += buffer.remaining();
+ buffer.position(buffer.limit());
+ }
+ expired.set(true);
+ // Only wrapping after expiry matters, regardless of how the
output was batched.
+ clearInvocations(_sslEngine);
+ return written;
+ });
+
+ assertTrue(_connection.doWork());
+
+ verify(_socket).write(any(ByteBuffer[].class));
+ verify(_sslEngine, never()).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class));
+ verify(_engine, never()).setTransportBlockedForWriting(anyBoolean());
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false, 512", "false, 2048", "true, 512", "true, 2048"})
+ void deadlineExpiresDuringTlsWrap(final boolean finalWrite, final int
bufferedBytes) throws Exception
+ {
+ createConnection(TransportEncryption.TLS);
+ _writable.set(true);
+ final AtomicBoolean expired = new AtomicBoolean();
+ installDeadline(expired);
+ final Runnable delegatedTask = mock(Runnable.class);
+ when(_sslEngine.getDelegatedTask()).thenReturn(delegatedTask,
(Runnable) null);
+ doAnswer(invocation ->
+ {
+ final SSLEngineResult result =
wrapTlsOutput(invocation.getArgument(0), invocation.getArgument(1),
+ SSLEngineResult.HandshakeStatus.NEED_TASK);
+ expired.set(true);
+ return result;
+ }).when(_sslEngine).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class));
+ queueOutput(bufferedBytes);
+ if (finalWrite)
+ {
+ _connection.close();
+ }
+
+ assertTrue(_connection.doWork());
+
+ assertAll(() -> verify(_sslEngine).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class)),
+ () -> verify(delegatedTask, never()).run(),
+ () -> verify(_socket,
never()).write(any(ByteBuffer[].class)),
+ this::assertClosedAndReleased);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void deadlineExpiresDuringLastTlsWrap(final boolean finalWrite) throws
Exception
+ {
+ createConnection(TransportEncryption.TLS);
+ _writable.set(true);
+ final AtomicBoolean expired = new AtomicBoolean();
+ installDeadline(expired);
+ doAnswer(invocation ->
+ {
+ final SSLEngineResult result =
wrapTlsOutput(invocation.getArgument(0), invocation.getArgument(1),
+ SSLEngineResult.HandshakeStatus.NEED_UNWRAP);
+
when(_sslEngine.getHandshakeStatus()).thenReturn(SSLEngineResult.HandshakeStatus.NEED_UNWRAP);
+ expired.set(true);
+ return result;
+ }).when(_sslEngine).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class));
+ queueOutput(BUFFER_SIZE / 2);
+ if (finalWrite)
+ {
+ _connection.close();
+ }
+
+ assertTrue(_connection.doWork());
+
+ verify(_sslEngine).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class));
+ verify(_socket, never()).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void abortDuringTlsWrapStopsFurtherOutput(final boolean finalWrite) throws
Exception
+ {
+ createConnection(TransportEncryption.TLS);
+ _writable.set(true);
+ doAnswer(invocation ->
+ {
+ final SSLEngineResult result =
wrapTlsOutput(invocation.getArgument(0), invocation.getArgument(1),
+ SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING);
+ _connection.abort();
+ return result;
+ }).when(_sslEngine).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class));
+ queueOutput();
+ if (finalWrite)
+ {
+ _connection.close();
+ }
+
+ assertTrue(_connection.doWork());
+
+ assertAll(() -> verify(_sslEngine).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class)),
+ () -> verify(_socket,
never()).write(any(ByteBuffer[].class)),
+ this::assertClosedAndReleased);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void deadlineExpiresDuringTlsTask(final boolean finalWrite) throws
Exception
+ {
+ createConnection(TransportEncryption.TLS);
+ _writable.set(true);
+ final AtomicBoolean expired = new AtomicBoolean();
+ installDeadline(expired);
+ final Runnable laterTask = mock(Runnable.class);
+ when(_sslEngine.getDelegatedTask()).thenReturn(() ->
expired.set(true), laterTask, null);
+ doAnswer(invocation -> wrapTlsOutput(invocation.getArgument(0),
invocation.getArgument(1),
+ SSLEngineResult.HandshakeStatus.NEED_TASK))
+ .when(_sslEngine).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class));
+ queueOutput();
+ if (finalWrite)
+ {
+ _connection.close();
+ }
+
+ assertTrue(_connection.doWork());
+
+ assertAll(() -> verify(_sslEngine).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class)),
+ () -> verify(laterTask, never()).run(),
+ () -> verify(_socket,
never()).write(any(ByteBuffer[].class)),
+ this::assertClosedAndReleased);
+ }
+
+ @Test
+ void gracefulTlsCloseWrapsOutputAndRunsTasksBeforeDeadline() throws
Exception
+ {
+ createConnection(TransportEncryption.TLS);
+ _writable.set(true);
+ installDeadline(new AtomicBoolean());
+ final Runnable delegatedTask = mock(Runnable.class);
+ when(_sslEngine.getDelegatedTask()).thenReturn(delegatedTask,
(Runnable) null);
+ doAnswer(invocation -> wrapTlsOutput(invocation.getArgument(0),
invocation.getArgument(1),
+ SSLEngineResult.HandshakeStatus.NEED_TASK))
+ .when(_sslEngine).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class));
+ queueOutput();
+ _connection.close();
+
+ assertTrue(_connection.doWork());
+
+ verify(delegatedTask).run();
+ verify(_socket).write(any(ByteBuffer[].class));
+ assertFalse(_connection.wantsWrite(), "Graceful close must wrap and
drain the plaintext backlog");
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void gracefulCloseDuringPendingWorkFlushesOutput(final TransportEncryption
encryption) throws Exception
+ {
+ createConnection(encryption);
+ _writable.set(true);
+ final Runnable pending = mock(Runnable.class);
+ final List<Runnable> work = List.of(() ->
+ {
+ queueOutput();
+ _connection.close();
+ }, pending);
+ when(_engine.processPendingIterator()).thenReturn(work.iterator());
+
+ assertTrue(_connection.doWork());
+
+ verify(pending, never()).run();
+ verify(_socket).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @Test
+ void expiryDoesNotWaitForTlsUnwrap() throws Exception
+ {
+ createConnection(TransportEncryption.TLS);
+
when(_sslEngine.getHandshakeStatus()).thenReturn(SSLEngineResult.HandshakeStatus.NEED_UNWRAP);
+ fillOutput();
+ _ticker.addTicker(new ConnectionClosingTicker(0, _connection));
+ clearInvocations(_socket, _sslEngine);
+
+ assertTrue(_connection.doWork());
+
+ verify(_socket, never()).write(any(ByteBuffer[].class));
+ verify(_sslEngine, never()).wrap(any(ByteBuffer[].class),
any(ByteBuffer.class));
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void abortCanOvertakeGracefulTransportClose(final TransportEncryption
encryption) throws Exception
+ {
+ createConnection(encryption);
+ fillOutput();
+ _connection.close();
+ _connection.abort();
+ clearInvocations(_socket);
+
+ assertTrue(_connection.doWork());
+
+ verify(_socket, never()).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void finalWriteRetriesCannotOutliveProtocolDeadline(final
TransportEncryption encryption) throws Exception
+ {
+ createConnection(encryption);
+ fillOutput();
+ final AtomicBoolean expired = new AtomicBoolean();
+ installDeadline(expired);
+ doAnswer(invocation ->
+ {
+ expired.set(true);
+ return 0L;
+ }).when(_socket).write(any(ByteBuffer[].class));
+ _connection.close();
+ clearInvocations(_socket);
+
+ assertTrue(_connection.doWork());
+
+ verify(_socket).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @Test
+ void repeatedDeadlineDoesNotExtendFinalWrite() throws Exception
+ {
+ createConnection(TransportEncryption.NONE);
+ fillOutput();
+ _connection.setCloseDeadline(new ConnectionClosingTicker(0,
_connection));
+ _connection.setCloseDeadline(new ConnectionClosingTicker(30_000,
_connection));
+ _connection.close();
+ clearInvocations(_socket);
+
+ assertTrue(_connection.doWork());
+
+ verify(_socket, never()).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void gracefulTransportCloseStillFlushesOutput(final TransportEncryption
encryption) throws Exception
+ {
+ createConnection(encryption);
+ fillOutput();
+ _writable.set(true);
+ clearInvocations(_socket);
+ _connection.close();
+
+ assertTrue(_connection.doWork());
+
+ verify(_socket).write(any(ByteBuffer[].class));
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void directTransportCloseIsBounded(final TransportEncryption encryption)
throws Exception
+ {
+ when(_port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(0L);
+ createConnection(encryption);
+ fillOutput();
+ _connection.close();
+
+ assertTrue(_connection.doWork(), "Direct native close must reach the
final-write timeout");
+
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void abortReleasesOutputEvenIfProtocolCleanupFails(final
TransportEncryption encryption) throws Exception
+ {
+ createConnection(encryption);
+ fillOutput();
+ doThrow(new IllegalStateException("Protocol cleanup
failed")).when(_engine).closed();
+ _connection.abort();
+
+ assertThrows(IllegalStateException.class, _connection::doWork);
+
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void expiredDeadlineInstalledDuringReadDiscardsOutput(final
TransportEncryption encryption) throws Exception
+ {
+ createConnection(encryption);
+ _writable.set(true);
+ receiveDuringRead(() ->
+ {
+ queueOutput();
+ final Ticker deadline = new ConnectionClosingTicker(0,
_connection);
+ _connection.setCloseDeadline(deadline);
+ _ticker.addTicker(deadline);
+ _connection.close();
+ });
+
+ final boolean closed = _connection.doWork();
+
+ verify(_engine).received(any(QpidByteBuffer.class));
+ assertNoIoAfterReadClose(closed);
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void deadlineExpiresDuringReadDiscardsOutput(final TransportEncryption
encryption) throws Exception
+ {
+ createConnection(encryption);
+ _writable.set(true);
+ final AtomicBoolean expired = new AtomicBoolean();
+ installDeadline(expired);
+ receiveDuringRead(() ->
+ {
+ queueOutput();
+ expired.set(true);
+ });
+
+ final boolean closed = _connection.doWork();
+
+ verify(_engine).received(any(QpidByteBuffer.class));
+ assertNoIoAfterReadClose(closed);
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void abortDuringReadDiscardsOutput(final TransportEncryption encryption)
throws Exception
+ {
+ createConnection(encryption);
+ _writable.set(true);
+ receiveDuringRead(() ->
+ {
+ queueOutput();
+ _connection.abort();
+ });
+
+ final boolean closed = _connection.doWork();
+
+ verify(_engine).received(any(QpidByteBuffer.class));
+ assertNoIoAfterReadClose(closed);
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void gracefulCloseDuringReadFlushesOutput(final TransportEncryption
encryption) throws Exception
+ {
+ createConnection(encryption);
+ _writable.set(true);
+ receiveDuringRead(() ->
+ {
+ queueOutput();
+ _connection.close();
+ });
+
+ assertTrue(_connection.doWork());
+
+ verify(_engine).received(any(QpidByteBuffer.class));
+ verify(_socket, atLeastOnce()).write(any(ByteBuffer[].class));
+ final InOrder shutdownOrder = inOrder(_engine, _socket);
+ shutdownOrder.verify(_engine).setIOThread(null);
+ shutdownOrder.verify(_socket,
atLeastOnce()).write(any(ByteBuffer[].class));
+ verify(_engine, never()).setTransportBlockedForWriting(anyBoolean());
+ assertFalse(_connection.wantsWrite(), "Graceful close must drain
queued output");
+ assertClosedAndReleased();
+ }
+
+ @ParameterizedTest
+ @EnumSource(TransportEncryption.class)
+ void unexpiredDeadlineInstalledDuringReadAllowsWrite(final
TransportEncryption encryption) throws Exception
+ {
+ createConnection(encryption);
+ _writable.set(true);
+ receiveDuringRead(() ->
+ {
+ queueOutput();
+ installDeadline(new AtomicBoolean());
+ });
+
+ assertFalse(_connection.doWork());
+
+ verify(_engine).received(any(QpidByteBuffer.class));
+ verify(_socket).write(any(ByteBuffer[].class));
+ verify(_socket, never()).close();
+ }
+
+ private void receiveDuringRead(final Runnable action) throws Exception
+ {
+ // Supply a TLS application-data record; the mocked unwrap below
produces one application byte.
+ final byte[] record = {23, 3, 3, 0, 1, 1};
+ when(_socket.read(any(ByteBuffer.class))).thenAnswer(invocation ->
+ {
+ final ByteBuffer target = invocation.getArgument(0);
+ target.put(record);
+ return record.length;
+ });
+ if (_sslEngine != null)
+ {
+ when(_sslEngine.unwrap(any(ByteBuffer.class),
any(ByteBuffer.class))).thenAnswer(invocation ->
+ {
+ final ByteBuffer source = invocation.getArgument(0);
+ final ByteBuffer target = invocation.getArgument(1);
+ final int consumed = source.remaining();
+ source.position(source.limit());
+ target.put((byte) 1);
+ return new SSLEngineResult(SSLEngineResult.Status.OK,
+ SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING,
consumed, 1);
+ });
+ }
+ doAnswer(invocation ->
+ {
+ final QpidByteBuffer input = invocation.getArgument(0);
+ input.position(input.limit());
+ action.run();
+ return null;
+ }).when(_engine).received(any(QpidByteBuffer.class));
+ }
+
+ private void assertNoIoAfterReadClose(final boolean closed)
+ {
+ assertAll(() -> assertTrue(closed, "Read-side closure must finish in
the same work cycle"),
+ () -> verify(_socket,
never()).write(any(ByteBuffer[].class)),
+ () ->
+ {
+ if (_sslEngine != null)
+ {
+ verify(_sslEngine,
never()).wrap(any(ByteBuffer[].class), any(ByteBuffer.class));
+ }
+ },
+ () -> verify(_engine,
never()).setTransportBlockedForWriting(anyBoolean()),
+ this::assertClosedAndReleased);
+ }
+
+ private void assertClosedAndReleased() throws IOException
+ {
+ verify(_socket).close();
+ verify(_engine).closed();
+ verify(_scheduler).removeConnection(_connection);
+ assertEquals(0, QpidByteBuffer.getNumberOfBuffersInUse(), "Shutdown
must release buffers");
+ }
+
+ private void createConnection(final TransportEncryption encryption) throws
Exception
+ {
+ if (encryption == TransportEncryption.TLS)
+ {
+ configureTls();
+ }
+ _connection = new NonBlockingConnection(_socket, _engine,
Set.of(encryption), () -> { }, _scheduler, _port);
+ _connection.setSelectionTask(_selectionTask);
+ }
+
+ private void configureTls() throws Exception
+ {
+ final SSLContext sslContext = mock(SSLContext.class);
+ _sslEngine = mock(SSLEngine.class);
+ final SSLSession session = mock(SSLSession.class);
+ when(_port.getSSLContext()).thenReturn(sslContext);
+ when(sslContext.createSSLEngine()).thenReturn(_sslEngine);
+ when(_sslEngine.getSession()).thenReturn(session);
+ when(_sslEngine.getEnabledProtocols()).thenReturn(new
String[]{"TLSv1.3"});
+ when(_sslEngine.getSupportedProtocols()).thenReturn(new
String[]{"TLSv1.3"});
+ when(_sslEngine.getEnabledCipherSuites()).thenReturn(new
String[]{"TLS_AES_128_GCM_SHA256"});
+ when(_sslEngine.getSupportedCipherSuites()).thenReturn(new
String[]{"TLS_AES_128_GCM_SHA256"});
+ when(session.getPacketBufferSize()).thenReturn(BUFFER_SIZE);
+ when(session.getApplicationBufferSize()).thenReturn(BUFFER_SIZE);
+ when(_port.getContextValue(Boolean.class,
AmqpPort.PORT_DIAGNOSIS_OF_SSL_ENGINE_LOOPING)).thenReturn(false);
+ when(_port.getContextValue(Integer.class,
AmqpPort.PORT_DIAGNOSIS_OF_SSL_ENGINE_LOOPING_WARN_THRESHOLD))
+ .thenReturn(0);
+ when(_port.getContextValue(Integer.class,
AmqpPort.PORT_DIAGNOSIS_OF_SSL_ENGINE_LOOPING_BREAK_THRESHOLD))
+ .thenReturn(0);
+
when(_sslEngine.getHandshakeStatus()).thenReturn(SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING);
+ when(_sslEngine.wrap(any(ByteBuffer[].class),
any(ByteBuffer.class))).thenAnswer(invocation ->
+ wrapTlsOutput(invocation.getArgument(0),
invocation.getArgument(1),
+ SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING));
+ }
+
+ private SSLEngineResult wrapTlsOutput(final ByteBuffer[] sources,
+ final ByteBuffer target,
+ final
SSLEngineResult.HandshakeStatus handshakeStatus)
+ {
+ int count = 0;
+ for (int i = 0; i < sources.length && target.hasRemaining(); i++)
+ {
+ while (sources[i].hasRemaining() && target.hasRemaining())
+ {
+ target.put(sources[i].get());
+ count++;
+ }
+ }
+ return new SSLEngineResult(SSLEngineResult.Status.OK, handshakeStatus,
count, count);
+ }
+
+ private void fillOutput()
+ {
+ queueOutput();
+ assertFalse(_connection.doWork());
+ assertTrue(_connection.wantsWrite() || _connection.wantsRead(),
+ "Blocked output must wait for socket capacity or TLS peer
input");
+ assertTrue(QpidByteBuffer.getNumberOfBuffersInUse() > 0);
+ }
+
+ private void queueOutput()
+ {
+ queueOutput(2 * BUFFER_SIZE);
+ }
+
+ private void queueOutput(final int size)
+ {
+ try (final QpidByteBuffer buffer =
QpidByteBuffer.allocateDirect(Math.max(BUFFER_SIZE, size)))
+ {
+ buffer.limit(size);
+ _connection.send(buffer);
+ }
+ }
+
+ private void installDeadline(final AtomicBoolean expired)
+ {
+ final Ticker deadline = mock(Ticker.class);
+ when(deadline.getTimeToNextTick(anyLong())).thenAnswer(invocation ->
expired.get() ? 0 : 1000);
+ _connection.setCloseDeadline(deadline);
+ _ticker.addTicker(deadline);
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionExceptionTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionExceptionTest.java
index 3532a38769..042c34409e 100644
---
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionExceptionTest.java
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionExceptionTest.java
@@ -21,17 +21,16 @@
package org.apache.qpid.server.transport;
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createPort;
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createSocketChannel;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import java.net.InetSocketAddress;
-import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Collections;
@@ -41,8 +40,6 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.server.logging.EventLogger;
-import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.transport.network.TransportEncryption;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
@@ -56,14 +53,8 @@ class NonBlockingConnectionExceptionTest extends UnitTestBase
@BeforeEach
void setUp()
{
- final Broker<?> broker = mock(Broker.class);
- when(broker.getEventLogger()).thenReturn(mock(EventLogger.class));
- _port = mock(AmqpPort.class);
- doReturn(broker).when(_port).getParent();
- when(_port.getNetworkBufferSize()).thenReturn(128);
+ _port = createPort(128, 100L);
when(_port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(100);
- when(_port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(100L);
- when(_port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
_scheduler = mock(NetworkConnectionScheduler.class);
}
@@ -72,8 +63,8 @@ class NonBlockingConnectionExceptionTest extends UnitTestBase
{
final ProtocolEngine failingEngine = mock(ProtocolEngine.class);
final ProtocolEngine healthyEngine = mock(ProtocolEngine.class);
- final SocketChannel failingChannel = mock(SocketChannel.class);
- final SocketChannel healthyChannel = mock(SocketChannel.class);
+ final SocketChannel failingChannel = createSocketChannel(() -> false);
+ final SocketChannel healthyChannel = createSocketChannel(() -> false);
final NonBlockingConnection failing = createConnection(failingEngine,
failingChannel);
final NonBlockingConnection healthy = createConnection(healthyEngine,
healthyChannel);
final ArithmeticException failure = new ArithmeticException("Injected
decoding failure");
@@ -101,10 +92,6 @@ class NonBlockingConnectionExceptionTest extends
UnitTestBase
private NonBlockingConnection createConnection(final ProtocolEngine
engine, final SocketChannel channel)
throws Exception
{
- final Socket socket = mock(Socket.class);
- when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 1000));
- when(socket.getLocalSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 5672));
- when(channel.socket()).thenReturn(socket);
when(channel.read(any(ByteBuffer.class))).thenAnswer(invocation ->
{
final ByteBuffer input = invocation.getArgument(0);
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionIdleTimeoutTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionIdleTimeoutTest.java
index 2d344a2ba5..13c259b03c 100644
---
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionIdleTimeoutTest.java
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionIdleTimeoutTest.java
@@ -19,13 +19,14 @@
package org.apache.qpid.server.transport;
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createPort;
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createSocketChannel;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -33,9 +34,6 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.lang.reflect.Field;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.SocketAddress;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collection;
@@ -50,8 +48,6 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.server.logging.EventLogger;
-import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.transport.network.TransportEncryption;
import org.apache.qpid.test.utils.UnitTestBase;
@@ -63,15 +59,9 @@ class NonBlockingConnectionIdleTimeoutTest extends
UnitTestBase
private NonBlockingConnection _connection;
@BeforeEach
- void beforeEach()
+ void beforeEach() throws Exception
{
- final SocketAddress localAddress = mock(SocketAddress.class);
- when(localAddress.toString()).thenReturn("127.0.0.1:5672");
- final Socket socket = mock(Socket.class);
- when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 1000));
- when(socket.getLocalSocketAddress()).thenReturn(localAddress);
- final SocketChannel socketChannel = mock(SocketChannel.class);
- when(socketChannel.socket()).thenReturn(socket);
+ final SocketChannel socketChannel = createSocketChannel(() -> false);
_ticker = mock(AggregateTicker.class);
_protocolEngine = mock(ProtocolEngine.class);
@@ -79,15 +69,8 @@ class NonBlockingConnectionIdleTimeoutTest extends
UnitTestBase
when(_protocolEngine.processPendingIterator()).thenAnswer(invocation
-> Collections.emptyIterator());
final NetworkConnectionScheduler scheduler =
mock(NetworkConnectionScheduler.class);
- final AmqpPort<?> port = mock(AmqpPort.class);
- when(port.getNetworkBufferSize()).thenReturn(1024);
+ final AmqpPort<?> port = createPort(1024, 100L);
when(port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(100);
- when(port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(100L);
- when(port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
- final EventLogger eventLogger = mock(EventLogger.class);
- final Broker<?> broker = mock(Broker.class);
- when(broker.getEventLogger()).thenReturn(eventLogger);
- doReturn(broker).when(port).getParent();
_connection = new NonBlockingConnection(socketChannel,
_protocolEngine, Set.of(TransportEncryption.NONE),
() -> { }, scheduler, port);
@@ -237,6 +220,10 @@ class NonBlockingConnectionIdleTimeoutTest extends
UnitTestBase
{
final Field delegateField =
NonBlockingConnection.class.getDeclaredField("_delegate");
delegateField.setAccessible(true);
+ final NonBlockingConnectionDelegate originalDelegate =
+ (NonBlockingConnectionDelegate) delegateField.get(_connection);
+ originalDelegate.shutdownInput();
+ originalDelegate.shutdownOutput();
delegateField.set(_connection, delegate);
}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTestHelper.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTestHelper.java
new file mode 100644
index 0000000000..4a704b9e9a
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTestHelper.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.qpid.server.transport;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.function.BooleanSupplier;
+
+import org.apache.qpid.server.logging.EventLogger;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.port.AmqpPort;
+
+final class NonBlockingConnectionTestHelper
+{
+ private NonBlockingConnectionTestHelper()
+ {
+ // utility class has private constructor
+ }
+
+ static AmqpPort<?> createPort(final int bufferSize, final long
finalWriteTimeout)
+ {
+ final Broker<?> broker = mock(Broker.class);
+ when(broker.getEventLogger()).thenReturn(new EventLogger());
+ return createPort(broker, bufferSize, finalWriteTimeout);
+ }
+
+ static AmqpPort<?> createPort(final Broker<?> broker, final int
bufferSize, final long finalWriteTimeout)
+ {
+ final AmqpPort<?> port = mock(AmqpPort.class);
+ doReturn(broker).when(port).getParent();
+ when(port.getNetworkBufferSize()).thenReturn(bufferSize);
+ when(port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(1);
+ when(port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(finalWriteTimeout);
+ when(port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
+ return port;
+ }
+
+ static SocketChannel createSocketChannel(final BooleanSupplier writable)
throws IOException
+ {
+ final Socket socket = mock(Socket.class);
+ when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 10000));
+ when(socket.getLocalSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 5672));
+ final SocketChannel channel = mock(SocketChannel.class);
+ when(channel.socket()).thenReturn(socket);
+ when(channel.write(any(ByteBuffer[].class))).thenAnswer(invocation ->
+ {
+ long written = 0;
+ if (writable.getAsBoolean())
+ {
+ for (final ByteBuffer buffer : (ByteBuffer[])
invocation.getArgument(0))
+ {
+ written += buffer.remaining();
+ buffer.position(buffer.limit());
+ }
+ }
+ return written;
+ });
+ return channel;
+ }
+}
diff --git
a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
index 9702be4759..ed37fe8e36 100644
---
a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
+++
b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
@@ -52,7 +52,6 @@ import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
-import org.apache.qpid.server.protocol.ConnectionClosingTicker;
import org.apache.qpid.server.protocol.ErrorCodes;
import org.apache.qpid.server.protocol.v0_10.transport.Binary;
import org.apache.qpid.server.protocol.v0_10.transport.ConnectionClose;
@@ -69,7 +68,6 @@ import
org.apache.qpid.server.protocol.v0_10.transport.SessionDetachCode;
import org.apache.qpid.server.protocol.v0_10.transport.SessionDetached;
import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.transport.AMQPConnection;
-import org.apache.qpid.server.transport.ServerNetworkConnection;
import org.apache.qpid.server.transport.network.NetworkConnection;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
@@ -164,14 +162,9 @@ public class ServerConnection extends ConnectionInvoker
lock.notifyAll();
}
- if(state == State.CLOSING)
+ if (state == State.CLOSING || state == State.CLOSE_RCVD)
{
- long timeoutTime = System.currentTimeMillis() +
getAmqpConnection().getContextValue(Long.class,
org.apache.qpid.server.model.Connection.CLOSE_RESPONSE_TIMEOUT);
-
- getAmqpConnection().getAggregateTicker().addTicker(new
ConnectionClosingTicker(timeoutTime, (ServerNetworkConnection)
getNetworkConnection()));
-
- // trigger a wakeup to ensure the ticker will be taken into account
- getAmqpConnection().notifyWork();
+ getAmqpConnection().startCloseDeadline();
}
}
@@ -290,6 +283,7 @@ public class ServerConnection extends ConnectionInvoker
void sendConnectionCloseAsync(final ConnectionCloseCode replyCode, final
String message)
{
+ getAmqpConnection().startCloseDeadline();
addAsyncTask(object ->
{
if(!isClosing())
@@ -649,6 +643,7 @@ public class ServerConnection extends ConnectionInvoker
protected void sendConnectionClose(ConnectionCloseCode replyCode, String
replyText, Option... _options)
{
+ getAmqpConnection().startCloseDeadline();
connectionClose(replyCode, replyText, _options);
}
diff --git
a/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/NativeCloseDeadline_0_10Test.java
b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/NativeCloseDeadline_0_10Test.java
new file mode 100644
index 0000000000..93c8aae86a
--- /dev/null
+++
b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/NativeCloseDeadline_0_10Test.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.qpid.server.protocol.v0_10;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.lang.reflect.Field;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import org.apache.qpid.server.model.Protocol;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.protocol.v0_10.transport.ConnectionClose;
+import org.apache.qpid.server.protocol.v0_10.transport.ConnectionCloseCode;
+import org.apache.qpid.server.protocol.v0_10.transport.ConnectionStartOk;
+import org.apache.qpid.server.protocol.v0_10.transport.ProtocolHeader;
+import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.server.transport.NativeCloseDeadlineTestBase;
+
+class NativeCloseDeadline_0_10Test extends NativeCloseDeadlineTestBase
+{
+ @ParameterizedTest
+ @NullAndEmptySource
+ @ValueSource(strings = "UNKNOWN")
+ void testAuthenticationRejectionHasCloseDeadline(final String mechanism)
throws Exception
+ {
+ setCloseTimeout(0);
+ final AMQPConnection_0_10Impl connection = new
AMQPConnection_0_10Impl(_broker, _network, _port, Transport.TCP,
+ 1L, _ticker);
+ installProtocol(connection);
+ final ServerConnection server = getServerConnection(connection);
+ final ServerConnectionDelegate delegate =
server.getConnectionDelegate();
+ delegate.init(server, new ProtocolHeader(1, 0, 10));
+
+ final ConnectionStartOk startOk = new ConnectionStartOk();
+ startOk.setMechanism(mechanism);
+ delegate.connectionStartOk(server, startOk);
+
+ assertTrue(_network.doWork(), "Authentication rejection must honor the
configured close deadline");
+ }
+
+ @Override
+ protected void receivePeerClose(final AMQPConnection<?> connection) throws
Exception
+ {
+ final ServerConnection server = getServerConnection(connection);
+ final ConnectionClose connectionClose = new
ConnectionClose(ConnectionCloseCode.NORMAL,
+ "Peer close reproduction");
+ server.getConnectionDelegate().connectionClose(server,
connectionClose);
+ }
+
+ @Override
+ protected AMQPConnection<?> createConnection(final Protocol protocol)
throws Exception
+ {
+ final AMQPConnection_0_10Impl connection = new
AMQPConnection_0_10Impl(_broker, _network, _port, Transport.TCP,
+ 1L, _ticker);
+ getServerConnection(connection).setState(ServerConnection.State.OPEN);
+ installProtocol(connection);
+ return connection;
+ }
+
+ @Override
+ protected Stream<Protocol> protocols()
+ {
+ return Stream.of(Protocol.AMQP_0_10);
+ }
+
+ private ServerConnection getServerConnection(final AMQPConnection<?>
connection) throws ReflectiveOperationException
+ {
+ final Field delegate =
AMQPConnection_0_10Impl.class.getDeclaredField("_connection");
+ delegate.setAccessible(true);
+ return (ServerConnection) delegate.get(connection);
+ }
+}
diff --git
a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
index f21a1a7abe..8708716a02 100644
---
a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
+++
b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
@@ -55,7 +55,6 @@ import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.plugin.ConnectionPropertyEnricher;
import org.apache.qpid.server.properties.ConnectionStartProperties;
-import org.apache.qpid.server.protocol.ConnectionClosingTicker;
import org.apache.qpid.server.protocol.ErrorCodes;
import org.apache.qpid.server.protocol.ProtocolVersion;
import org.apache.qpid.server.protocol.v0_8.transport.AMQDataBlock;
@@ -604,6 +603,7 @@ public class AMQPConnection_0_8Impl
private void sendConnectionClose(final int channelId, final AMQFrame frame)
{
+ startCloseDeadline();
if (_orderlyClose.compareAndSet(false, true))
{
try
@@ -613,19 +613,7 @@ public class AMQPConnection_0_8Impl
}
finally
{
- try
- {
- writeFrame(frame);
- }
- finally
- {
- final long timeoutTime = System.currentTimeMillis()
- + getContextValue(Long.class,
Connection.CLOSE_RESPONSE_TIMEOUT);
-
- getAggregateTicker().addTicker(new
ConnectionClosingTicker(timeoutTime, getNetwork()));
- // trigger a wakeup to ensure the ticker will be taken
into account
- notifyWork();
- }
+ writeFrame(frame);
}
}
}
@@ -866,6 +854,7 @@ public class AMQPConnection_0_8Impl
public void sendConnectionCloseAsync(final CloseReason reason, final
String description)
{
stopConnection();
+ startCloseDeadline();
final int cause;
switch(reason)
{
@@ -1102,6 +1091,7 @@ public class AMQPConnection_0_8Impl
try
{
+ startCloseDeadline();
if (_orderlyClose.compareAndSet(false, true))
{
completeAndCloseAllChannels();
diff --git
a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Test.java
b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Test.java
index db01465aab..1b3728edb1 100644
---
a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Test.java
+++
b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Test.java
@@ -608,6 +608,7 @@ class AMQPConnection_0_8Test extends UnitTestBase
@Test
void testIncompleteDiscardRetainsCloseTimeout()
{
+
when(_port.getContext()).thenReturn(Map.of(Connection.CLOSE_RESPONSE_TIMEOUT,
"0"));
final AMQPConnection_0_8Impl connection = new
AMQPConnection_0_8Impl(_broker, _network, _port, _transport,
_protocol, 0, _ticker);
openConnection(connection, ProtocolVersion.v0_8);
@@ -623,9 +624,8 @@ class AMQPConnection_0_8Test extends UnitTestBase
assertFalse(frame.hasRemaining());
verify(_network, never()).close();
- final long closeTimeout = connection.getContextValue(Long.class,
Connection.CLOSE_RESPONSE_TIMEOUT);
- _ticker.tick(System.currentTimeMillis() + closeTimeout + 1);
- verify(_network).close();
+ _ticker.tick(System.currentTimeMillis());
+ verify(_network).abort();
}
}
diff --git
a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/NativeCloseDeadline_0_8Test.java
b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/NativeCloseDeadline_0_8Test.java
new file mode 100644
index 0000000000..fb2d80a620
--- /dev/null
+++
b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/NativeCloseDeadline_0_8Test.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.qpid.server.protocol.v0_8;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.stream.Stream;
+
+import org.apache.qpid.server.model.Protocol;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.protocol.ProtocolVersion;
+import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.server.transport.NativeCloseDeadlineTestBase;
+
+class NativeCloseDeadline_0_8Test extends NativeCloseDeadlineTestBase
+{
+ @Override
+ protected void receivePeerClose(final AMQPConnection<?> connection)
+ {
+ ((AMQPConnection_0_8Impl) connection).receiveConnectionClose(200,
null, 0, 0);
+ }
+
+ @Override
+ protected AMQPConnection<?> createConnection(final Protocol protocol)
throws Exception
+ {
+ final AMQPConnection_0_8Impl connection = new
AMQPConnection_0_8Impl(_broker, _network, _port, Transport.TCP,
+ protocol, 1L, _ticker);
+ final ProtocolVersion version = switch (protocol)
+ {
+ case AMQP_0_8 -> ProtocolVersion.v0_8;
+ case AMQP_0_9 -> ProtocolVersion.v0_9;
+ default -> ProtocolVersion.v0_91;
+ };
+ final Method setVersion =
AMQPConnection_0_8Impl.class.getDeclaredMethod("setProtocolVersion",
+ ProtocolVersion.class);
+ setVersion.setAccessible(true);
+ setVersion.invoke(connection, version);
+ final Field state =
AMQPConnection_0_8Impl.class.getDeclaredField("_state");
+ state.setAccessible(true);
+ state.set(connection, AMQPConnection_0_8Impl.ConnectionState.OPEN);
+ installProtocol(connection);
+ return connection;
+ }
+
+ @Override
+ protected Stream<Protocol> protocols()
+ {
+ return Stream.of(Protocol.AMQP_0_8, Protocol.AMQP_0_9,
Protocol.AMQP_0_9_1);
+ }
+}
diff --git
a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
index 4b34581a53..880838e584 100644
---
a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
+++
b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
@@ -60,7 +60,6 @@ import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.plugin.ConnectionPropertyEnricher;
-import org.apache.qpid.server.protocol.ConnectionClosingTicker;
import
org.apache.qpid.server.protocol.v1_0.codec.DescribedTypeConstructorRegistry;
import org.apache.qpid.server.protocol.v1_0.codec.FrameWriter;
import org.apache.qpid.server.protocol.v1_0.codec.ProtocolHandler;
@@ -372,7 +371,7 @@ public class AMQPConnection_1_0Impl extends
AbstractAMQPConnection<AMQPConnectio
_saslComplete = true;
disposeSaslNegotiator();
_connectionState = ConnectionState.CLOSED;
- addCloseTicker();
+ startCloseDeadline();
}
private void disposeSaslNegotiator()
@@ -514,6 +513,7 @@ public class AMQPConnection_1_0Impl extends
AbstractAMQPConnection<AMQPConnectio
"Connection close sent before connection was opened");
break;
case OPENED:
+ startCloseDeadline();
_connectionState = ConnectionState.CLOSE_RECEIVED;
closeReceived();
if(close.getError() != null)
@@ -527,7 +527,6 @@ public class AMQPConnection_1_0Impl extends
AbstractAMQPConnection<AMQPConnectio
sendClose(new Close());
_connectionState = ConnectionState.CLOSED;
_orderlyClose.set(true);
- addCloseTicker();
break;
case CLOSE_SENT:
closeReceived();
@@ -969,14 +968,17 @@ public class AMQPConnection_1_0Impl extends
AbstractAMQPConnection<AMQPConnectio
error.setInfo(Map.of(Symbols.SOLE_CONNECTION_ENFORCEMENT, true));
final EventLogger logger = getEventLogger();
- final List<CompletableFuture<Void>> rescheduleFutures = new
ArrayList<>();
+ final List<CompletableFuture<?>> rescheduleFutures = new
ArrayList<>();
for (final AMQPConnection_1_0<?> connection :
e.getExistingConnections())
{
if (!connection.isClosing())
{
LOGGER.debug("Closing existing connection '{}': {}",
connection.getName(), e.getMessage());
- rescheduleFutures.add(connection.doOnIOThreadAsync(() ->
connection.close(error)));
+ connection.stopConnection();
+ connection.startCloseDeadline();
+
rescheduleFutures.add(CompletableFuture.anyOf(connection.getTransportClosedFuture(),
+ connection.doOnIOThreadAsync(() ->
connection.close(error))));
logger.message(ResourceLimitMessages.INFO(
String.format("Closing existing connection '%s'",
connection.getName()), e.getMessage()));
}
@@ -1158,12 +1160,10 @@ public class AMQPConnection_1_0Impl extends
AbstractAMQPConnection<AMQPConnectio
case OPENED:
sendClose(close);
_connectionState = ConnectionState.CLOSE_SENT;
- addCloseTicker();
break;
case CLOSE_RECEIVED:
sendClose(close);
_connectionState = ConnectionState.CLOSED;
- addCloseTicker();
break;
case CLOSE_SENT:
case CLOSED:
@@ -1458,16 +1458,6 @@ public class AMQPConnection_1_0Impl extends
AbstractAMQPConnection<AMQPConnectio
}
}
- private void addCloseTicker()
- {
- long timeoutTime = System.currentTimeMillis() +
getContextValue(Long.class, Connection.CLOSE_RESPONSE_TIMEOUT);
-
- getAggregateTicker().addTicker(new
ConnectionClosingTicker(timeoutTime, getNetwork()));
-
- // trigger a wakeup to ensure the ticker will be taken into account
- notifyWork();
- }
-
@Override
public boolean isTransportBlockedForWriting()
{
@@ -1544,8 +1534,8 @@ public class AMQPConnection_1_0Impl extends
AbstractAMQPConnection<AMQPConnectio
@Override
public void sendConnectionCloseAsync(final CloseReason reason, final
String description)
{
-
stopConnection();
+ startCloseDeadline();
final ErrorCondition cause;
switch(reason)
{
@@ -1753,6 +1743,7 @@ public class AMQPConnection_1_0Impl extends
AbstractAMQPConnection<AMQPConnectio
private void sendClose(Close closeToSend)
{
+ startCloseDeadline();
sendFrame(CONNECTION_CONTROL_CHANNEL, closeToSend);
closeSender();
}
diff --git
a/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/NativeCloseDeadline_1_0Test.java
b/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/NativeCloseDeadline_1_0Test.java
new file mode 100644
index 0000000000..5febff3d35
--- /dev/null
+++
b/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/protocol/v1_0/NativeCloseDeadline_1_0Test.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.qpid.server.protocol.v1_0;
+
+import java.lang.reflect.Field;
+import java.util.stream.Stream;
+
+import org.apache.qpid.server.model.Protocol;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Close;
+import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.server.transport.NativeCloseDeadlineTestBase;
+
+class NativeCloseDeadline_1_0Test extends NativeCloseDeadlineTestBase
+{
+ @Override
+ protected void receivePeerClose(final AMQPConnection<?> connection)
+ {
+ ((AMQPConnection_1_0Impl) connection).receiveClose(0, new Close());
+ }
+
+ @Override
+ protected AMQPConnection<?> createConnection(final Protocol protocol)
throws Exception
+ {
+ final AMQPConnection_1_0Impl connection = new
AMQPConnection_1_0Impl(_broker, _network, _port, Transport.TCP,
+ 1L, _ticker);
+ final Field state =
AMQPConnection_1_0Impl.class.getDeclaredField("_connectionState");
+ state.setAccessible(true);
+ state.set(connection, ConnectionState.OPENED);
+ installProtocol(connection);
+ return connection;
+ }
+
+ @Override
+ protected Stream<Protocol> protocols()
+ {
+ return Stream.of(Protocol.AMQP_1_0);
+ }
+}
diff --git
a/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/transport/AMQPConnection_1_0CloseExistingTest.java
b/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/transport/AMQPConnection_1_0CloseExistingTest.java
new file mode 100644
index 0000000000..fc6b32cc24
--- /dev/null
+++
b/broker-plugins/amqp-1-0-protocol/src/test/java/org/apache/qpid/server/transport/AMQPConnection_1_0CloseExistingTest.java
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.qpid.server.transport;
+
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createPort;
+import static
org.apache.qpid.server.transport.NonBlockingConnectionTestHelper.createSocketChannel;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import javax.security.auth.Subject;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.BrokerTestHelper;
+import org.apache.qpid.server.model.Connection;
+import org.apache.qpid.server.model.NamedAddressSpace;
+import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.Protocol;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.plugin.ProtocolEngineCreator;
+import org.apache.qpid.server.protocol.v1_0.AMQPConnection_1_0Impl;
+import org.apache.qpid.server.protocol.v1_0.ProtocolEngineCreator_1_0_0;
+import org.apache.qpid.server.protocol.v1_0.constants.Bytes;
+import org.apache.qpid.server.protocol.v1_0.constants.Symbols;
+import org.apache.qpid.server.protocol.v1_0.type.Symbol;
+import
org.apache.qpid.server.protocol.v1_0.type.extensions.soleconn.SoleConnectionEnforcementPolicy;
+import
org.apache.qpid.server.protocol.v1_0.type.extensions.soleconn.StrongConnectionEstablishmentLimiter;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Open;
+import org.apache.qpid.server.security.SubjectCreator;
+import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
+import
org.apache.qpid.server.security.auth.manager.AnonymousAuthenticationManager;
+import org.apache.qpid.server.security.limit.ConnectionSlot;
+import org.apache.qpid.server.session.AMQPSession;
+import org.apache.qpid.server.transport.network.TransportEncryption;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class AMQPConnection_1_0CloseExistingTest extends UnitTestBase
+{
+ private static final int BUFFER_SIZE = 1024;
+
+ private final List<ConnectionFixture> _connections = new ArrayList<>();
+ private Broker<?> _broker;
+ private AmqpPort<?> _port;
+
+ @BeforeEach
+ void setUp()
+ {
+ _broker = BrokerTestHelper.createNewBrokerMock();
+ when(_broker.getNetworkBufferSize()).thenReturn(BUFFER_SIZE);
+ when(_broker.getContextValue(Integer.class,
Connection.MAX_MESSAGE_SIZE))
+ .thenReturn(Connection.DEFAULT_MAX_MESSAGE_SIZE);
+ when(_broker.getContextValue(Boolean.class,
Broker.BROKER_MSG_AUTH)).thenReturn(false);
+ when(_broker.getContextValue(Integer.class,
Broker.MESSAGE_COMPRESSION_THRESHOLD_SIZE)).thenReturn(0);
+ _port = createPort(_broker, BUFFER_SIZE, 1000L);
+ when(_port.getContextValue(Integer.class, Connection.MAX_MESSAGE_SIZE))
+ .thenReturn(Connection.DEFAULT_MAX_MESSAGE_SIZE);
+ doReturn(_broker.getModel()).when(_port).getModel();
+ doReturn(_broker.getTaskExecutor()).when(_port).getTaskExecutor();
+ doReturn(_broker.getChildExecutor()).when(_port).getChildExecutor();
+
when(_port.getContext()).thenReturn(Map.of(Connection.CLOSE_RESPONSE_TIMEOUT,
"30000"));
+ when(_port.getContextValue(Long.class,
Port.CONNECTION_MAXIMUM_AUTHENTICATION_DELAY)).thenReturn(30000L);
+
+ final AnonymousAuthenticationManager authenticationProvider =
mock(AnonymousAuthenticationManager.class);
+
when(authenticationProvider.getMechanisms()).thenReturn(List.of(AnonymousAuthenticationManager.MECHANISM_NAME));
+ when(authenticationProvider.getAvailableMechanisms(false))
+
.thenReturn(List.of(AnonymousAuthenticationManager.MECHANISM_NAME));
+ when(authenticationProvider.getAnonymousPrincipal()).thenReturn(() ->
"anonymous");
+
doReturn(authenticationProvider).when(_port).getAuthenticationProvider();
+ final SubjectCreator subjectCreator = mock(SubjectCreator.class);
+ when(_port.getSubjectCreator(anyBoolean(),
isNull())).thenReturn(subjectCreator);
+ when(subjectCreator.createSubjectWithGroups(any())).thenReturn(new
Subject(true,
+ Set.of(new AuthenticatedPrincipal(() -> "anonymous")),
Set.of(), Set.of()));
+
+ final NamedAddressSpace addressSpace = mock(NamedAddressSpace.class);
+ when(addressSpace.getId()).thenReturn(UUID.randomUUID());
+ when(addressSpace.getPrincipal()).thenReturn(() -> "test");
+ when(addressSpace.isActive()).thenReturn(true);
+ when(addressSpace.authoriseCreateConnection(any())).thenReturn(true);
+ when(_port.getAddressSpace("test")).thenReturn(addressSpace);
+ final StrongConnectionEstablishmentLimiter limiter = new
StrongConnectionEstablishmentLimiter();
+ final Map<AMQPConnection<?>, ConnectionSlot> slots = new HashMap<>();
+ doAnswer(invocation ->
+ {
+ final AMQPConnection<?> connection = invocation.getArgument(0);
+ slots.put(connection, limiter.register(connection));
+ return null;
+ }).when(addressSpace).registerConnection(any());
+ doAnswer(invocation ->
+ {
+ final ConnectionSlot slot =
slots.remove(invocation.getArgument(0));
+ if (slot != null)
+ {
+ slot.free();
+ }
+ return null;
+ }).when(addressSpace).deregisterConnection(any());
+ }
+
+ @AfterEach
+ void tearDown()
+ {
+ for (final ConnectionFixture connection : _connections)
+ {
+ connection._network.abort();
+ connection._network.doWork();
+ }
+ _connections.clear();
+ }
+
+ @Test
+ void closeExistingDeadlineAbortsBlockedConnectionAndResumesOpen() throws
IOException
+ {
+
when(_port.getContext()).thenReturn(Map.of(Connection.CLOSE_RESPONSE_TIMEOUT,
"0"));
+ final ConnectionFixture existing = newConnection();
+ existing.open();
+ final CompletableFuture<Void> pending = existing.blockOutput();
+ final ConnectionFixture replacement = newConnection();
+
+ replacement.open();
+ assertFalse(replacement._connection.isOpen());
+ assertFalse(existing._connection.isClosing(), "The queued close has
not run");
+
+ assertTrue(existing._network.doWork(), "The policy must install a
deadline before queuing the close");
+ verify(existing._socket).close();
+ assertFalse(pending.isDone(), "Backpressure must prevent queued
protocol work from running");
+ assertFalse(replacement._network.doWork());
+ assertTrue(replacement._connection.isOpen(), "Transport closure must
release the waiting open");
+ }
+
+ @Test
+ void transportLossBeforeQueuedCloseResumesOpen() throws IOException
+ {
+ final ConnectionFixture existing = newConnection();
+ existing.open();
+ final CompletableFuture<Void> pending = existing.blockOutput();
+ final ConnectionFixture replacement = newConnection();
+ replacement.open();
+ assertFalse(replacement._connection.isOpen());
+ assertFalse(existing._network.doWork());
+ assertFalse(existing._connection.isClosing(), "Backpressure must
prevent the close task from running");
+
+ existing._network.abort();
+ assertTrue(existing._network.doWork());
+ assertFalse(pending.isDone());
+ assertFalse(replacement._network.doWork());
+ assertTrue(replacement._connection.isOpen(), "Transport loss must
complete the policy's waiting future");
+ }
+
+ @Test
+ void queuedCloseResumesOpenWithoutWaitingForTransportClosure() throws
IOException
+ {
+ final ConnectionFixture existing = newConnection();
+ existing.open();
+ final ConnectionFixture replacement = newConnection();
+ replacement.open();
+ assertFalse(replacement._connection.isOpen());
+
+ assertFalse(existing._network.doWork());
+ assertTrue(existing._connection.isClosing());
+ assertFalse(existing._connection.isClosed());
+ verify(existing._socket, never()).close();
+ assertFalse(replacement._network.doWork());
+ assertTrue(replacement._connection.isOpen());
+ }
+
+ @Test
+ void queuedCloseTakesPrecedenceOverPendingSessionWork() throws IOException
+ {
+ final ConnectionFixture existing = newConnection();
+ existing.open();
+ final AMQPSession<?, ?> session = mock(AMQPSession.class);
+ when(session.processPending()).thenReturn(true, false);
+ existing._connection.notifyWork(session);
+ final ConnectionFixture replacement = newConnection();
+
+ replacement.open();
+ assertFalse(replacement._connection.isOpen());
+ assertFalse(existing._network.doWork());
+
+ verify(session, never()).processPending();
+ assertTrue(existing._connection.isClosing(), "Pending session work
must not delay the queued close");
+ verify(existing._socket, never()).close();
+ assertFalse(replacement._network.doWork());
+ assertTrue(replacement._connection.isOpen(), "The replacement must
open while session work remains pending");
+ }
+
+ private ConnectionFixture newConnection() throws IOException
+ {
+ final ConnectionFixture fixture = new
ConnectionFixture(_connections.size());
+ _connections.add(fixture);
+ return fixture;
+ }
+
+ private final class ConnectionFixture
+ {
+ private final AtomicBoolean _writable = new AtomicBoolean(true);
+ private final SocketChannel _socket =
createSocketChannel(_writable::get);
+ private final NonBlockingConnection _network;
+ private AMQPConnection_1_0Impl _connection;
+
+ private ConnectionFixture(final long id) throws IOException
+ {
+ final ProtocolEngineCreator creator = new
ProtocolEngineCreator_1_0_0()
+ {
+ @Override
+ public ProtocolEngine newProtocolEngine(final Broker<?> broker,
+ final
ServerNetworkConnection networkConnection,
+ final AmqpPort<?> port,
+ final Transport
transport,
+ final long
connectionId,
+ final AggregateTicker
ticker)
+ {
+ _connection = (AMQPConnection_1_0Impl)
super.newProtocolEngine(broker, networkConnection, port,
+ transport, connectionId, ticker);
+ return _connection;
+ }
+ };
+ final MultiVersionProtocolEngine engine = new
MultiVersionProtocolEngine(_broker,
+ Set.of(Protocol.AMQP_1_0), Protocol.AMQP_1_0, _port,
Transport.TCP, id,
+ new ProtocolEngineCreator[]{creator}, null);
+ _network = new NonBlockingConnection(_socket, engine,
Set.of(TransportEncryption.NONE), () -> { },
+ mock(NetworkConnectionScheduler.class), _port);
+
_network.setSelectionTask(mock(SelectorThread.SelectionTask.class));
+ engine.setNetworkConnection(_network);
+ try (final QpidByteBuffer header =
QpidByteBuffer.wrap(Bytes.amqpHeader()))
+ {
+ engine.received(header);
+ }
+ }
+
+ private void open()
+ {
+ final Open open = new Open();
+ open.setContainerId("shared-container");
+ open.setHostname("test");
+ open.setDesiredCapabilities(new
Symbol[]{Symbols.SOLE_CONNECTION_FOR_CONTAINER});
+
open.setProperties(Map.of(Symbols.SOLE_CONNECTION_ENFORCEMENT_POLICY,
+
SoleConnectionEnforcementPolicy.CLOSE_EXISTING.getValue()));
+ _connection.receiveOpen(0, open);
+ }
+
+ private CompletableFuture<Void> blockOutput()
+ {
+ assertTrue(_connection.isOpen());
+ _writable.set(false);
+ try (final QpidByteBuffer output = QpidByteBuffer.wrap(new byte[2
* BUFFER_SIZE]))
+ {
+ _network.send(output);
+ }
+ final CompletableFuture<Void> pending =
_connection.doOnIOThreadAsync(() -> { });
+ assertFalse(_network.doWork());
+ assertTrue(_network.wantsWrite());
+ assertFalse(pending.isDone());
+ return pending;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
