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 fff3dbeba0 QPID-8772: [Broker-J] Excessive CPU usage from continuous
rescheduling of blocked TCP/TLS writes (#453)
fff3dbeba0 is described below
commit fff3dbeba0728ead7ce4f6dab62deff6d85d0bbb
Author: coolkaito <[email protected]>
AuthorDate: Mon Oct 26 10:22:24 2026 +0100
QPID-8772: [Broker-J] Excessive CPU usage from continuous rescheduling of
blocked TCP/TLS writes (#453)
---
.../server/transport/NonBlockingConnection.java | 20 +-
.../transport/NonBlockingConnectionDelegate.java | 6 +
.../NonBlockingConnectionTLSDelegate.java | 12 +-
.../NonBlockingConnectionSchedulingTest.java | 459 +++++++++++++++++++
.../NonBlockingConnectionTLSDelegateTest.java | 106 +++++
.../transport/NonBlockingConnectionTLSTest.java | 370 +++++++++++++++
.../transport/NonBlockingConnectionWriteTest.java | 508 +++++++++++++++++++++
.../qpid/server/transport/TlsTestContext.java | 54 +++
.../server/protocol/v0_10/ConsumerTarget_0_10.java | 21 +-
.../v0_10/ConsumerTarget_0_10FlushTest.java | 246 ++++++++++
.../server/protocol/v0_8/ConsumerTarget_0_8.java | 47 +-
.../qpid/server/protocol/v0_8/AMQChannelTest.java | 132 ++++++
12 files changed, 1959 insertions(+), 22 deletions(-)
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 343e2b3343..1d0cc188fe 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
@@ -225,12 +225,12 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
boolean wantsRead()
{
- return _fullyWritten;
+ return (_fullyWritten || _pendingIterator == null ||
_delegate.needsReadForWrite()) && _delegate.readyForRead();
}
boolean wantsWrite()
{
- return !_fullyWritten;
+ return !_fullyWritten && !_delegate.needsReadForWrite();
}
public boolean isStateChanged()
@@ -257,6 +257,7 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
public boolean doWork()
{
_protocolEngine.clearWork();
+ _partialRead = false;
if (!_closed.get())
{
try
@@ -278,19 +279,26 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
if (processPendingComplete)
{
_pendingIterator = null;
- _protocolEngine.setTransportBlockedForWriting(false);
+ }
+
+ if (processPendingComplete ||
_delegate.needsReadForWrite())
+ {
final boolean dataRead = doRead();
-
_protocolEngine.setTransportBlockedForWriting(!doWrite());
+ doWrite();
- if (!_fullyWritten || dataRead ||
(_delegate.needsWork() && _delegate.getNetInputBuffer().position() != 0))
+ if (_fullyWritten && (dataRead ||
(_delegate.needsWork() &&
+ _delegate.getNetInputBuffer().position() !=
0)))
{
_protocolEngine.notifyWork();
}
}
- else
+
+ if (!processPendingComplete && (_fullyWritten ||
getBufferedSize() < _port.getNetworkBufferSize()))
{
_protocolEngine.notifyWork();
}
+
+
_protocolEngine.setTransportBlockedForWriting(!_fullyWritten);
}
}
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionDelegate.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionDelegate.java
index f827768b05..873e7d0e37 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionDelegate.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionDelegate.java
@@ -54,6 +54,12 @@ interface NonBlockingConnectionDelegate
boolean readyForRead();
+ /** Whether pending output requires peer input before another write can
make progress. */
+ default boolean needsReadForWrite()
+ {
+ return false;
+ }
+
boolean processData() throws IOException;
Principal getPeerPrincipal();
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 6a4dab02f5..d860a8216c 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
@@ -59,6 +59,7 @@ public class NonBlockingConnectionTLSDelegate implements
NonBlockingConnectionDe
private Certificate _peerCertificate;
private boolean _principalChecked;
private volatile boolean _hostChecked;
+ private volatile boolean _needsReadForWrite;
private QpidByteBuffer _netInputBuffer;
private QpidByteBuffer _netOutputBuffer;
private QpidByteBuffer _applicationBuffer;
@@ -139,11 +140,11 @@ public class NonBlockingConnectionTLSDelegate implements
NonBlockingConnectionDe
}
tasksRun = runSSLEngineTasks(_status);
- _applicationBuffer.flip();
if(_applicationBuffer.position() > oldAppBufPos)
{
readData = true;
}
+ _applicationBuffer.flip();
_parent.processAmqpData(_applicationBuffer);
@@ -208,9 +209,18 @@ public class NonBlockingConnectionTLSDelegate implements
NonBlockingConnectionDe
}
}
}
+
+ _needsReadForWrite = _encryptedOutput.isEmpty() &&
+ _sslEngine.getHandshakeStatus() ==
SSLEngineResult.HandshakeStatus.NEED_UNWRAP;
return new WriteResult(bufsSent && _encryptedOutput.isEmpty(),
totalConsumed);
}
+ @Override
+ public boolean needsReadForWrite()
+ {
+ return _needsReadForWrite;
+ }
+
protected void restoreApplicationBufferForWrite()
{
try (QpidByteBuffer oldApplicationBuffer = _applicationBuffer)
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionSchedulingTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionSchedulingTest.java
new file mode 100644
index 0000000000..ecc396461e
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionSchedulingTest.java
@@ -0,0 +1,459 @@
+/*
+ * 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.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+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.ValueSource;
+
+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.Action;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class NonBlockingConnectionSchedulingTest extends UnitTestBase
+{
+ private static final int TIMEOUT_MILLIS = 10000;
+
+ private final List<Thread> _threads = new CopyOnWriteArrayList<>();
+ private final List<TestConnection> _connections = new ArrayList<>();
+ private NetworkConnectionScheduler _scheduler;
+ private SSLContext _sslContext;
+
+ @BeforeAll
+ void createTlsContext() throws Exception
+ {
+ _sslContext = TlsTestContext.create();
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {2, 4})
+ void blockedSocketReturnsWorkersAndRecoversPendingOutput(final int
workers) throws Exception
+ {
+ startScheduler(workers);
+ final TestConnection blocked = createConnection(1024);
+ final TestConnection independent = createConnection(1024);
+ final byte[] payload = new byte[2 * 1024 * 1024];
+ Arrays.fill(payload, (byte) 42);
+ final byte[] marker = new byte[]{1, 2, 3};
+ for (int i = 0; i < 3; i++)
+ {
+ blocked.submit(List.of(() -> blocked.send(payload), () ->
blocked.send(marker)));
+ blocked.awaitWriteInterest();
+
+ independent.submit(List.of(() -> independent.send(marker)));
+ assertArrayEquals(marker,
independent._client.getInputStream().readNBytes(marker.length),
+ "An independent connection must progress while the first
socket remains blocked");
+
+ assertArrayEquals(payload,
blocked._client.getInputStream().readNBytes(payload.length));
+ assertArrayEquals(marker,
blocked._client.getInputStream().readNBytes(marker.length),
+ "The suspended protocol iterator must resume after socket
capacity becomes available");
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"TLSv1.2, 2", "TLSv1.2, 4", "TLSv1.3, 2", "TLSv1.3, 4"})
+ void blockedTlsSocketReturnsWorkersAndRecoversPendingOutput(final String
protocol,
+ final int
workers) throws Exception
+ {
+ startScheduler(workers);
+ final TestConnection blocked = createTlsConnection(protocol);
+ final TestConnection independent = createTlsConnection(protocol);
+ final byte[] payload = new byte[2 * 1024 * 1024];
+ for (int i = 0; i < payload.length; i++)
+ {
+ payload[i] = (byte) i;
+ }
+ final byte[] marker = new byte[]{11, 22, 33};
+ for (int round = 0; round < 3; round++)
+ {
+ blocked.submit(List.of(() -> blocked.send(payload), () ->
blocked.send(marker)));
+ blocked.awaitWriteInterest();
+ assertTrue(blocked._blocked.get());
+
+ independent.submit(List.of(() -> independent.send(marker)));
+ assertArrayEquals(marker,
independent._client.getInputStream().readNBytes(marker.length),
+ "An independent TLS connection must progress while the
first socket is blocked");
+
+ blocked._client.getOutputStream().write(42 + round);
+ assertEquals(Integer.valueOf(42 + round),
blocked._received.poll(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS),
+ "TLS input must still be decrypted while encrypted output
is blocked");
+ blocked.awaitWriteInterest();
+ assertTrue(blocked._blocked.get(), "Reading peer input must not
briefly resume blocked consumers");
+
+ assertArrayEquals(payload,
blocked._client.getInputStream().readNBytes(payload.length));
+ assertArrayEquals(marker,
blocked._client.getInputStream().readNBytes(marker.length),
+ "Pending output must resume without losing or reordering
application bytes");
+ blocked.awaitSelection(false);
+ assertFalse(blocked._blocked.get(), "Draining encrypted output
must resume consumers");
+ }
+ }
+
+ @Test
+ void incomingHeartbeatsPreventReadIdleTimeoutWhileOutputIsBlocked() throws
Exception
+ {
+ startScheduler(2);
+ final TestConnection blocked = createConnection(4 * 1024 * 1024);
+ blocked.submit(List.of(() -> blocked.send(new byte[2 * 1024 * 1024])));
+ blocked.awaitWriteInterest();
+
+ final int readIdleMillis = 2000;
+ final AtomicLong currentTime = new AtomicLong();
+ final AtomicLong lastReadTime = new AtomicLong();
+ final BlockingQueue<Integer> heartbeats = new LinkedBlockingQueue<>();
+ final AtomicBoolean readerIdle = new AtomicBoolean();
+ when(blocked._engine.getLastReadTime()).thenAnswer(invocation ->
lastReadTime.get());
+ doAnswer(invocation ->
+ {
+ final QpidByteBuffer input = invocation.getArgument(0);
+ lastReadTime.set(currentTime.get());
+ while (input.hasRemaining())
+ {
+ heartbeats.add((int) input.get());
+ }
+ return null;
+ }).when(blocked._engine).received(any());
+ doAnswer(invocation ->
+ {
+ readerIdle.set(true);
+ blocked._connection.close();
+ return null;
+ }).when(blocked._engine).readerIdle();
+
+ // Keep wall-clock scheduling timestamps out of the idle calculation
so only the test advances time.
+ blocked._ticker.addTicker(new
ServerIdleReadTimeoutTicker(mock(ServerNetworkConnection.class),
+ blocked._engine, readIdleMillis)
+ {
+ @Override
+ public int getTimeToNextTick(final long ignored)
+ {
+ return super.getTimeToNextTick(currentTime.get());
+ }
+ });
+ blocked.awaitWriteInterest();
+
+ for (int i = 0; i < 8; i++)
+ {
+ currentTime.addAndGet(readIdleMillis / 4);
+ blocked._client.getOutputStream().write(42);
+ assertEquals(Integer.valueOf(42), heartbeats.poll(TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS),
+ "Incoming heartbeats must be consumed while the peer is
not reading output");
+ blocked.awaitWriteInterest();
+ assertFalse(readerIdle.get(), "A peer sending heartbeats must not
be considered idle");
+ }
+
+ assertTrue(blocked._connection.wantsWrite(), "Output must remain
blocked throughout the heartbeat checks");
+ currentTime.addAndGet(readIdleMillis - 1);
+ blocked.awaitWriteInterest();
+ assertFalse(readerIdle.get(), "The connection must remain open until
the read deadline");
+
+ currentTime.incrementAndGet();
+ blocked._connection.getSelectionTask().wakeup();
+ blocked._closed.get(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+ assertTrue(readerIdle.get(), "The read timeout must still close the
connection after heartbeats stop");
+ }
+
+ private void startScheduler(final int workers)
+ {
+ _scheduler = new NetworkConnectionScheduler(getTestName(), 1, workers,
1L, task ->
+ {
+ final Thread thread = new Thread(task, getTestName() + "-" +
_threads.size());
+ _threads.add(thread);
+ return thread;
+ });
+ _scheduler.start();
+ }
+
+ private TestConnection createConnection(final int networkBufferSize)
throws Exception
+ {
+ final TestConnection connection = new TestConnection(_scheduler,
networkBufferSize);
+ _connections.add(connection);
+ return connection;
+ }
+
+ private TestConnection createTlsConnection(final String protocol) throws
Exception
+ {
+ final SSLSocket client = (SSLSocket)
_sslContext.getSocketFactory().createSocket();
+ client.setEnabledProtocols(new String[]{protocol});
+ final TestConnection connection = new TestConnection(_scheduler, 64 *
1024, client, _sslContext);
+ _connections.add(connection);
+ client.startHandshake();
+ assertEquals(protocol, client.getSession().getProtocol());
+ return connection;
+ }
+
+ @AfterEach
+ void tearDown() throws Exception
+ {
+ if (_scheduler != null)
+ {
+ try
+ {
+ for (final TestConnection connection : _connections)
+ {
+ connection._connection.close();
+ }
+ for (final TestConnection connection : _connections)
+ {
+ connection._closed.get(TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS);
+ }
+ }
+ finally
+ {
+ for (final TestConnection connection : _connections)
+ {
+ connection._client.close();
+ connection._channel.close();
+ }
+ _scheduler.close();
+ try
+ {
+ for (final Thread thread : _threads)
+ {
+ thread.join(TIMEOUT_MILLIS);
+ assertFalse(thread.isAlive(), "Scheduler worker did
not terminate: " + thread.getName());
+ }
+ }
+ finally
+ {
+ for (final Thread thread : _threads)
+ {
+ thread.interrupt();
+ }
+ for (final Thread thread : _threads)
+ {
+ thread.join(TIMEOUT_MILLIS);
+ }
+ }
+ }
+ }
+ }
+
+ private static final class TestConnection
+ {
+ private final AtomicBoolean _hasWork = new AtomicBoolean();
+ private final AtomicBoolean _blocked = new AtomicBoolean();
+ private final AtomicReference<Iterator<Runnable>> _pending =
+ new AtomicReference<>(Collections.emptyIterator());
+ private final AtomicReference<SelectionInterest> _selectionInterest =
new AtomicReference<>();
+ private final BlockingQueue<Integer> _received = new
LinkedBlockingQueue<>();
+ private final AggregateTicker _ticker = new AggregateTicker()
+ {
+ @Override
+ public void resetModified()
+ {
+ super.resetModified();
+ signalSelection();
+ }
+ };
+ private final CompletableFuture<Void> _closed = new
CompletableFuture<>();
+ private final Socket _client;
+ private final SocketChannel _channel;
+ private final ProtocolEngine _engine = mock(ProtocolEngine.class);
+ private final NonBlockingConnection _connection;
+
+ private TestConnection(final NetworkConnectionScheduler scheduler,
final int networkBufferSize) throws Exception
+ {
+ this(scheduler, networkBufferSize, new Socket(), null);
+ }
+
+ private TestConnection(final NetworkConnectionScheduler scheduler,
+ final int networkBufferSize,
+ final Socket client,
+ final SSLContext sslContext) throws Exception
+ {
+ _client = client;
+ try (final ServerSocketChannel listener =
ServerSocketChannel.open())
+ {
+ listener.bind(new
InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
+ _client.setReceiveBufferSize(4096);
+ _client.setSoTimeout(TIMEOUT_MILLIS);
+ _client.connect(listener.getLocalAddress(), TIMEOUT_MILLIS);
+ _channel = listener.accept();
+ }
+ _channel.configureBlocking(false);
+ _channel.socket().setSendBufferSize(4096);
+ when(_engine.getAggregateTicker()).thenReturn(_ticker);
+ when(_engine.hasWork()).thenAnswer(invocation -> _hasWork.get());
+ when(_engine.processPendingIterator()).thenAnswer(invocation ->
_pending.get());
+ doAnswer(invocation ->
+ {
+ final QpidByteBuffer input = invocation.getArgument(0);
+ while (input.hasRemaining())
+ {
+ _received.add((int) input.get());
+ }
+ return null;
+ }).when(_engine).received(any());
+ doAnswer(invocation ->
+ {
+ _hasWork.set(false);
+ return null;
+ }).when(_engine).clearWork();
+ doAnswer(invocation ->
+ {
+ final boolean blocked = invocation.getArgument(0);
+ if (_blocked.getAndSet(blocked) && !blocked)
+ {
+ _engine.notifyWork();
+ }
+ return null;
+ }).when(_engine).setTransportBlockedForWriting(anyBoolean());
+ doAnswer(invocation ->
+ {
+ _closed.complete(null);
+ return null;
+ }).when(_engine).closed();
+ final AmqpPort<?> port = mock(AmqpPort.class);
+ when(port.getNetworkBufferSize()).thenReturn(networkBufferSize);
+ when(port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(1);
+ when(port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(0L);
+ when(port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
+ when(port.getSSLContext()).thenReturn(sslContext);
+ 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(1000);
+ when(port.getContextValue(Integer.class,
AmqpPort.PORT_DIAGNOSIS_OF_SSL_ENGINE_LOOPING_BREAK_THRESHOLD))
+ .thenReturn(10000);
+ final Broker<?> broker = mock(Broker.class);
+ when(broker.getEventLogger()).thenReturn(mock(EventLogger.class));
+ doReturn(broker).when(port).getParent();
+ final AtomicReference<Action<ProtocolEngine>> workListener = new
AtomicReference<>();
+ 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();
+ final TransportEncryption transportEncryption = sslContext == null
+ ? TransportEncryption.NONE
+ : TransportEncryption.TLS;
+ final Set<TransportEncryption> encryptionSet =
Set.of(transportEncryption);
+ _connection = new NonBlockingConnection(_channel, _engine,
encryptionSet, () -> { }, scheduler, port);
+ scheduler.addConnection(_connection);
+ }
+
+ private void submit(final List<Runnable> work)
+ {
+ _pending.set(work.iterator());
+ _engine.notifyWork();
+ }
+
+ private void send(final byte[] bytes)
+ {
+ try (final QpidByteBuffer buffer = QpidByteBuffer.wrap(bytes))
+ {
+ _connection.send(buffer);
+ }
+ }
+
+ private void awaitWriteInterest() throws Exception
+ {
+ awaitSelection(true);
+ }
+
+ private void awaitSelection(final boolean wantsWrite) throws Exception
+ {
+ final SelectionInterest interest = new
SelectionInterest(wantsWrite, new CompletableFuture<>());
+ _selectionInterest.set(interest);
+ _connection.getSelectionTask().wakeup();
+ try
+ {
+ interest.ready().get(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ fail("Socket never returned to readiness selection with
wantsWrite=" + wantsWrite, e);
+ }
+ finally
+ {
+ _selectionInterest.compareAndSet(interest, null);
+ }
+ }
+
+ private void signalSelection()
+ {
+ final SelectionInterest interest = _selectionInterest.get();
+ if (interest != null)
+ {
+ final SelectionKey key =
_channel.keyFor(_connection.getSelectionTask().getSelector());
+ final int expectedOps = (interest.wantsWrite() ?
SelectionKey.OP_WRITE : 0) |
+ (_connection.wantsRead() ? SelectionKey.OP_READ : 0);
+ if (_connection.wantsWrite() == interest.wantsWrite() &&
!_connection.isStateChanged() &&
+ !_connection.isPartialRead() && key != null &&
+ key.isValid() && key.interestOps() == expectedOps)
+ {
+ interest.ready().complete(null);
+ }
+ }
+ }
+
+ private record SelectionInterest(boolean wantsWrite,
CompletableFuture<Void> ready)
+ {
+ }
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegateTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegateTest.java
new file mode 100644
index 0000000000..2eb9bb5392
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegateTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+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.Test;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class NonBlockingConnectionTLSDelegateTest extends UnitTestBase
+{
+ @Test
+ void handshakeOutputIsFlushedBeforeWaitingForPeerInput() throws Exception
+ {
+ final NonBlockingConnection parent = mock(NonBlockingConnection.class);
+ final SSLEngine engine = mock(SSLEngine.class);
+ final AtomicReference<SSLEngineResult.HandshakeStatus> status =
+ new
AtomicReference<>(SSLEngineResult.HandshakeStatus.NEED_WRAP);
+ when(engine.getHandshakeStatus()).thenAnswer(invocation ->
status.get());
+ when(engine.wrap(any(ByteBuffer[].class),
any(ByteBuffer.class))).thenAnswer(invocation ->
+ {
+ final ByteBuffer output = invocation.getArgument(1);
+ output.put((byte) 1);
+ status.set(SSLEngineResult.HandshakeStatus.NEED_UNWRAP);
+ return new SSLEngineResult(SSLEngineResult.Status.OK,
status.get(), 0, 1);
+ });
+ final SSLSession session = mock(SSLSession.class);
+ when(session.getPacketBufferSize()).thenReturn(1024);
+ when(engine.getSession()).thenReturn(session);
+ when(engine.getEnabledProtocols()).thenReturn(new String[]{"TLSv1.3"});
+ when(engine.getSupportedProtocols()).thenReturn(new
String[]{"TLSv1.3"});
+ when(engine.getEnabledCipherSuites()).thenReturn(new
String[]{"TLS_AES_128_GCM_SHA256"});
+ when(engine.getSupportedCipherSuites()).thenReturn(new
String[]{"TLS_AES_128_GCM_SHA256"});
+ final SSLContext context = mock(SSLContext.class);
+ when(context.createSSLEngine()).thenReturn(engine);
+ final AmqpPort<?> port = mock(AmqpPort.class);
+ when(port.getSSLContext()).thenReturn(context);
+ when(port.getNetworkBufferSize()).thenReturn(4096);
+ 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(1000);
+ when(port.getContextValue(Integer.class,
AmqpPort.PORT_DIAGNOSIS_OF_SSL_ENGINE_LOOPING_BREAK_THRESHOLD))
+ .thenReturn(10000);
+ final NonBlockingConnectionTLSDelegate delegate = new
NonBlockingConnectionTLSDelegate(parent, port);
+ try (final QpidByteBuffer application = QpidByteBuffer.wrap(new
byte[128]))
+ {
+ final List<QpidByteBuffer> buffers = List.of(application);
+ assertFalse(delegate.doWrite(buffers).isComplete());
+ assertFalse(delegate.needsReadForWrite(), "Buffered handshake
output must still select for writing");
+
+ when(parent.writeToTransport(any())).thenAnswer(invocation ->
+ {
+ long written = 0;
+ final Collection<QpidByteBuffer> encrypted =
invocation.getArgument(0);
+ for (final QpidByteBuffer buffer : encrypted)
+ {
+ written += buffer.remaining();
+ buffer.position(buffer.limit());
+ }
+ return written;
+ });
+
+ assertFalse(delegate.doWrite(buffers).isComplete());
+ assertTrue(application.hasRemaining(), "Application bytes must
wait for the peer handshake response");
+ assertTrue(delegate.needsReadForWrite(), "A writable socket cannot
advance NEED_UNWRAP");
+ }
+ finally
+ {
+ delegate.shutdownInput();
+ delegate.shutdownOutput();
+ }
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSTest.java
new file mode 100644
index 0000000000..303bd7f7ae
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSTest.java
@@ -0,0 +1,370 @@
+/*
+ * 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.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+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;
+
+class NonBlockingConnectionTLSTest extends UnitTestBase
+{
+ private static final int NETWORK_BUFFER_SIZE = 64 * 1024;
+ private SSLContext _context;
+
+ @BeforeAll
+ void createTlsContext() throws Exception
+ {
+ _context = TlsTestContext.create();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"TLSv1.2", "TLSv1.3"})
+ void
handshakeWaitsForEncryptedOutputBeforePeerInputAndRestoresWriteInterest(final
String protocol) throws Exception
+ {
+ try (final TlsConnection tls = new TlsConnection(protocol))
+ {
+ final byte[] payload = new byte[]{1, 2, 3};
+ tls.send(payload);
+ tls._writeAllowance = 0;
+ tls.wrapPeer(ByteBuffer.allocate(0));
+ tls._connection.doWork();
+
+ assertTrue(tls._connection.wantsWrite(), "The blocked server
handshake must select for writing");
+ tls._connection.doWork();
+ assertFalse(tls._connection.isStateChanged(), "Blocked encrypted
output must not cause a busy loop");
+ assertFalse(tls._connection.isPartialRead());
+
+ tls._writeAllowance = 1;
+ tls._connection.doWork();
+ assertEquals(1, tls._toPeer.position(), "Only the explicitly
granted write capacity may be consumed");
+ assertTrue(tls._connection.wantsWrite(), "A partial handshake
write must retain write interest");
+
+ tls._writeAllowance = Integer.MAX_VALUE;
+ tls._connection.doWork();
+ assertEquals(SSLEngineResult.HandshakeStatus.NEED_UNWRAP,
tls._server.getHandshakeStatus());
+ assertTrue(tls._connection.wantsRead());
+ assertFalse(tls._connection.wantsWrite(), "Pending application
output must wait for the peer handshake");
+ assertFalse(tls._connection.isStateChanged());
+
+ tls.finishHandshake();
+ assertArrayEquals(payload, tls._peerReceived.toByteArray());
+ tls._peerReceived.reset();
+
+ tls._writeAllowance = 0;
+ tls.send(payload);
+ tls._connection.doWork();
+ assertTrue(tls._connection.wantsWrite(), "NEED_UNWRAP must not
remain cached after the handshake");
+ assertFalse(tls._connection.isStateChanged());
+
+ tls._writeAllowance = Integer.MAX_VALUE;
+ tls._connection.doWork();
+ tls.readAtPeer();
+ assertArrayEquals(payload, tls._peerReceived.toByteArray());
+ assertFalse(tls._connection.wantsWrite());
+ assertFalse(tls._blocked);
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"TLSv1.2, false", "TLSv1.2, true", "TLSv1.3, false", "TLSv1.3,
true"})
+ void applicationReadRequestsContinuationOnlyWhenOutputIsDrained(final
String protocol,
+ final
boolean blocked) throws Exception
+ {
+ try (final TlsConnection tls = new TlsConnection(protocol))
+ {
+ tls.finishHandshake();
+ final byte[] outbound = new byte[]{11, 12, 13};
+ final byte[] inbound = new byte[]{21, 22, 23};
+ tls._writeAllowance = blocked ? 0 : Integer.MAX_VALUE;
+ tls.send(outbound);
+ tls.wrapPeer(ByteBuffer.wrap(inbound));
+
+ tls._connection.doWork();
+
+ assertArrayEquals(inbound, tls._received.toByteArray(), "The real
TLS engine must decrypt peer input");
+ assertEquals(blocked, tls._connection.wantsWrite());
+ assertEquals(!blocked, tls._connection.isStateChanged(),
+ "Decrypted input must request a continuation only when
encrypted output has drained");
+
+ tls._connection.doWork();
+ assertFalse(tls._connection.isStateChanged(), "A pass without new
input must not keep rescheduling");
+ assertFalse(tls._connection.isPartialRead());
+ tls._writeAllowance = Integer.MAX_VALUE;
+ tls._connection.doWork();
+ tls.readAtPeer();
+ assertArrayEquals(outbound, tls._peerReceived.toByteArray());
+ assertFalse(tls._connection.wantsWrite());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"TLSv1.2", "TLSv1.3"})
+ void bufferedTlsInputRequestsContinuationAfterOutputDrains(final String
protocol) throws Exception
+ {
+ try (final TlsConnection tls = new TlsConnection(protocol))
+ {
+ tls.finishHandshake();
+ final byte[] inbound = new byte[]{31, 32, 33};
+ final byte[] outbound = new byte[]{41, 42, 43};
+ tls.wrapPeer(ByteBuffer.wrap(inbound));
+ tls._readAllowance = 5;
+ tls._writeAllowance = 0;
+ tls.send(outbound);
+ tls._connection.doWork();
+
+ assertEquals(0, tls._received.size());
+ assertTrue(tls._connection.wantsWrite());
+ assertFalse(tls._connection.isStateChanged(), "Buffered TLS input
must not spin behind a blocked write");
+ tls._connection.doWork();
+ assertFalse(tls._connection.isStateChanged());
+ assertFalse(tls._connection.isPartialRead());
+
+ tls._writeAllowance = Integer.MAX_VALUE;
+ tls._connection.doWork();
+ assertFalse(tls._connection.wantsWrite());
+ assertFalse(tls._connection.isPartialRead(), "No new socket bytes
were supplied during the drain");
+ assertTrue(tls._connection.isStateChanged(),
+ "Retained TLS input must request a continuation after
draining");
+
+ tls._readAllowance = Integer.MAX_VALUE;
+ tls._connection.doWork();
+ assertArrayEquals(inbound, tls._received.toByteArray());
+ tls._connection.doWork();
+ assertFalse(tls._connection.isStateChanged(), "Consuming the
buffered record must end the continuation");
+ tls.readAtPeer();
+ assertArrayEquals(outbound, tls._peerReceived.toByteArray());
+ }
+ }
+
+ /** Real TLS engines with explicitly granted socket capacity; all work
runs on the test thread */
+ private final class TlsConnection implements AutoCloseable
+ {
+ private final SSLEngine _server = _context.createSSLEngine();
+ private final SSLEngine _peer = _context.createSSLEngine();
+ private final ByteBuffer _fromPeer = ByteBuffer.allocate(4 *
NETWORK_BUFFER_SIZE);
+ private final ByteBuffer _toPeer = ByteBuffer.allocate(4 *
NETWORK_BUFFER_SIZE);
+ private final ByteArrayOutputStream _received = new
ByteArrayOutputStream();
+ private final ByteArrayOutputStream _peerReceived = new
ByteArrayOutputStream();
+ private final NonBlockingConnection _connection;
+ private int _writeAllowance = Integer.MAX_VALUE;
+ private int _readAllowance = Integer.MAX_VALUE;
+ private boolean _hasWork;
+ private boolean _blocked;
+
+ private TlsConnection(final String protocol) throws Exception
+ {
+ _peer.setUseClientMode(true);
+ _peer.setEnabledProtocols(new String[]{protocol});
+ _peer.beginHandshake();
+ final Socket socket = mock(Socket.class);
+ when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 1000));
+ when(socket.getLocalSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 5672));
+ final SocketChannel channel = mock(SocketChannel.class);
+ when(channel.socket()).thenReturn(socket);
+ when(channel.read(any(ByteBuffer.class))).thenAnswer(invocation ->
+ {
+ _fromPeer.flip();
+ final int read = transfer(_fromPeer,
invocation.getArgument(0), _readAllowance);
+ _fromPeer.compact();
+ _readAllowance -= read;
+ return read;
+ });
+ when(channel.write(any(ByteBuffer[].class))).thenAnswer(invocation
->
+ {
+ long written = 0;
+ for (final ByteBuffer buffer : (ByteBuffer[])
invocation.getArgument(0))
+ {
+ final int count = transfer(buffer, _toPeer,
_writeAllowance);
+ written += count;
+ _writeAllowance -= count;
+ }
+ return written;
+ });
+ final ProtocolEngine engine = mock(ProtocolEngine.class);
+ when(engine.getAggregateTicker()).thenReturn(new
AggregateTicker());
+ when(engine.processPendingIterator()).thenAnswer(invocation ->
Collections.emptyIterator());
+ when(engine.hasWork()).thenAnswer(invocation -> _hasWork);
+ doAnswer(invocation ->
+ {
+ _hasWork = false; return null;
+ }).when(engine).clearWork();
+ doAnswer(invocation ->
+ {
+ _hasWork = true; return null;
+ }).when(engine).notifyWork();
+ doAnswer(invocation ->
+ {
+ _blocked = invocation.getArgument(0); return null;
+ }).when(engine).setTransportBlockedForWriting(anyBoolean());
+ doAnswer(invocation ->
+ {
+ final QpidByteBuffer input = invocation.getArgument(0);
+ final byte[] bytes = new byte[input.remaining()];
+ input.get(bytes);
+ _received.writeBytes(bytes);
+ return null;
+ }).when(engine).received(any());
+ final SSLContext serverContext = mock(SSLContext.class);
+ when(serverContext.createSSLEngine()).thenReturn(_server);
+ final AmqpPort<?> port = mock(AmqpPort.class);
+ when(port.getNetworkBufferSize()).thenReturn(NETWORK_BUFFER_SIZE);
+ when(port.getSSLContext()).thenReturn(serverContext);
+ when(port.getTlsProtocolAllowList()).thenReturn(List.of(protocol));
+ 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(1000);
+ when(port.getContextValue(Integer.class,
AmqpPort.PORT_DIAGNOSIS_OF_SSL_ENGINE_LOOPING_BREAK_THRESHOLD))
+ .thenReturn(10000);
+ when(port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(1);
+ when(port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(0L);
+ when(port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
+ final Broker<?> broker = mock(Broker.class);
+ when(broker.getEventLogger()).thenReturn(mock(EventLogger.class));
+ doReturn(broker).when(port).getParent();
+ final Set<TransportEncryption> encryptionSet =
Set.of(TransportEncryption.TLS);
+ _connection = new NonBlockingConnection(channel, engine,
encryptionSet, () -> { }, mock(), port);
+
_connection.setSelectionTask(mock(SelectorThread.SelectionTask.class));
+ }
+
+ private void finishHandshake() throws Exception
+ {
+ for (int step = 0; step < 100; step++)
+ {
+ readAtPeer();
+ if (_peer.getHandshakeStatus() ==
SSLEngineResult.HandshakeStatus.NEED_WRAP)
+ {
+ wrapPeer(ByteBuffer.allocate(0));
+ }
+ _connection.doWork();
+ if (_peer.getHandshakeStatus() ==
SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING &&
+ _server.getHandshakeStatus() ==
SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING &&
+ _fromPeer.position() == 0 && _toPeer.position() == 0)
+ {
+ assertEquals(_peer.getSession().getProtocol(),
_server.getSession().getProtocol());
+ assertFalse(_connection.wantsWrite());
+ return;
+ }
+ }
+ fail("TLS handshake did not complete: peer=" +
_peer.getHandshakeStatus() + ", server=" +
+ _server.getHandshakeStatus());
+ }
+
+ private void wrapPeer(final ByteBuffer application) throws Exception
+ {
+ do
+ {
+ final SSLEngineResult result = _peer.wrap(application,
_fromPeer);
+ assertEquals(SSLEngineResult.Status.OK, result.getStatus());
+ runPeerTasks();
+ assertTrue(result.bytesConsumed() > 0 ||
result.bytesProduced() > 0,
+ "Peer wrap made no progress");
+ }
+ while (application.hasRemaining() || _peer.getHandshakeStatus() ==
SSLEngineResult.HandshakeStatus.NEED_WRAP);
+ }
+
+ private void readAtPeer() throws Exception
+ {
+ _toPeer.flip();
+ while (_toPeer.hasRemaining())
+ {
+ final ByteBuffer application =
ByteBuffer.allocate(NETWORK_BUFFER_SIZE);
+ final SSLEngineResult result = _peer.unwrap(_toPeer,
application);
+ runPeerTasks();
+ _peerReceived.write(application.array(), 0,
application.position());
+ if (result.getStatus() ==
SSLEngineResult.Status.BUFFER_UNDERFLOW ||
+ _peer.getHandshakeStatus() ==
SSLEngineResult.HandshakeStatus.NEED_WRAP)
+ {
+ break;
+ }
+ assertEquals(SSLEngineResult.Status.OK, result.getStatus());
+ assertTrue(result.bytesConsumed() > 0 ||
result.bytesProduced() > 0, "Peer unwrap made no progress");
+ }
+ _toPeer.compact();
+ }
+
+ private void runPeerTasks()
+ {
+ Runnable task;
+ while ((task = _peer.getDelegatedTask()) != null)
+ {
+ task.run();
+ }
+ }
+
+ private void send(final byte[] bytes)
+ {
+ try (final QpidByteBuffer buffer = QpidByteBuffer.wrap(bytes))
+ {
+ _connection.send(buffer);
+ }
+ }
+
+ @Override
+ public void close()
+ {
+ _writeAllowance = Integer.MAX_VALUE;
+ _connection.close();
+ _connection.doWork();
+ _peer.closeOutbound();
+ }
+ }
+
+ private static int transfer(final ByteBuffer source, final ByteBuffer
destination, final int allowance)
+ {
+ final int length = Math.min(allowance, Math.min(source.remaining(),
destination.remaining()));
+ final ByteBuffer part = source.duplicate();
+ part.limit(part.position() + length);
+ destination.put(part);
+ source.position(source.position() + length);
+ return length;
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionWriteTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionWriteTest.java
new file mode 100644
index 0000000000..e24e9c31dc
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionWriteTest.java
@@ -0,0 +1,508 @@
+/*
+ * 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.junit.jupiter.api.Assertions.assertEquals;
+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.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.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+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.ValueSource;
+
+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;
+
+class NonBlockingConnectionWriteTest extends UnitTestBase
+{
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private final AtomicBoolean _hasWork = new AtomicBoolean();
+ private final AtomicBoolean _blocked = new AtomicBoolean();
+ private final AtomicInteger _writableBytes = new AtomicInteger();
+ private final List<Boolean> _blockedTransitions = new ArrayList<>();
+ private ProtocolEngine _engine;
+ private SocketChannel _channel;
+ private NonBlockingConnection _connection;
+
+ @BeforeEach
+ void setUp() throws Exception
+ {
+ _hasWork.set(false);
+ _blocked.set(false);
+ _writableBytes.set(0);
+ _blockedTransitions.clear();
+ final Socket socket = mock(Socket.class);
+ when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 1000));
+ when(socket.getLocalSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 5672));
+ _channel = mock(SocketChannel.class);
+ when(_channel.socket()).thenReturn(socket);
+ when(_channel.write(any(ByteBuffer[].class))).thenAnswer(invocation ->
+ {
+ final ByteBuffer[] buffers = invocation.getArgument(0);
+ final int offset = 0;
+ final int length = buffers.length;
+ long written = 0;
+ for (int i = offset; i < offset + length; i++)
+ {
+ final int size = Math.min(buffers[i].remaining(),
_writableBytes.get());
+ buffers[i].position(buffers[i].position() + size);
+ _writableBytes.addAndGet(-size);
+ written += size;
+ }
+ return written;
+ });
+
+ _engine = mock(ProtocolEngine.class);
+ final AggregateTicker ticker = mock(AggregateTicker.class);
+
when(ticker.getTimeToNextTick(anyLong())).thenReturn(Integer.MAX_VALUE);
+ when(_engine.getAggregateTicker()).thenReturn(ticker);
+ when(_engine.processPendingIterator()).thenAnswer(invocation ->
Collections.emptyIterator());
+ when(_engine.hasWork()).thenAnswer(invocation -> _hasWork.get());
+ when(_engine.isTransportBlockedForWriting()).thenAnswer(invocation ->
_blocked.get());
+ doAnswer(invocation ->
+ {
+ _hasWork.set(false);
+ return null;
+ }).when(_engine).clearWork();
+ doAnswer(invocation ->
+ {
+ _hasWork.set(true);
+ return null;
+ }).when(_engine).notifyWork();
+ doAnswer(invocation ->
+ {
+ final boolean blocked = invocation.getArgument(0);
+ if (_blocked.getAndSet(blocked) != blocked)
+ {
+ _blockedTransitions.add(blocked);
+ if (!blocked)
+ {
+ _engine.notifyWork();
+ }
+ }
+ return null;
+ }).when(_engine).setTransportBlockedForWriting(anyBoolean());
+
+ final AmqpPort<?> port = mock(AmqpPort.class);
+ when(port.getNetworkBufferSize()).thenReturn(NETWORK_BUFFER_SIZE);
+ when(port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(1);
+ when(port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(0L);
+ when(port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
+ final Broker<?> broker = mock(Broker.class);
+ when(broker.getEventLogger()).thenReturn(mock(EventLogger.class));
+ doReturn(broker).when(port).getParent();
+ _connection = new NonBlockingConnection(_channel, _engine,
Set.of(TransportEncryption.NONE),
+ () -> { }, mock(NetworkConnectionScheduler.class), port);
+ _connection.setSelectionTask(mock(SelectorThread.SelectionTask.class));
+ }
+
+ @AfterEach
+ void tearDown()
+ {
+ _writableBytes.set(Integer.MAX_VALUE);
+ _connection.close();
+ _connection.doWork();
+ }
+
+ @Test
+ void blockedOutputWaitsForSocketReadiness()
+ {
+ send(128);
+
+ assertFalse(_connection.doWork());
+
+ assertTrue(_connection.wantsWrite());
+ assertTrue(_connection.wantsRead(),
+ "Completed protocol work must allow incoming traffic during a
blocked write");
+ assertFalse(_connection.isStateChanged(), "Pending output alone must
not reschedule the connection");
+ assertEquals(List.of(true), _blockedTransitions);
+
+ _engine.notifyWork();
+ _connection.doWork();
+ assertEquals(List.of(true), _blockedTransitions, "A failed retry must
not briefly resume consumers");
+ assertFalse(_connection.isStateChanged());
+
+ _writableBytes.set(32);
+ _connection.doWork();
+ assertTrue(_connection.wantsWrite(), "Partial progress must retain
write interest");
+ assertEquals(List.of(true), _blockedTransitions);
+ assertFalse(_connection.isStateChanged());
+
+ _writableBytes.set(96);
+ _connection.doWork();
+ assertEquals(List.of(true, false), _blockedTransitions);
+ assertTrue(_connection.isStateChanged(), "Unblocking must resume
pending protocol output");
+ assertTrue(_connection.wantsRead());
+ assertFalse(_connection.wantsWrite());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, NETWORK_BUFFER_SIZE / 4})
+ void pendingIteratorIsRescheduledWhenSecondWriteDrainsOutput(final int
firstWriteBytes) throws Exception
+ {
+ final AtomicInteger completed = new AtomicInteger();
+ final List<Runnable> tasks = List.of(() -> send(2 *
NETWORK_BUFFER_SIZE), completed::incrementAndGet);
+ when(_engine.processPendingIterator()).thenReturn(tasks.iterator());
+ final AtomicInteger writeCalls = new AtomicInteger();
+ doAnswer(invocation ->
+ {
+ final ByteBuffer[] buffers = invocation.getArgument(0);
+ int writableBytes = writeCalls.getAndIncrement() == 0 ?
firstWriteBytes : Integer.MAX_VALUE;
+ long written = 0;
+ for (final ByteBuffer buffer : buffers)
+ {
+ final int size = Math.min(buffer.remaining(), writableBytes);
+ buffer.position(buffer.position() + size);
+ writableBytes -= size;
+ written += size;
+ }
+ return written;
+ }).when(_channel).write(any(ByteBuffer[].class));
+
+ assertFalse(_connection.doWork());
+
+ assertEquals(2, writeCalls.get(), "The second write must drain output
after the iterator is suspended");
+ assertEquals(0, completed.get(), "The iterator still has runnable work
after the second write");
+ assertTrue(_connection.wantsRead());
+ assertFalse(_connection.wantsWrite());
+ assertFalse(_connection.isPartialRead());
+ assertFalse(_blocked.get());
+ assertTrue(_blockedTransitions.isEmpty(), "There is no backpressure
transition to supply a wake-up");
+ assertTrue(_connection.isStateChanged(),
+ "Pending protocol work must be rescheduled when output drains
in the same pass");
+
+ assertFalse(_connection.doWork());
+ assertEquals(1, completed.get(), "The retained iterator must resume
exactly once");
+ assertFalse(_connection.isStateChanged(), "An exhausted iterator must
not keep rescheduling");
+ verify(_engine).processPendingIterator();
+ }
+
+ @Test
+ void pendingIteratorResumesAfterOutputDrains()
+ {
+ final AtomicInteger completed = new AtomicInteger();
+ final List<Runnable> tasks = List.of(() -> send(NETWORK_BUFFER_SIZE),
completed::incrementAndGet);
+ when(_engine.processPendingIterator()).thenReturn(tasks.iterator());
+
+ _connection.doWork();
+
+ assertEquals(0, completed.get());
+ assertTrue(_connection.wantsWrite());
+ assertFalse(_connection.wantsRead(), "Suspended protocol work must
continue to apply read backpressure");
+ assertFalse(_connection.isStateChanged(), "A suspended iterator must
wait for its buffered output to drain");
+
+ _writableBytes.set(NETWORK_BUFFER_SIZE);
+ _connection.doWork();
+
+ assertEquals(1, completed.get(), "Suspended work must resume exactly
once");
+ assertFalse(_connection.wantsWrite());
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false, 768", "true, 768", "false, 1023", "true, 1023"})
+ void partialWriteBelowBufferLimitResumesWorkAndReads(final boolean
pendingTask,
+ final int
remainingOutput) throws Exception
+ {
+ final AtomicInteger completed = new AtomicInteger();
+ final AtomicInteger received = new AtomicInteger();
+ final List<Runnable> tasks = new ArrayList<>();
+ tasks.add(() -> send(remainingOutput + NETWORK_BUFFER_SIZE / 4));
+ if (pendingTask)
+ {
+ tasks.add(completed::incrementAndGet);
+ }
+ when(_engine.processPendingIterator()).thenReturn(tasks.iterator());
+ _writableBytes.set(NETWORK_BUFFER_SIZE / 4);
+ when(_channel.read(any(ByteBuffer.class))).thenAnswer(invocation ->
+ {
+ final ByteBuffer input = invocation.getArgument(0);
+ input.put((byte) 42);
+ return 1;
+ }).thenReturn(0);
+ doAnswer(invocation ->
+ {
+ final QpidByteBuffer input = invocation.getArgument(0);
+ assertEquals(42, input.get());
+ received.incrementAndGet();
+ return null;
+ }).when(_engine).received(any());
+
+ _connection.doWork();
+
+ assertEquals(0, completed.get());
+ assertEquals(0, received.get());
+ assertEquals(0, _writableBytes.get());
+ assertFalse(_connection.isPartialRead());
+ assertTrue(_connection.isStateChanged(), "Freed buffer space must
schedule a continuation");
+ assertTrue(_connection.wantsWrite(), "The partial write must leave
output pending");
+ assertEquals(List.of(true), _blockedTransitions);
+
+ _connection.doWork();
+
+ assertEquals(pendingTask ? 1 : 0, completed.get());
+ assertEquals(1, received.get(), "Incoming data must be read without
further write capacity");
+ assertTrue(_connection.wantsRead(), "Completing the iterator must
restore read interest");
+ assertTrue(_connection.wantsWrite());
+ assertTrue(_connection.isPartialRead());
+ assertFalse(_connection.isStateChanged());
+ assertEquals(List.of(true), _blockedTransitions, "Consumers must
remain blocked until output drains");
+
+ _connection.doWork();
+
+ assertEquals(pendingTask ? 1 : 0, completed.get(), "Pending work must
run exactly once");
+ assertEquals(1, received.get());
+ assertTrue(_connection.wantsRead());
+ assertTrue(_connection.wantsWrite());
+ assertFalse(_connection.isPartialRead());
+ assertFalse(_connection.isStateChanged(), "The connection must return
to selector waiting");
+ assertEquals(List.of(true), _blockedTransitions);
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false, 1024", "true, 1024", "false, 1025", "true, 1025"})
+ void partialWriteAtOrAboveBufferLimitWaitsForSocketReadiness(final boolean
pendingTask,
+ final int
remainingOutput) throws Exception
+ {
+ final AtomicInteger completed = new AtomicInteger();
+ final List<Runnable> tasks = new ArrayList<>();
+ tasks.add(() -> send(remainingOutput + NETWORK_BUFFER_SIZE / 4));
+ if (pendingTask)
+ {
+ tasks.add(completed::incrementAndGet);
+ }
+ when(_engine.processPendingIterator()).thenReturn(tasks.iterator());
+ _writableBytes.set(NETWORK_BUFFER_SIZE / 4);
+
+ _connection.doWork();
+
+ assertEquals(0, _writableBytes.get());
+ assertEquals(0, completed.get());
+ assertFalse(_connection.isStateChanged(), "A full buffer must wait for
socket write readiness");
+ assertFalse(_connection.isPartialRead());
+ assertFalse(_connection.wantsRead());
+ assertTrue(_connection.wantsWrite());
+
+ _engine.notifyWork();
+ _connection.doWork();
+
+ assertEquals(0, completed.get());
+ assertFalse(_connection.isStateChanged(), "Retrying with no write
progress must not cause a busy loop");
+ assertFalse(_connection.isPartialRead());
+ assertFalse(_connection.wantsRead());
+ assertTrue(_connection.wantsWrite());
+ assertEquals(List.of(true), _blockedTransitions);
+ verify(_channel, never()).read(any(ByteBuffer.class));
+ }
+
+ @Test
+ void resumedIteratorWaitsAgainWhenOutputReachesBufferLimit()
+ {
+ final AtomicInteger completed = new AtomicInteger();
+ final List<Runnable> tasks = List.of(() -> send(NETWORK_BUFFER_SIZE),
() ->
+ send(NETWORK_BUFFER_SIZE / 4), completed::incrementAndGet);
+ when(_engine.processPendingIterator()).thenReturn(tasks.iterator());
+ _writableBytes.set(NETWORK_BUFFER_SIZE / 4);
+
+ _connection.doWork();
+
+ assertTrue(_connection.isStateChanged(), "The partial write must allow
pending work to resume");
+
+ _connection.doWork();
+
+ assertEquals(0, completed.get(), "Refilling the buffer must suspend
the remaining task");
+ assertFalse(_connection.isStateChanged());
+ assertFalse(_connection.isPartialRead());
+ assertFalse(_connection.wantsRead());
+ assertTrue(_connection.wantsWrite());
+ assertEquals(List.of(true), _blockedTransitions);
+
+ _writableBytes.set(NETWORK_BUFFER_SIZE);
+ _connection.doWork();
+
+ assertEquals(1, completed.get(), "Write readiness must eventually
resume the remaining task");
+ assertFalse(_connection.wantsWrite());
+ assertEquals(List.of(true, false), _blockedTransitions);
+ }
+
+ @Test
+ void blockedPendingOutputClearsPreviousReadIndication() throws Exception
+ {
+ when(_channel.read(any(ByteBuffer.class))).thenReturn(1);
+ _connection.doWork();
+ assertTrue(_connection.isPartialRead());
+ final List<Runnable> tasks = List.of(() -> send(NETWORK_BUFFER_SIZE),
() -> { });
+ when(_engine.processPendingIterator()).thenReturn(tasks.iterator());
+
+ _connection.doWork();
+
+ assertTrue(_connection.wantsWrite());
+ assertFalse(_connection.isPartialRead(), "Skipping reads must not
reuse the preceding pass's read indication");
+ assertFalse(_connection.isStateChanged());
+ }
+
+ @Test
+ void explicitProtocolWorkRunsWhileOutputIsBlocked()
+ {
+ send(128);
+ _connection.doWork();
+ final AtomicInteger completed = new AtomicInteger();
+
when(_engine.processPendingIterator()).thenReturn(List.<Runnable>of(completed::incrementAndGet).iterator());
+ _engine.notifyWork();
+
+ _connection.doWork();
+
+ assertEquals(1, completed.get());
+ assertEquals(List.of(true), _blockedTransitions);
+ assertFalse(_connection.isStateChanged());
+ }
+
+ @Test
+ void closeDeadlineStillRunsWithBlockedOutput() throws Exception
+ {
+ send(NETWORK_BUFFER_SIZE);
+ _connection.doWork();
+ final AggregateTicker ticker = _engine.getAggregateTicker();
+ when(ticker.getTimeToNextTick(anyLong())).thenReturn(0);
+ when(ticker.tick(anyLong())).thenAnswer(invocation ->
+ {
+ _connection.close();
+ return Integer.MAX_VALUE;
+ });
+
+ assertTrue(_connection.doWork());
+
+ verify(_engine).closed();
+ verify(_channel).close();
+ }
+
+ @Test
+ void pendingTlsWrapDoesNotSelectForReading() throws Exception
+ {
+ final NonBlockingConnectionDelegate delegate =
mock(NonBlockingConnectionDelegate.class);
+ when(delegate.readyForRead()).thenReturn(false);
+ when(delegate.doWrite(any())).thenReturn(new
NonBlockingConnectionDelegate.WriteResult(false, 0));
+ final Field field =
NonBlockingConnection.class.getDeclaredField("_delegate");
+ field.setAccessible(true);
+ ((NonBlockingConnectionDelegate)
field.get(_connection)).shutdownInput();
+ field.set(_connection, delegate);
+
+ send(128);
+ _connection.doWork();
+
+ assertTrue(_connection.wantsWrite());
+ assertFalse(_connection.wantsRead(), "TLS NEED_WRAP cannot make
progress from read readiness");
+ assertFalse(_connection.isStateChanged(), "Pending TLS output must not
reschedule the connection");
+
+ when(delegate.readyForRead()).thenReturn(true);
+ assertTrue(_connection.wantsRead(), "Read interest must return when
TLS can process incoming data");
+ }
+
+ @Test
+ void pendingOutputCanWaitForTlsInput() throws Exception
+ {
+ final AtomicBoolean inputProcessed = new AtomicBoolean();
+ final AtomicInteger completed = new AtomicInteger();
+ final NonBlockingConnectionDelegate delegate =
mock(NonBlockingConnectionDelegate.class);
+ when(delegate.needsReadForWrite()).thenAnswer(invocation ->
!inputProcessed.get());
+ when(delegate.readyForRead()).thenReturn(true);
+ when(delegate.processData()).thenAnswer(invocation ->
+ {
+ inputProcessed.set(true);
+ return false;
+ });
+ when(delegate.doWrite(any())).thenAnswer(invocation ->
+ {
+ long consumed = 0;
+ if (inputProcessed.get())
+ {
+ final Collection<QpidByteBuffer> buffers =
invocation.getArgument(0);
+ for (final QpidByteBuffer buffer : buffers)
+ {
+ consumed += buffer.remaining();
+ buffer.position(buffer.limit());
+ }
+ }
+ return new
NonBlockingConnectionDelegate.WriteResult(inputProcessed.get(), consumed);
+ });
+ final Field field =
NonBlockingConnection.class.getDeclaredField("_delegate");
+ field.setAccessible(true);
+ ((NonBlockingConnectionDelegate)
field.get(_connection)).shutdownInput();
+ field.set(_connection, delegate);
+ try (final QpidByteBuffer input =
QpidByteBuffer.allocate(NETWORK_BUFFER_SIZE))
+ {
+ when(delegate.getNetInputBuffer()).thenReturn(input);
+ final List<Runnable> tasks = List.of(() ->
send(NETWORK_BUFFER_SIZE), completed::incrementAndGet);
+
when(_engine.processPendingIterator()).thenReturn(tasks.iterator());
+
+ _connection.doWork();
+
+ assertTrue(_connection.wantsRead(), "TLS must read the peer's
response with a suspended iterator");
+ assertFalse(_connection.wantsWrite(), "Writable socket readiness
cannot satisfy a TLS input wait");
+ assertFalse(_connection.isStateChanged());
+ assertEquals(0, completed.get());
+
+ when(_channel.read(any(ByteBuffer.class))).thenReturn(1, 0);
+ _connection.doWork();
+ assertTrue(inputProcessed.get());
+ _connection.doWork();
+
+ assertEquals(1, completed.get(), "The TLS input wait must not
strand pending protocol work");
+ assertFalse(_connection.wantsWrite());
+ assertEquals(List.of(true, false), _blockedTransitions);
+ }
+ }
+
+ private void send(final int size)
+ {
+ try (final QpidByteBuffer buffer = QpidByteBuffer.wrap(new byte[size]))
+ {
+ _connection.send(buffer);
+ }
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/TlsTestContext.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/TlsTestContext.java
new file mode 100644
index 0000000000..47d47ee729
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/TlsTestContext.java
@@ -0,0 +1,54 @@
+/*
+ * 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 java.security.KeyStore;
+import java.security.cert.Certificate;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+
+import org.apache.qpid.test.utils.tls.KeyCertificatePair;
+import org.apache.qpid.test.utils.tls.TlsResourceBuilder;
+
+final class TlsTestContext
+{
+ private TlsTestContext()
+ {
+ // utility class has private constructor
+ }
+
+ static SSLContext create() throws Exception
+ {
+ final KeyCertificatePair certificate =
TlsResourceBuilder.createSelfSigned("CN=localhost");
+ final char[] password = "test-password".toCharArray();
+ final KeyStore keys = KeyStore.getInstance("PKCS12");
+ keys.load(null, password);
+ keys.setKeyEntry("server", certificate.privateKey(), password, new
Certificate[]{certificate.certificate()});
+ final KeyManagerFactory keyManagers =
KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ keyManagers.init(keys, password);
+ final TrustManagerFactory trustManagers = TrustManagerFactory
+ .getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ trustManagers.init(keys);
+ final SSLContext context = SSLContext.getInstance("TLS");
+ context.init(keyManagers.getKeyManagers(),
trustManagers.getTrustManagers(), null);
+ return context;
+ }
+}
diff --git
a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
index eca3192e26..eaaab58a5c 100644
---
a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
+++
b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
@@ -75,6 +75,7 @@ public class ConsumerTarget_0_10 extends
AbstractConsumerTarget<ConsumerTarget_0
private volatile FlowCreditManager_0_10 _creditManager;
private int _deferredMessageCredit;
private long _deferredSizeCredit;
+ private boolean _flushing;
private final StateChangeListener<MessageInstance, EntryState>
_unacknowledgedMessageListener =
new StateChangeListener<>()
@@ -129,8 +130,8 @@ public class ConsumerTarget_0_10 extends
AbstractConsumerTarget<ConsumerTarget_0
{
final AMQPConnection_0_10 amqpConnection =
_session.getAMQPConnection();
- boolean state = !amqpConnection.isTransportBlockedForWriting()
- && getCreditManager().hasCredit();
+ final boolean state = (_flushing ||
!amqpConnection.isTransportBlockedForWriting()) &&
+ getCreditManager().hasCredit();
setNotifyWorkDesired(state);
}
@@ -564,9 +565,19 @@ public class ConsumerTarget_0_10 extends
AbstractConsumerTarget<ConsumerTarget_0
public void flush()
{
- flushCreditState(true);
- while(sendNextMessage());
- stop();
+ _flushing = true;
+ try
+ {
+ flushCreditState(true);
+ updateNotifyWorkDesired();
+ while (sendNextMessage());
+ stop();
+ }
+ finally
+ {
+ _flushing = false;
+ updateNotifyWorkDesired();
+ }
}
@Override
diff --git
a/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10FlushTest.java
b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10FlushTest.java
new file mode 100644
index 0000000000..2ed00ec2f9
--- /dev/null
+++
b/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10FlushTest.java
@@ -0,0 +1,246 @@
+/*
+ *
+ * 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.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+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.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+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.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+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.EnumSource;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.InOrder;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.consumer.ConsumerOption;
+import org.apache.qpid.server.model.BrokerTestHelper;
+import org.apache.qpid.server.model.Consumer;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.protocol.v0_10.transport.DeliveryProperties;
+import org.apache.qpid.server.protocol.v0_10.transport.Header;
+import org.apache.qpid.server.protocol.v0_10.transport.MessageAcceptMode;
+import org.apache.qpid.server.protocol.v0_10.transport.MessageAcquireMode;
+import org.apache.qpid.server.protocol.v0_10.transport.MessageCreditUnit;
+import org.apache.qpid.server.protocol.v0_10.transport.MessageFlowMode;
+import org.apache.qpid.server.protocol.v0_10.transport.MessageFlush;
+import org.apache.qpid.server.protocol.v0_10.transport.MessageProperties;
+import org.apache.qpid.server.protocol.v0_10.transport.MessageTransfer;
+import org.apache.qpid.server.protocol.v0_10.transport.Option;
+import org.apache.qpid.server.store.StoredMemoryMessage;
+import org.apache.qpid.server.transport.AggregateTicker;
+import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
+import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class ConsumerTarget_0_10FlushTest extends UnitTestBase
+{
+ private static final int MESSAGE_SIZE = 10;
+
+ private final List<MessageTransfer> _transfers = new ArrayList<>();
+
+ private QueueManagingVirtualHost<?> _host;
+ private Queue<?> _queue;
+ private AMQPConnection_0_10 _connection;
+ private ServerSession _session;
+ private ConsumerTarget_0_10 _target;
+
+ @BeforeEach
+ void setUp() throws Exception
+ {
+ _transfers.clear();
+ _target = null;
+ _host = BrokerTestHelper.createVirtualHost(getTestName(), this);
+ _queue = _host.createChild(Queue.class, Map.of(Queue.NAME, "queue"));
+ _connection = mock(AMQPConnection_0_10.class);
+ when(_connection.isIOThread()).thenReturn(true);
+ when(_connection.getAggregateTicker()).thenReturn(new
AggregateTicker());
+ when(_connection.getContextValue(Long.class,
Consumer.SUSPEND_NOTIFICATION_PERIOD)).thenReturn(0L);
+ doAnswer(invocation ->
+ {
+ ((Runnable) invocation.getArgument(0)).run();
+ return CompletableFuture.completedFuture(null);
+ }).when(_connection).doOnIOThreadAsync(any(Runnable.class));
+
+ _session = mock(ServerSession.class);
+ when(_session.getAMQPConnection()).thenReturn(_connection);
+ doReturn(_host).when(_session).getAddressSpace();
+ final Session_0_10 model = mock(Session_0_10.class);
+ when(model.getAMQPConnection()).thenReturn(_connection);
+ when(_session.getModelObject()).thenReturn(model);
+ final ServerConnection connection = mock(ServerConnection.class);
+
when(connection.getConnectionDelegate()).thenReturn(mock(ServerConnectionDelegate.class));
+ when(_session.getConnection()).thenReturn(connection);
+ doAnswer(invocation ->
+ {
+ _transfers.add(invocation.getArgument(0));
+ return null;
+ }).when(_session).sendMessage(any(MessageTransfer.class),
any(Runnable.class));
+ }
+
+ @AfterEach
+ void tearDown()
+ {
+ try
+ {
+ if (_target != null)
+ {
+ _target.close();
+ }
+ }
+ finally
+ {
+ _host.close();
+ }
+ }
+
+ @ParameterizedTest(name = "{0}, blocked={1}, queued={2}, bytes={3},
messages={4}, delivered={5}")
+ @MethodSource("flushCases")
+ void flushDeliversAvailableMessagesWithinCredit(final MessageFlowMode
flowMode,
+ final boolean
transportBlocked,
+ final int queued,
+ final long byteCredit,
+ final long messageCredit,
+ final int
expectedDeliveries) throws Exception
+ {
+
when(_connection.isTransportBlockedForWriting()).thenReturn(transportBlocked);
+ createTarget(flowMode, byteCredit, messageCredit);
+ enqueueMessages(queued);
+ final MessageFlush flush = new MessageFlush("consumer", Option.SYNC);
+
+ new ServerSessionDelegate().command(_session, flush);
+
+ assertEquals(expectedDeliveries, _transfers.size());
+ assertEquals(queued - expectedDeliveries,
_queue.getAvailableMessages());
+ assertFalse(_target.getCreditManager().hasCredit(), "Flush must clear
remaining credit");
+ assertTrue(_target.isSuspended());
+ final InOrder order = inOrder(_session);
+ if (expectedDeliveries > 0)
+ {
+ order.verify(_session,
times(expectedDeliveries)).sendMessage(any(MessageTransfer.class),
+ any(Runnable.class));
+ }
+ else
+ {
+ verify(_session, never()).sendMessage(any(MessageTransfer.class),
any(Runnable.class));
+ }
+ order.verify(_session).processed(flush);
+ order.verify(_session).flushProcessed();
+
+ _target.addCredit(MessageCreditUnit.MESSAGE, 10L);
+ _target.addCredit(MessageCreditUnit.BYTE,
FlowCreditManager_0_10.INFINITE_CREDIT);
+ assertEquals(transportBlocked, _target.isSuspended(),
+ "New credit must still respect transport backpressure after
the flush");
+ verify(_connection,
never()).setTransportBlockedForWriting(anyBoolean());
+ }
+
+ @ParameterizedTest
+ @EnumSource(MessageFlowMode.class)
+ void failedFlushRestoresTransportBackpressure(final MessageFlowMode
flowMode) throws Exception
+ {
+ when(_connection.isTransportBlockedForWriting()).thenReturn(true);
+ createTarget(flowMode, -1L, 2L);
+ enqueueMessages(1);
+ final ConnectionScopedRuntimeException failure = new
ConnectionScopedRuntimeException("Delivery failed");
+
doThrow(failure).when(_session).sendMessage(any(MessageTransfer.class),
any(Runnable.class));
+
+ assertSame(failure,
assertThrows(ConnectionScopedRuntimeException.class, _target::flush));
+
+ assertTrue(_target.getCreditManager().hasCredit());
+ _target.updateNotifyWorkDesired();
+ assertTrue(_target.isSuspended(), "A failed flush must not leave the
consumer ignoring backpressure");
+ when(_connection.isTransportBlockedForWriting()).thenReturn(false);
+ _target.transportStateChanged();
+ assertFalse(_target.isSuspended());
+ verify(_connection,
never()).setTransportBlockedForWriting(anyBoolean());
+ }
+
+ private static Stream<Arguments> flushCases()
+ {
+ final List<Arguments> cases = new ArrayList<>();
+ for (final MessageFlowMode flowMode : MessageFlowMode.values())
+ {
+ for (final boolean blocked : new boolean[]{false, true})
+ {
+ cases.add(Arguments.of(flowMode, blocked, 3, -1L, 2L, 2));
+ cases.add(Arguments.of(flowMode, blocked, 3, 2L *
MESSAGE_SIZE, 3L, 2));
+ cases.add(Arguments.of(flowMode, blocked, 1, -1L, 0L, 0));
+ cases.add(Arguments.of(flowMode, blocked, 1, 0L, 3L, 0));
+ cases.add(Arguments.of(flowMode, blocked, 0, -1L, 2L, 0));
+ cases.add(Arguments.of(flowMode, blocked, 1, -1L, 2L, 1));
+ }
+ }
+ return cases.stream();
+ }
+
+ private void createTarget(final MessageFlowMode flowMode,
+ final long byteCredit,
+ final long messageCredit) throws Exception
+ {
+ final FlowCreditManager_0_10 creditManager = flowMode ==
MessageFlowMode.CREDIT
+ ? new CreditCreditManager(byteCredit, messageCredit)
+ : new WindowCreditManager(byteCredit, messageCredit);
+ _target = new ConsumerTarget_0_10(_session, "consumer",
MessageAcceptMode.EXPLICIT,
+ MessageAcquireMode.PRE_ACQUIRED, flowMode, creditManager,
Map.of(), false);
+ when(_session.getSubscription("consumer")).thenReturn(_target);
+ _queue.addConsumer(_target, null, MessageTransferMessage.class,
"consumer",
+ EnumSet.of(ConsumerOption.ACQUIRES), null);
+ _target.updateNotifyWorkDesired();
+ }
+
+ private void enqueueMessages(final int count)
+ {
+ for (int i = 0; i < count; i++)
+ {
+ final Header header = new Header(new DeliveryProperties(), new
MessageProperties());
+ final long timeMillis = System.currentTimeMillis();
+ final MessageMetaData_0_10 metaData = new
MessageMetaData_0_10(header, MESSAGE_SIZE, timeMillis);
+ final StoredMemoryMessage<MessageMetaData_0_10> stored = new
StoredMemoryMessage<>(i + 1L, metaData);
+ try (final QpidByteBuffer content = QpidByteBuffer.wrap(new
byte[MESSAGE_SIZE]))
+ {
+ stored.addContent(content);
+ }
+ _queue.enqueue(new MessageTransferMessage(stored, null), null,
null);
+ }
+ }
+}
diff --git
a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
index b86a1c74e0..feb8346708 100644
---
a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
+++
b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
@@ -114,7 +114,18 @@ public abstract class ConsumerTarget_0_8 extends
AbstractConsumerTarget<Consumer
ClientDeliveryMethod deliveryMethod,
boolean multiQueue)
{
- super(channel, consumerTag, filters, creditManager,
deliveryMethod, multiQueue);
+ this(channel, consumerTag, filters, creditManager, deliveryMethod,
multiQueue, false);
+ }
+
+ private NoAckConsumer(final AMQChannel channel,
+ final AMQShortString consumerTag,
+ final FieldTable filters,
+ final FlowCreditManager_0_8 creditManager,
+ final ClientDeliveryMethod deliveryMethod,
+ final boolean multiQueue,
+ final boolean isGet)
+ {
+ super(channel, consumerTag, filters, creditManager,
deliveryMethod, multiQueue, isGet);
_txn = new
AutoCommitTransaction(channel.getAddressSpace().getMessageStore());
}
@@ -172,7 +183,7 @@ public abstract class ConsumerTarget_0_8 extends
AbstractConsumerTarget<Consumer
FlowCreditManager_0_8 creditManager,
ClientDeliveryMethod deliveryMethod)
{
- super(channel, consumerTag, filters, creditManager,
deliveryMethod, false);
+ super(channel, consumerTag, filters, creditManager,
deliveryMethod, false, true);
}
}
@@ -188,7 +199,7 @@ public abstract class ConsumerTarget_0_8 extends
AbstractConsumerTarget<Consumer
consumerTag,
filters, creditManager,
channel.getClientDeliveryMethod(),
- multiQueue, true);
+ multiQueue, true, false);
}
@@ -197,7 +208,7 @@ public abstract class ConsumerTarget_0_8 extends
AbstractConsumerTarget<Consumer
FlowCreditManager_0_8
creditManager,
ClientDeliveryMethod
deliveryMethod)
{
- return new AckConsumer(channel, consumerTag, filters, creditManager,
deliveryMethod, false, false);
+ return new AckConsumer(channel, consumerTag, filters, creditManager,
deliveryMethod, false, false, true);
}
static final class AckConsumer extends ConsumerTarget_0_8
@@ -209,9 +220,10 @@ public abstract class ConsumerTarget_0_8 extends
AbstractConsumerTarget<Consumer
FlowCreditManager_0_8 creditManager,
ClientDeliveryMethod deliveryMethod,
boolean multiQueue,
- final boolean usesCredit)
+ final boolean usesCredit,
+ final boolean isGet)
{
- super(channel, consumerTag, filters, creditManager,
deliveryMethod, multiQueue);
+ super(channel, consumerTag, filters, creditManager,
deliveryMethod, multiQueue, isGet);
_usesCredit = usesCredit;
}
@@ -243,12 +255,25 @@ public abstract class ConsumerTarget_0_8 extends
AbstractConsumerTarget<Consumer
private final Boolean _autoClose;
+ private final boolean _isGet;
+
public ConsumerTarget_0_8(AMQChannel channel,
AMQShortString consumerTag,
FieldTable arguments,
FlowCreditManager_0_8 creditManager,
ClientDeliveryMethod deliveryMethod,
boolean multiQueue)
+ {
+ this(channel, consumerTag, arguments, creditManager, deliveryMethod,
multiQueue, false);
+ }
+
+ private ConsumerTarget_0_8(final AMQChannel channel,
+ final AMQShortString consumerTag,
+ final FieldTable arguments,
+ final FlowCreditManager_0_8 creditManager,
+ final ClientDeliveryMethod deliveryMethod,
+ final boolean multiQueue,
+ final boolean isGet)
{
super(multiQueue, channel.getAMQPConnection());
@@ -258,6 +283,7 @@ public abstract class ConsumerTarget_0_8 extends
AbstractConsumerTarget<Consumer
_creditManager = creditManager;
_deliveryMethod = deliveryMethod;
+ _isGet = isGet;
if (arguments != null)
{
@@ -313,10 +339,11 @@ public abstract class ConsumerTarget_0_8 extends
AbstractConsumerTarget<Consumer
{
final AMQPConnection_0_8 amqpConnection = (AMQPConnection_0_8)
_channel.getAMQPConnection();
- boolean state = _channel.isChannelFlow()
- && !amqpConnection.isTransportBlockedForWriting()
- // this last condition does not need to exist for
browsers and no ack
- && getCreditManager().hasCredit();
+ boolean state = _channel.isChannelFlow() &&
+ // Basic.Get must report queue availability even when its
response will be buffered
+ (_isGet || !amqpConnection.isTransportBlockedForWriting()) &&
+ // this last condition does not need to exist for browsers and
no ack
+ getCreditManager().hasCredit();
setNotifyWorkDesired(state);
}
diff --git
a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
index 608c55d52c..4305b16142 100644
---
a/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
+++
b/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
@@ -20,7 +20,9 @@
*/
package org.apache.qpid.server.protocol.v0_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -43,13 +45,17 @@ import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import java.security.Principal;
+import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicReference;
import javax.security.auth.Subject;
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.mockito.ArgumentCaptor;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
@@ -63,14 +69,21 @@ import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.BrokerModel;
+import org.apache.qpid.server.model.BrokerTestHelper;
import org.apache.qpid.server.model.Connection;
+import org.apache.qpid.server.model.Consumer;
import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.Session;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.protocol.ErrorCodes;
import org.apache.qpid.server.protocol.ProtocolVersion;
import org.apache.qpid.server.protocol.converter.MessageConversionException;
+import org.apache.qpid.server.protocol.v0_8.transport.AMQFrame;
import
org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicGetEmptyBody;
+import org.apache.qpid.server.protocol.v0_8.transport.ContentHeaderBody;
+import org.apache.qpid.server.protocol.v0_8.transport.MessagePublishInfo;
import org.apache.qpid.server.protocol.v0_8.transport.MethodRegistry;
import org.apache.qpid.server.security.AccessDeniedException;
import org.apache.qpid.server.security.SubjectExecutionContext;
@@ -82,6 +95,7 @@ import org.apache.qpid.server.store.NullMessageStore;
import org.apache.qpid.server.store.StorableMessageMetaData;
import org.apache.qpid.server.store.StoredMemoryMessage;
import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.server.transport.AggregateTicker;
import org.apache.qpid.server.txn.LocalTransaction;
import org.apache.qpid.server.util.GZIPUtils.GZIPInflationLimitException;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
@@ -154,6 +168,76 @@ class AMQChannelTest extends UnitTestBase
_messageDestination = mock(MessageDestination.class);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void basicGetDeliversAvailableMessageWhileTransportBlocked(final boolean
noAck) throws Exception
+ {
+ try (final ConsumerTestContext context = new ConsumerTestContext())
+ {
+ final AMQMessage message = context.enqueueMessage();
+
when(_amqConnection.isTransportBlockedForWriting()).thenReturn(true);
+
+ context._channel.receiveBasicGet(context._queueName, noAck);
+
+ verify(_amqConnection.getProtocolOutputConverter())
+ .writeGetOk(eq(message), any(InstanceProperties.class),
eq(1), eq(1L), anyInt());
+ verify(_amqConnection, never()).writeFrame(any(AMQFrame.class));
+ verify(_amqConnection,
never()).setTransportBlockedForWriting(anyBoolean());
+ assertEquals(0, context._queue.getAvailableMessages());
+ assertEquals(noAck ? 0 : 1,
context._channel.getUnacknowledgedMessageCount());
+ if (!noAck)
+ {
+ context._channel.receiveBasicAck(1L, false);
+ }
+ assertEquals(0, context._queue.getQueueDepthMessages());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void basicGetReturnsEmptyForEmptyQueueWhileTransportBlocked(final boolean
noAck) throws Exception
+ {
+ try (final ConsumerTestContext context = new ConsumerTestContext())
+ {
+
when(_amqConnection.isTransportBlockedForWriting()).thenReturn(true);
+
+ context._channel.receiveBasicGet(context._queueName, noAck);
+
+ final ArgumentCaptor<AMQFrame> frame =
ArgumentCaptor.forClass(AMQFrame.class);
+ verify(_amqConnection).writeFrame(frame.capture());
+ assertInstanceOf(BasicGetEmptyBody.class,
frame.getValue().getBodyFrame());
+ verifyNoInteractions(_amqConnection.getProtocolOutputConverter());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void subscriptionWaitsForTransportToUnblock(final boolean noAck) throws
Exception
+ {
+ final ClientDeliveryMethod deliveryMethod =
mock(ClientDeliveryMethod.class);
+
when(_amqConnection.createDeliveryMethod(1)).thenReturn(deliveryMethod);
+ try (final ConsumerTestContext context = new ConsumerTestContext())
+ {
+ final AMQMessage message = context.enqueueMessage();
+
when(_amqConnection.isTransportBlockedForWriting()).thenReturn(true);
+ context._channel.receiveBasicConsume(context._queueName,
AMQShortString.valueOf("consumer"),
+ false, noAck, false, true, null);
+
+ assertFalse(context._channel.processPending());
+ assertEquals(0, context._queue.getConsumerCountWithCredit());
+ assertEquals(1, context._queue.getAvailableMessages());
+ verifyNoInteractions(deliveryMethod);
+
+
when(_amqConnection.isTransportBlockedForWriting()).thenReturn(false);
+ context._channel.transportStateChanged();
+ context._channel.processPending();
+
+
verify(deliveryMethod).deliverToClient(any(ConsumerTarget_0_8.class),
eq(message),
+ any(InstanceProperties.class), eq(1L));
+ assertEquals(0, context._queue.getAvailableMessages());
+ }
+ }
+
@Test
void receiveExchangeDeleteWhenIfUsedIsSetAndExchangeHasBindings()
{
@@ -535,4 +619,52 @@ class AMQChannelTest extends UnitTestBase
verify(_amqConnection).closeSessionAsync(channel,
AMQPConnection.CloseReason.RESOURCE_LIMIT,
failure.getMessage());
}
+
+ private final class ConsumerTestContext implements AutoCloseable
+ {
+ private final QueueManagingVirtualHost<?> _host;
+ private final Queue<?> _queue;
+ private final AMQShortString _queueName =
AMQShortString.valueOf("queue");
+ private final AMQChannel _channel;
+
+ private ConsumerTestContext() throws Exception
+ {
+ _host = BrokerTestHelper.createVirtualHost(getTestName(),
AMQChannelTest.this);
+ _queue = _host.createChild(Queue.class, Map.of(Queue.NAME,
_queueName.toString()));
+
when(_virtualHost.getAttainedMessageSource(_queueName.toString())).thenReturn(_queue);
+
when(_virtualHost.getMessageStore()).thenReturn(_host.getMessageStore());
+ when(_amqConnection.isIOThread()).thenReturn(true);
+ when(_amqConnection.getAggregateTicker()).thenReturn(new
AggregateTicker());
+ when(_amqConnection.getContextValue(Long.class,
Consumer.SUSPEND_NOTIFICATION_PERIOD))
+ .thenReturn(Consumer.SUSPEND_NOTIFICATION_PERIOD_DEFAULT);
+ doAnswer(invocation ->
+ {
+ ((Runnable) invocation.getArgument(0)).run();
+ return CompletableFuture.completedFuture(null);
+ }).when(_amqConnection).doOnIOThreadAsync(any(Runnable.class));
+ _channel = new AMQChannel(_amqConnection, 1,
_host.getMessageStore());
+ }
+
+ private AMQMessage enqueueMessage()
+ {
+ final ContentHeaderBody contentHeaderBody = new
ContentHeaderBody(new BasicContentHeaderProperties());
+ final MessageMetaData metaData = new MessageMetaData(new
MessagePublishInfo(), contentHeaderBody);
+ final AMQMessage message = new AMQMessage(new
StoredMemoryMessage<>(1L, metaData));
+ _queue.enqueue(message, null, null);
+ return message;
+ }
+
+ @Override
+ public void close()
+ {
+ try
+ {
+ _channel.close();
+ }
+ finally
+ {
+ _host.close();
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
