Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1071: JUnit 5 Support Improvements

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/1071

- Remove JUnit4 dependency from `RabbitAvailableCondition` (minor breaking API change)
- Add `purgeAfterEach` to `@RabbitAvailable`
- tabs not spaces in `RabbitAvailableCondition` (review with `?w=1`)
- `@LogLevels` now requires `level`

* Sonar, javadoc fixes; default log level,  per review comments; convert more tests.

* Remove unnecessary `defaultPort` field; 2 more conversions
This commit is contained in:
Gary Russell
2019-08-20 20:30:57 -04:00
committed by Artem Bilan
parent 3d5fe16be8
commit 50c17ea0e0
53 changed files with 1063 additions and 810 deletions
Download Patch File Download Diff File Expand all files Collapse all files

371
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java
View File

@@ -18,17 +18,7 @@ package org.springframework.amqp.rabbit.junit;
import static org.junit.Assert.fail;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -37,13 +27,9 @@ import org.junit.rules.TestWatcher;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
import org.springframework.util.Base64Utils;
import org.springframework.util.StringUtils;
import org.springframework.amqp.rabbit.junit.BrokerRunningSupport.BrokerNotAliveException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.http.client.Client;
/**
* A rule that prevents integration tests from failing if the Rabbit broker application is
@@ -68,8 +54,8 @@ import com.rabbitmq.http.client.Client;
* <p>Call {@link #removeTestQueues(String...)} from an {@code @After} method to remove
* those queues (and optionally others).
* <p>If you wish to enforce the broker being available, for example, on a CI server,
* set the environment variable {@value #BROKER_REQUIRED} to {@code true} and the
* tests will fail fast.
* set the environment variable {@value BrokerRunningSupport#BROKER_REQUIRED} to
* {@code true} and the tests will fail fast.
*
* @author Dave Syer
* @author Gary Russell
@@ -80,95 +66,30 @@ import com.rabbitmq.http.client.Client;
*/
public final class BrokerRunning extends TestWatcher {
private static final int SIXTEEN = 16;
private static final Log LOGGER = LogFactory.getLog(BrokerRunningSupport.class);
public static final String BROKER_ADMIN_URI = "RABBITMQ_TEST_ADMIN_URI";
public static final String BROKER_HOSTNAME = "RABBITMQ_TEST_HOSTNAME";
public static final String BROKER_PORT = "RABBITMQ_TEST_PORT";
public static final String BROKER_USER = "RABBITMQ_TEST_USER";
public static final String BROKER_PW = "RABBITMQ_TEST_PASSWORD";
public static final String BROKER_ADMIN_USER = "RABBITMQ_TEST_ADMIN_USER";
public static final String BROKER_ADMIN_PW = "RABBITMQ_TEST_ADMIN_PASSWORD";
public static final String BROKER_REQUIRED = "RABBITMQ_SERVER_REQUIRED";
private static final String DEFAULT_QUEUE_NAME = BrokerRunning.class.getName();
private static final String GUEST = "guest";
private static final Log logger = LogFactory.getLog(BrokerRunning.class); // NOSONAR - lower case
// Static so that we only test once on failure: speeds up test suite
private static final Map<Integer, Boolean> brokerOnline = new HashMap<Integer, Boolean>(); // NOSONAR - lower case
// Static so that we only test once on failure
private static final Map<Integer, Boolean> brokerOffline = new HashMap<Integer, Boolean>(); // NOSONAR - lower case
private static final Map<String, String> environmentOverrides = new HashMap<>(); // NOSONAR - lower case
private final BrokerRunningSupport brokerRunning;
private final boolean assumeOnline;
private final boolean purge;
private final boolean management;
private final String[] queues;
private final int defaultPort = fromEnvironment(BROKER_PORT, null) == null ? BrokerTestUtils.getPort()
: Integer.valueOf(fromEnvironment(BROKER_PORT, null));
private int port;
private String hostName = fromEnvironment(BROKER_HOSTNAME, "localhost");
private String adminUri = fromEnvironment(BROKER_ADMIN_URI, null);
private ConnectionFactory connectionFactory;
private String user = fromEnvironment(BROKER_USER, GUEST);
private String password = fromEnvironment(BROKER_PW, GUEST);
private String adminUser = fromEnvironment(BROKER_ADMIN_USER, GUEST);
private String adminPassword = fromEnvironment(BROKER_ADMIN_PW, GUEST);
private String fromEnvironment(String key, String defaultValue) {
String environmentValue = environmentOverrides.get(key);
if (!StringUtils.hasText(environmentValue)) {
environmentValue = System.getenv(key);
}
if (StringUtils.hasText(environmentValue)) {
return environmentValue;
}
else {
return defaultValue;
}
}
/**
* Set environment variable overrides for host, port etc. Will override any real
* environment variables, if present.
* <p><b>The variables will only apply to rule instances that are created after this
* method is called.</b>
* The overrides will remain until
* The overrides will remain until {@link #clearEnvironmentVariableOverrides()} is
* called.
* @param environmentVariables the variables.
*/
public static void setEnvironmentVariableOverrides(Map<String, String> environmentVariables) {
environmentOverrides.putAll(environmentVariables);
BrokerRunningSupport.setEnvironmentVariableOverrides(environmentVariables);
}
/**
* Clear any environment variable overrides set in {@link #setEnvironmentVariableOverrides(Map)}.
*/
public static void clearEnvironmentVariableOverrides() {
environmentOverrides.clear();
BrokerRunningSupport.clearEnvironmentVariableOverrides();
}
/**
@@ -218,15 +139,7 @@ public final class BrokerRunning extends TestWatcher {
private BrokerRunning(boolean assumeOnline, boolean purge, boolean management, String... queues) {
this.assumeOnline = assumeOnline;
if (queues != null) {
this.queues = Arrays.copyOf(queues, queues.length);
}
else {
this.queues = null;
}
this.purge = purge;
this.management = management;
setPort(this.defaultPort);
this.brokerRunning = new BrokerRunningSupport(assumeOnline, purge, management, queues);
}
private BrokerRunning(boolean assumeOnline, String... queues) {
@@ -234,31 +147,25 @@ public final class BrokerRunning extends TestWatcher {
}
private BrokerRunning(boolean assumeOnline) {
this(assumeOnline, DEFAULT_QUEUE_NAME);
this(assumeOnline, BrokerRunningSupport.DEFAULT_QUEUE_NAME);
}
private BrokerRunning(boolean assumeOnline, boolean purge, boolean management) {
this(assumeOnline, purge, management, DEFAULT_QUEUE_NAME);
this(assumeOnline, purge, management, BrokerRunningSupport.DEFAULT_QUEUE_NAME);
}
/**
* @param port the port to set
*/
public void setPort(int port) {
this.port = port;
if (!brokerOffline.containsKey(port)) {
brokerOffline.put(port, true);
}
if (!brokerOnline.containsKey(port)) {
brokerOnline.put(port, true);
}
this.brokerRunning.setPort(port);
}
/**
* @param hostName the hostName to set
*/
public void setHostName(String hostName) {
this.hostName = hostName;
this.brokerRunning.setHostName(hostName);
}
/**
@@ -267,7 +174,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public void setUser(String user) {
this.user = user;
this.brokerRunning.setUser(user);
}
/**
@@ -276,7 +183,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public void setPassword(String password) {
this.password = password;
this.brokerRunning.setPassword(password);
}
/**
@@ -285,7 +192,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public void setAdminUri(String adminUri) {
this.adminUri = adminUri;
this.brokerRunning.setAdminUri(adminUri);
}
/**
@@ -294,7 +201,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public void setAdminUser(String user) {
this.adminUser = user;
this.brokerRunning.setAdminUser(user);
}
/**
@@ -303,7 +210,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public void setAdminPassword(String password) {
this.adminPassword = password;
this.brokerRunning.setAdminPassword(password);
}
/**
@@ -312,7 +219,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public int getPort() {
return this.port;
return this.brokerRunning.getPort();
}
/**
@@ -321,7 +228,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public String getHostName() {
return this.hostName;
return this.brokerRunning.getHostName();
}
/**
@@ -330,7 +237,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public String getUser() {
return this.user;
return this.brokerRunning.getUser();
}
/**
@@ -339,7 +246,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public String getPassword() {
return this.password;
return this.brokerRunning.getPassword();
}
/**
@@ -348,7 +255,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public String getAdminUser() {
return this.adminUser;
return this.brokerRunning.getAdminUser();
}
/**
@@ -357,30 +264,17 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public String getAdminPassword() {
return this.adminPassword;
return this.brokerRunning.getAdminPassword();
}
@Override
public Statement apply(Statement base, Description description) {
// Check at the beginning, so this can be used as a static field
if (this.assumeOnline) {
Assume.assumeTrue(brokerOnline.get(this.port));
}
else {
Assume.assumeTrue(brokerOffline.get(this.port));
}
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
channel = createQueues(connection);
this.brokerRunning.test();
}
catch (Exception e) {
logger.warn("Not executing tests because basic connectivity test failed: " + e.getMessage());
brokerOnline.put(this.port, false);
catch (BrokerNotAliveException e) {
LOGGER.warn("Not executing tests because basic connectivity test failed: " + e.getMessage());
if (this.assumeOnline) {
if (fatal()) {
fail("RabbitMQ Broker is required, but not available");
@@ -390,74 +284,16 @@ public final class BrokerRunning extends TestWatcher {
}
}
}
finally {
closeResources(connection, channel);
}
return super.apply(base, description);
}
public void isUp() throws IOException, TimeoutException, URISyntaxException {
Connection connection = getConnectionFactory().newConnection(); // NOSONAR - closeResources()
Channel channel = null;
try {
channel = createQueues(connection);
}
finally {
closeResources(connection, channel);
}
}
private Connection getConnection(ConnectionFactory connectionFactory) throws IOException, TimeoutException {
Connection connection = connectionFactory.newConnection();
connection.setId(generateId());
return connection;
}
private Channel createQueues(Connection connection) throws IOException, MalformedURLException, URISyntaxException {
Channel channel;
channel = connection.createChannel();
for (String queueName : this.queues) {
if (this.purge) {
logger.debug("Deleting queue: " + queueName);
// Delete completely - gets rid of consumers and bindings as well
channel.queueDelete(queueName);
}
if (isDefaultQueue(queueName)) {
// Just for test probe.
channel.queueDelete(queueName);
}
else {
channel.queueDeclare(queueName, true, false, false, null);
}
}
brokerOffline.put(this.port, false);
if (!this.assumeOnline) {
Assume.assumeTrue(brokerOffline.get(this.port));
}
if (this.management) {
Client client = new Client(getAdminUri(), this.adminUser, this.adminPassword);
if (!client.alivenessTest("/")) {
throw new BrokerNotAliveException("Aliveness test failed for localhost:15672 guest/quest; "
+ "management not available");
}
}
return channel;
public void isUp() {
this.brokerRunning.test();
}
public static boolean fatal() {
String serversRequired = System.getenv(BROKER_REQUIRED);
if (Boolean.parseBoolean(serversRequired)) {
logger.error("RABBITMQ IS REQUIRED BUT NOT AVAILABLE");
return true;
}
else {
return false;
}
return BrokerRunningSupport.fatal();
}
/**
@@ -466,15 +302,7 @@ public final class BrokerRunning extends TestWatcher {
* @return the id.
*/
public String generateId() {
UUID uuid = UUID.randomUUID();
ByteBuffer bb = ByteBuffer.wrap(new byte[SIXTEEN]);
bb.putLong(uuid.getMostSignificantBits())
.putLong(uuid.getLeastSignificantBits());
return "SpringBrokerRunning." + Base64Utils.encodeToUrlSafeString(bb.array()).replaceAll("=", "");
}
private boolean isDefaultQueue(String queue) {
return DEFAULT_QUEUE_NAME.equals(queue);
return this.brokerRunning.generateId();
}
/**
@@ -484,30 +312,7 @@ public final class BrokerRunning extends TestWatcher {
* tests.
*/
public void removeTestQueues(String... additionalQueues) {
List<String> queuesToRemove = Arrays.asList(this.queues);
if (additionalQueues != null) {
queuesToRemove = new ArrayList<>(queuesToRemove);
queuesToRemove.addAll(Arrays.asList(additionalQueues));
}
logger.debug("deleting test queues: " + queuesToRemove);
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
connection.setId(generateId() + ".queueDelete");
channel = connection.createChannel();
for (String queue : queuesToRemove) {
channel.queueDelete(queue);
}
}
catch (Exception e) {
logger.warn("Failed to delete queues", e);
}
finally {
closeResources(connection, channel);
}
this.brokerRunning.removeTestQueues(additionalQueues);
}
/**
@@ -515,19 +320,7 @@ public final class BrokerRunning extends TestWatcher {
* a test might leave stale data and multiple tests use the same queue.
*/
public void purgeTestQueues() {
removeTestQueues();
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
channel = createQueues(connection);
}
catch (Exception e) {
logger.warn("Failed to re-declare queues during purge: " + e.getMessage());
}
finally {
closeResources(connection, channel);
}
this.brokerRunning.purgeTestQueues();
}
/**
@@ -535,24 +328,7 @@ public final class BrokerRunning extends TestWatcher {
* @param queuesToDelete the queues to delete.
*/
public void deleteQueues(String... queuesToDelete) {
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
connection.setId(generateId() + ".queueDelete");
channel = connection.createChannel();
for (String queue : queuesToDelete) {
channel.queueDelete(queue);
}
}
catch (Exception e) {
logger.warn("Failed to delete queues", e);
}
finally {
closeResources(connection, channel);
}
this.brokerRunning.deleteQueues(queuesToDelete);
}
/**
@@ -560,24 +336,7 @@ public final class BrokerRunning extends TestWatcher {
* @param exchanges the exchanges to delete.
*/
public void deleteExchanges(String... exchanges) {
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
connection.setId(generateId() + ".exchangeDelete");
channel = connection.createChannel();
for (String exchange : exchanges) {
channel.exchangeDelete(exchange);
}
}
catch (Exception e) {
logger.warn("Failed to delete queues", e);
}
finally {
closeResources(connection, channel);
}
this.brokerRunning.deleteExchanges(exchanges);
}
/**
@@ -585,20 +344,7 @@ public final class BrokerRunning extends TestWatcher {
* @return the connection factory.
*/
public ConnectionFactory getConnectionFactory() {
if (this.connectionFactory == null) {
this.connectionFactory = new ConnectionFactory();
if (StringUtils.hasText(this.hostName)) {
this.connectionFactory.setHost(this.hostName);
}
else {
this.connectionFactory.setHost("localhost");
}
this.connectionFactory.setPort(this.port);
this.connectionFactory.setUsername(this.user);
this.connectionFactory.setPassword(this.password);
this.connectionFactory.setAutomaticRecoveryEnabled(false);
}
return this.connectionFactory;
return this.brokerRunning.getConnectionFactory();
}
/**
@@ -607,48 +353,7 @@ public final class BrokerRunning extends TestWatcher {
* @since 1.7.2
*/
public String getAdminUri() {
if (!StringUtils.hasText(this.adminUri)) {
if (!StringUtils.hasText(this.hostName)) {
this.adminUri = "http://localhost:15672/api/";
}
else {
this.adminUri = "http://" + this.hostName + ":15672/api/";
}
}
return this.adminUri;
}
private void closeResources(Connection connection, Channel channel) {
if (channel != null) {
try {
channel.close();
}
catch (@SuppressWarnings("unused") IOException | TimeoutException e) {
// Ignore
}
}
if (connection != null) {
try {
connection.close();
}
catch (@SuppressWarnings("unused") IOException e) {
// Ignore
}
}
}
/**
* The {@link RuntimeException} thrown when broker is not available
* on the provided host port.
*/
public static class BrokerNotAliveException extends RuntimeException {
private static final long serialVersionUID = 1L;
BrokerNotAliveException(String message) {
super(message);
}
return this.brokerRunning.getAdminUri();
}
}

622
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunningSupport.java Normal file
View File

@@ -0,0 +1,622 @@
/*
* Copyright 2002-2019 the original author or authors.
*
* Licensed 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
*
* https://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.springframework.amqp.rabbit.junit;
import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Assume;
import org.springframework.util.Base64Utils;
import org.springframework.util.StringUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.http.client.Client;
/**
* A class that can be used to prevent integration tests from failing if the Rabbit broker application is
* not running or not accessible. If the Rabbit broker is not running in the background
* all the tests here will simply be skipped (by default) because of a violated assumption
* (showing as successful).
* <p>
* If you wish to enforce the broker being available, for example, on a CI server,
* set the environment variable {@value #BROKER_REQUIRED} to {@code true} and the
* tests will fail fast.
*
* @author Dave Syer
* @author Gary Russell
*
* @since 2.2
*/
public final class BrokerRunningSupport {
private static final int SIXTEEN = 16;
public static final String BROKER_ADMIN_URI = "RABBITMQ_TEST_ADMIN_URI";
public static final String BROKER_HOSTNAME = "RABBITMQ_TEST_HOSTNAME";
public static final String BROKER_PORT = "RABBITMQ_TEST_PORT";
public static final String BROKER_USER = "RABBITMQ_TEST_USER";
public static final String BROKER_PW = "RABBITMQ_TEST_PASSWORD";
public static final String BROKER_ADMIN_USER = "RABBITMQ_TEST_ADMIN_USER";
public static final String BROKER_ADMIN_PW = "RABBITMQ_TEST_ADMIN_PASSWORD";
public static final String BROKER_REQUIRED = "RABBITMQ_SERVER_REQUIRED";
public static final String DEFAULT_QUEUE_NAME = BrokerRunningSupport.class.getName();
private static final String GUEST = "guest";
private static final Log LOGGER = LogFactory.getLog(BrokerRunningSupport.class);
// Static so that we only test once on failure: speeds up test suite
private static final Map<Integer, Boolean> BROKER_ONLINE = new HashMap<>();
// Static so that we only test once on failure
private static final Map<Integer, Boolean> BROKER_OFFLINE = new HashMap<>();
private static final Map<String, String> ENVIRONMENT_OVERRIDES = new HashMap<>();
private final boolean assumeOnline;
private final boolean purge;
private final boolean management;
private final String[] queues;
private int port;
private String hostName = fromEnvironment(BROKER_HOSTNAME, "localhost");
private String adminUri = fromEnvironment(BROKER_ADMIN_URI, null);
private ConnectionFactory connectionFactory;
private String user = fromEnvironment(BROKER_USER, GUEST);
private String password = fromEnvironment(BROKER_PW, GUEST);
private String adminUser = fromEnvironment(BROKER_ADMIN_USER, GUEST);
private String adminPassword = fromEnvironment(BROKER_ADMIN_PW, GUEST);
private boolean purgeAfterEach;
private static String fromEnvironment(String key, String defaultValue) {
String environmentValue = ENVIRONMENT_OVERRIDES.get(key);
if (!StringUtils.hasText(environmentValue)) {
environmentValue = System.getenv(key);
}
if (StringUtils.hasText(environmentValue)) {
return environmentValue;
}
else {
return defaultValue;
}
}
/**
* Set environment variable overrides for host, port etc. Will override any real
* environment variables, if present.
* <p><b>The variables will only apply to rule instances that are created after this
* method is called.</b>
* The overrides will remain until
* @param environmentVariables the variables.
*/
public static void setEnvironmentVariableOverrides(Map<String, String> environmentVariables) {
ENVIRONMENT_OVERRIDES.putAll(environmentVariables);
}
/**
* Clear any environment variable overrides set in {@link #setEnvironmentVariableOverrides(Map)}.
*/
public static void clearEnvironmentVariableOverrides() {
ENVIRONMENT_OVERRIDES.clear();
}
/**
* Ensure the broker is running and has a empty queue(s) with the specified name(s) in the
* default exchange.
*
* @param names the queues to declare for the test.
* @return a new rule that assumes an existing running broker
*/
public static BrokerRunningSupport isRunningWithEmptyQueues(String... names) {
return new BrokerRunningSupport(true, true, names);
}
/**
* @return a new rule that assumes an existing running broker
*/
public static BrokerRunningSupport isRunning() {
return new BrokerRunningSupport(true);
}
/**
* @return a new rule that assumes there is no existing broker
*/
public static BrokerRunningSupport isNotRunning() {
return new BrokerRunningSupport(false);
}
/**
* @return a new rule that assumes an existing broker with the management plugin
*/
public static BrokerRunningSupport isBrokerAndManagementRunning() {
return new BrokerRunningSupport(true, false, true);
}
/**
* @param queues the queues.
* @return a new rule that assumes an existing broker with the management plugin with
* the provided queues declared (and emptied if needed)..
*/
public static BrokerRunningSupport isBrokerAndManagementRunningWithEmptyQueues(String...queues) {
return new BrokerRunningSupport(true, false, true, queues);
}
private BrokerRunningSupport(boolean assumeOnline, boolean purge, String... queues) {
this(assumeOnline, purge, false, queues);
}
BrokerRunningSupport(boolean assumeOnline, boolean purge, boolean management, String... queues) {
this.assumeOnline = assumeOnline;
if (queues != null) {
this.queues = Arrays.copyOf(queues, queues.length);
}
else {
this.queues = null;
}
this.purge = purge;
this.management = management;
setPort(fromEnvironment(BROKER_PORT, null) == null
? BrokerTestUtils.getPort()
: Integer.valueOf(fromEnvironment(BROKER_PORT, null)));
}
private BrokerRunningSupport(boolean assumeOnline, String... queues) {
this(assumeOnline, false, queues);
}
private BrokerRunningSupport(boolean assumeOnline) {
this(assumeOnline, DEFAULT_QUEUE_NAME);
}
private BrokerRunningSupport(boolean assumeOnline, boolean purge, boolean management) {
this(assumeOnline, purge, management, DEFAULT_QUEUE_NAME);
}
/**
* @param port the port to set
*/
public void setPort(int port) {
this.port = port;
if (!BROKER_OFFLINE.containsKey(port)) {
BROKER_OFFLINE.put(port, true);
}
if (!BROKER_ONLINE.containsKey(port)) {
BROKER_ONLINE.put(port, true);
}
}
/**
* @param hostName the hostName to set
*/
public void setHostName(String hostName) {
this.hostName = hostName;
}
/**
* Set the user for the amqp connection default "guest".
* @param user the user.
*/
public void setUser(String user) {
this.user = user;
}
/**
* Set the password for the amqp connection default "guest".
* @param password the password.
*/
public void setPassword(String password) {
this.password = password;
}
/**
* Set the uri for the REST API.
* @param adminUri the uri.
*/
public void setAdminUri(String adminUri) {
this.adminUri = adminUri;
}
/**
* Set the user for the management REST API connection default "guest".
* @param user the user.
*/
public void setAdminUser(String user) {
this.adminUser = user;
}
/**
* Set the password for the management REST API connection default "guest".
* @param password the password.
*/
public void setAdminPassword(String password) {
this.adminPassword = password;
}
/**
* Return the port.
* @return the port.
*/
public int getPort() {
return this.port;
}
/**
* Return the port.
* @return the port.
*/
public String getHostName() {
return this.hostName;
}
/**
* Return the user.
* @return the user.
*/
public String getUser() {
return this.user;
}
/**
* Return the password.
* @return the password.
*/
public String getPassword() {
return this.password;
}
/**
* Return the admin user.
* @return the user.
*/
public String getAdminUser() {
return this.adminUser;
}
/**
* Return the admin password.
* @return the password.
*/
public String getAdminPassword() {
return this.adminPassword;
}
public boolean isPurgeAfterEach() {
return this.purgeAfterEach;
}
/**
* Purge the test queues after each test (JUnit 5).
* @param purgeAfterEach true to purge.
*/
public void setPurgeAfterEach(boolean purgeAfterEach) {
this.purgeAfterEach = purgeAfterEach;
}
public void test() {
// Check at the beginning, so this can be used as a static field
if (this.assumeOnline) {
if (Boolean.FALSE.equals(BROKER_ONLINE.get(this.port))) {
throw new BrokerNotAliveException("Require broker online and it's not");
}
}
else {
if (Boolean.FALSE.equals(BROKER_OFFLINE.get(this.port))) {
throw new BrokerNotAliveException("Require broker offline and it's not");
}
}
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
channel = createQueues(connection);
}
catch (Exception e) {
LOGGER.warn("Not executing tests because basic connectivity test failed: " + e.getMessage());
BROKER_ONLINE.put(this.port, false);
if (this.assumeOnline) {
if (fatal()) {
throw new BrokerNotAliveException("RabbitMQ Broker is required, but not available");
}
else {
Assume.assumeNoException(e);
}
}
}
finally {
closeResources(connection, channel);
}
}
private Connection getConnection(ConnectionFactory cf) throws IOException, TimeoutException {
Connection connection = cf.newConnection();
connection.setId(generateId());
return connection;
}
private Channel createQueues(Connection connection) throws IOException, URISyntaxException {
Channel channel;
channel = connection.createChannel();
for (String queueName : this.queues) {
if (this.purge) {
LOGGER.debug("Deleting queue: " + queueName);
// Delete completely - gets rid of consumers and bindings as well
channel.queueDelete(queueName);
}
if (isDefaultQueue(queueName)) {
// Just for test probe.
channel.queueDelete(queueName);
}
else {
channel.queueDeclare(queueName, true, false, false, null);
}
}
BROKER_OFFLINE.put(this.port, false);
if (!this.assumeOnline) {
Assume.assumeTrue(BROKER_OFFLINE.get(this.port));
}
if (this.management) {
Client client = new Client(getAdminUri(), this.adminUser, this.adminPassword);
if (!client.alivenessTest("/")) {
throw new BrokerNotAliveException("Aliveness test failed for localhost:15672 guest/quest; "
+ "management not available");
}
}
return channel;
}
public static boolean fatal() {
String serversRequired = System.getenv(BROKER_REQUIRED);
if (Boolean.parseBoolean(serversRequired)) {
LOGGER.error("RABBITMQ IS REQUIRED BUT NOT AVAILABLE");
return true;
}
else {
return false;
}
}
/**
* Generate the connection id for the connection used by the rule's
* connection factory.
* @return the id.
*/
public String generateId() {
UUID uuid = UUID.randomUUID();
ByteBuffer bb = ByteBuffer.wrap(new byte[SIXTEEN]);
bb.putLong(uuid.getMostSignificantBits())
.putLong(uuid.getLeastSignificantBits());
return "SpringBrokerRunning." + Base64Utils.encodeToUrlSafeString(bb.array()).replaceAll("=", "");
}
private boolean isDefaultQueue(String queue) {
return DEFAULT_QUEUE_NAME.equals(queue);
}
/**
* Remove any test queues that were created by an
* {@link #isRunningWithEmptyQueues(String...)} method.
* @param additionalQueues additional queues to remove that might have been created by
* tests.
*/
public void removeTestQueues(String... additionalQueues) {
List<String> queuesToRemove = Arrays.asList(this.queues);
if (additionalQueues != null) {
queuesToRemove = new ArrayList<>(queuesToRemove);
queuesToRemove.addAll(Arrays.asList(additionalQueues));
}
LOGGER.debug("deleting test queues: " + queuesToRemove);
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
connection.setId(generateId() + ".queueDelete");
channel = connection.createChannel();
for (String queue : queuesToRemove) {
channel.queueDelete(queue);
}
}
catch (Exception e) {
LOGGER.warn("Failed to delete queues", e);
}
finally {
closeResources(connection, channel);
}
}
/**
* Delete and re-declare all the configured queues. Can be used between tests when
* a test might leave stale data and multiple tests use the same queue.
*/
public void purgeTestQueues() {
removeTestQueues();
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
channel = createQueues(connection);
}
catch (Exception e) {
LOGGER.warn("Failed to re-declare queues during purge: " + e.getMessage());
}
finally {
closeResources(connection, channel);
}
}
/**
* Delete arbitrary queues from the broker.
* @param queuesToDelete the queues to delete.
*/
public void deleteQueues(String... queuesToDelete) {
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
connection.setId(generateId() + ".queueDelete");
channel = connection.createChannel();
for (String queue : queuesToDelete) {
channel.queueDelete(queue);
}
}
catch (Exception e) {
LOGGER.warn("Failed to delete queues", e);
}
finally {
closeResources(connection, channel);
}
}
/**
* Delete arbitrary exchanges from the broker.
* @param exchanges the exchanges to delete.
*/
public void deleteExchanges(String... exchanges) {
Connection connection = null; // NOSONAR (closeResources())
Channel channel = null;
try {
connection = getConnection(getConnectionFactory());
connection.setId(generateId() + ".exchangeDelete");
channel = connection.createChannel();
for (String exchange : exchanges) {
channel.exchangeDelete(exchange);
}
}
catch (Exception e) {
LOGGER.warn("Failed to delete queues", e);
}
finally {
closeResources(connection, channel);
}
}
/**
* Get the connection factory used by this rule.
* @return the connection factory.
*/
public ConnectionFactory getConnectionFactory() {
if (this.connectionFactory == null) {
this.connectionFactory = new ConnectionFactory();
if (StringUtils.hasText(this.hostName)) {
this.connectionFactory.setHost(this.hostName);
}
else {
this.connectionFactory.setHost("localhost");
}
this.connectionFactory.setPort(this.port);
this.connectionFactory.setUsername(this.user);
this.connectionFactory.setPassword(this.password);
this.connectionFactory.setAutomaticRecoveryEnabled(false);
}
return this.connectionFactory;
}
/**
* Return the admin uri.
* @return the uri.
*/
public String getAdminUri() {
if (!StringUtils.hasText(this.adminUri)) {
if (!StringUtils.hasText(this.hostName)) {
this.adminUri = "http://localhost:15672/api/";
}
else {
this.adminUri = "http://" + this.hostName + ":15672/api/";
}
}
return this.adminUri;
}
private void closeResources(Connection connection, Channel channel) {
if (channel != null) {
try {
channel.close();
}
catch (@SuppressWarnings("unused") IOException | TimeoutException e) {
// Ignore
}
}
if (connection != null) {
try {
connection.close();
}
catch (@SuppressWarnings("unused") IOException e) {
// Ignore
}
}
}
/**
* The {@link RuntimeException} thrown when broker is not available
* on the provided host port.
*/
public static class BrokerNotAliveException extends RuntimeException {
private static final long serialVersionUID = 1L;
BrokerNotAliveException(String message) {
super(message);
}
}
}

4
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/LogLevels.java
View File

@@ -53,8 +53,8 @@ public @interface LogLevels {
/**
* The Log4j level name to switch the categories to during the test.
* @return the level.
* @return the level (default DEBUG).
*/
String level() default "";
String level() default "DEBUG";
}

20
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/LogLevelsCondition.java
View File

@@ -74,10 +74,12 @@ public class LogLevelsCondition
store = parent.getStore(Namespace.create(getClass(), parent));
logLevels = store.get(STORE_ANNOTATION_KEY, LogLevels.class);
}
store.put(STORE_CONTAINER_KEY, JUnitUtils.adjustLogLevels(context.getDisplayName(),
Arrays.asList((logLevels.classes())),
Arrays.asList(logLevels.categories()),
Level.toLevel(logLevels.level())));
if (logLevels != null) {
store.put(STORE_CONTAINER_KEY, JUnitUtils.adjustLogLevels(context.getDisplayName(),
Arrays.asList((logLevels.classes())),
Arrays.asList(logLevels.categories()),
Level.toLevel(logLevels.level())));
}
}
@Override
@@ -91,10 +93,12 @@ public class LogLevelsCondition
container = store.get(STORE_CONTAINER_KEY, LevelsContainer.class);
parentStore = true;
}
JUnitUtils.revertLevels(context.getDisplayName(), container);
store.remove(STORE_CONTAINER_KEY);
if (!parentStore) {
store.remove(STORE_ANNOTATION_KEY);
if (container != null) {
JUnitUtils.revertLevels(context.getDisplayName(), container);
store.remove(STORE_CONTAINER_KEY);
if (!parentStore) {
store.remove(STORE_ANNOTATION_KEY);
}
}
}

7
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailable.java
View File

@@ -62,4 +62,11 @@ public @interface RabbitAvailable {
*/
boolean management() default false;
/**
* Purge the test queues after each test.
* @return true to purge (default).
* @since 2.2
*/
boolean purgeAfterEach() default true;
}

171
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java
View File

@@ -20,6 +20,7 @@ import java.lang.reflect.AnnotatedElement;
import java.util.Optional;
import org.junit.jupiter.api.extension.AfterAllCallback;
import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.ConditionEvaluationResult;
import org.junit.jupiter.api.extension.ExecutionCondition;
import org.junit.jupiter.api.extension.ExtensionContext;
@@ -35,100 +36,110 @@ import org.springframework.util.Assert;
import com.rabbitmq.client.ConnectionFactory;
/**
* JUnit5 {@link ExecutionCondition}.
* Looks for {@code @RabbitAvailable} annotated classes and disables
* if found the broker is not available.
* JUnit5 {@link ExecutionCondition}. Looks for {@code @RabbitAvailable} annotated classes
* and disables if found the broker is not available.
*
* @author Gary Russell
* @since 2.0.2
*
*/
public class RabbitAvailableCondition implements ExecutionCondition, AfterAllCallback, ParameterResolver {
public class RabbitAvailableCondition
implements ExecutionCondition, AfterEachCallback, AfterAllCallback, ParameterResolver {
private static final String BROKER_RUNNING_BEAN = "brokerRunning";
private static final String BROKER_RUNNING_BEAN = "brokerRunning";
private static final ConditionEvaluationResult ENABLED = ConditionEvaluationResult.enabled(
"@RabbitAvailable is not present");
private static final ConditionEvaluationResult ENABLED = ConditionEvaluationResult.enabled(
"@RabbitAvailable is not present");
private static final ThreadLocal<BrokerRunning> brokerRunningHolder = new ThreadLocal<>(); // NOSONAR - lower case
private static final ThreadLocal<BrokerRunningSupport> BROKER_RUNNING_HOLDER = new ThreadLocal<>();
@Override
public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) {
Optional<AnnotatedElement> element = context.getElement();
MergedAnnotations annotations = MergedAnnotations.from(element.get(),
MergedAnnotations.SearchStrategy.TYPE_HIERARCHY);
if (annotations.get(RabbitAvailable.class).isPresent()) {
RabbitAvailable rabbit = annotations.get(RabbitAvailable.class).synthesize();
try {
String[] queues = rabbit.queues();
BrokerRunning brokerRunning = getStore(context).get(BROKER_RUNNING_BEAN, BrokerRunning.class);
if (brokerRunning == null) {
if (rabbit.management()) {
brokerRunning = BrokerRunning.isBrokerAndManagementRunningWithEmptyQueues(queues);
}
else {
brokerRunning = BrokerRunning.isRunningWithEmptyQueues(queues);
}
}
brokerRunning.isUp();
brokerRunningHolder.set(brokerRunning);
Store store = getStore(context);
store.put(BROKER_RUNNING_BEAN, brokerRunning);
store.put("queuesToDelete", queues);
return ConditionEvaluationResult.enabled("RabbitMQ is available");
}
catch (Exception e) {
if (BrokerRunning.fatal()) {
throw new IllegalStateException("Required RabbitMQ is not available", e);
}
return ConditionEvaluationResult.disabled("RabbitMQ is not available");
}
}
return ENABLED;
}
@Override
public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) {
Optional<AnnotatedElement> element = context.getElement();
MergedAnnotations annotations = MergedAnnotations.from(element.get(),
MergedAnnotations.SearchStrategy.TYPE_HIERARCHY);
if (annotations.get(RabbitAvailable.class).isPresent()) {
RabbitAvailable rabbit = annotations.get(RabbitAvailable.class).synthesize();
try {
String[] queues = rabbit.queues();
BrokerRunningSupport brokerRunning = getStore(context).get(BROKER_RUNNING_BEAN,
BrokerRunningSupport.class);
if (brokerRunning == null) {
if (rabbit.management()) {
brokerRunning = BrokerRunningSupport.isBrokerAndManagementRunningWithEmptyQueues(queues);
}
else {
brokerRunning = BrokerRunningSupport.isRunningWithEmptyQueues(queues);
}
}
brokerRunning.setPurgeAfterEach(rabbit.purgeAfterEach());
brokerRunning.test();
BROKER_RUNNING_HOLDER.set(brokerRunning);
Store store = getStore(context);
store.put(BROKER_RUNNING_BEAN, brokerRunning);
store.put("queuesToDelete", queues);
return ConditionEvaluationResult.enabled("RabbitMQ is available");
}
catch (Exception e) {
if (BrokerRunningSupport.fatal()) {
throw new IllegalStateException("Required RabbitMQ is not available", e);
}
return ConditionEvaluationResult.disabled("RabbitMQ is not available");
}
}
return ENABLED;
}
@Override
public void afterAll(ExtensionContext context) {
brokerRunningHolder.remove();
Store store = getStore(context);
BrokerRunning brokerRunning = store.remove(BROKER_RUNNING_BEAN, BrokerRunning.class);
if (brokerRunning != null) {
brokerRunning.removeTestQueues();
}
}
@Override
public void afterEach(ExtensionContext context) {
BrokerRunningSupport brokerRunning = BROKER_RUNNING_HOLDER.get();
if (brokerRunning != null && brokerRunning.isPurgeAfterEach()) {
brokerRunning.purgeTestQueues();
}
}
@Override
public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext)
throws ParameterResolutionException {
Class<?> type = parameterContext.getParameter().getType();
return type.equals(ConnectionFactory.class) || type.equals(BrokerRunning.class);
}
@Override
public void afterAll(ExtensionContext context) {
BROKER_RUNNING_HOLDER.remove();
Store store = getStore(context);
BrokerRunningSupport brokerRunning = store.remove(BROKER_RUNNING_BEAN, BrokerRunningSupport.class);
if (brokerRunning != null) {
brokerRunning.removeTestQueues();
}
}
@Override
public Object resolveParameter(ParameterContext parameterContext, ExtensionContext context)
throws ParameterResolutionException {
// in parent for method injection, Composite key causes a store miss
BrokerRunning brokerRunning =
getParentStore(context).get(BROKER_RUNNING_BEAN, BrokerRunning.class) == null
? getStore(context).get(BROKER_RUNNING_BEAN, BrokerRunning.class)
: getParentStore(context).get(BROKER_RUNNING_BEAN, BrokerRunning.class);
Assert.state(brokerRunning != null, "Could not find brokerRunning instance");
Class<?> type = parameterContext.getParameter().getType();
return type.equals(ConnectionFactory.class) ? brokerRunning.getConnectionFactory()
: brokerRunning;
}
@Override
public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext)
throws ParameterResolutionException {
Class<?> type = parameterContext.getParameter().getType();
return type.equals(ConnectionFactory.class) || type.equals(BrokerRunningSupport.class);
}
private Store getStore(ExtensionContext context) {
return context.getStore(Namespace.create(getClass(), context));
}
@Override
public Object resolveParameter(ParameterContext parameterContext, ExtensionContext context)
throws ParameterResolutionException {
// in parent for method injection, Composite key causes a store miss
BrokerRunningSupport brokerRunning = getParentStore(context).get(BROKER_RUNNING_BEAN,
BrokerRunningSupport.class) == null
? getStore(context).get(BROKER_RUNNING_BEAN, BrokerRunningSupport.class)
: getParentStore(context).get(BROKER_RUNNING_BEAN, BrokerRunningSupport.class);
Assert.state(brokerRunning != null, "Could not find brokerRunning instance");
Class<?> type = parameterContext.getParameter().getType();
return type.equals(ConnectionFactory.class) ? brokerRunning.getConnectionFactory()
: brokerRunning;
}
private Store getParentStore(ExtensionContext context) {
ExtensionContext parent = context.getParent().get();
return parent.getStore(Namespace.create(getClass(), parent));
}
private Store getStore(ExtensionContext context) {
return context.getStore(Namespace.create(getClass(), context));
}
public static BrokerRunning getBrokerRunning() {
return brokerRunningHolder.get();
}
private Store getParentStore(ExtensionContext context) {
ExtensionContext parent = context.getParent().get();
return parent.getStore(Namespace.create(getClass(), parent));
}
public static BrokerRunningSupport getBrokerRunning() {
return BROKER_RUNNING_HOLDER.get();
}
}

4
spring-rabbit-junit/src/test/java/org/springframework/amqp/rabbit/junit/BrokerRunningTests.java
View File

@@ -72,8 +72,8 @@ public class BrokerRunningTests {
assertThat(connectionFactory.getUsername()).isEqualTo("FIZ");
assertThat(connectionFactory.getPassword()).isEqualTo("QUX");
DirectFieldAccessor dfa = new DirectFieldAccessor(brokerRunning);
assertThat(dfa.getPropertyValue("adminUser")).isEqualTo("BAR");
assertThat(dfa.getPropertyValue("adminPassword")).isEqualTo("FOO");
assertThat(dfa.getPropertyValue("brokerRunning.adminUser")).isEqualTo("BAR");
assertThat(dfa.getPropertyValue("brokerRunning.adminPassword")).isEqualTo("FOO");
BrokerRunning.clearEnvironmentVariableOverrides();
}

2
spring-rabbit-junit/src/test/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCTORInjectionTests.java
View File

@@ -35,7 +35,7 @@ public class RabbitAvailableCTORInjectionTests {
private final ConnectionFactory connectionFactory;
public RabbitAvailableCTORInjectionTests(BrokerRunning brokerRunning) {
public RabbitAvailableCTORInjectionTests(BrokerRunningSupport brokerRunning) {
this.connectionFactory = brokerRunning.getConnectionFactory();
}

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactoryTests.java
View File

@@ -36,7 +36,7 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.amqp.utils.test.TestUtils;

16
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachePropertiesTests.java
View File

@@ -21,18 +21,15 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.rabbitmq.client.Channel;
@@ -41,14 +38,11 @@ import com.rabbitmq.client.Channel;
* @since 1.6
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
@RabbitAvailable
public class CachePropertiesTests {
@ClassRule
public static BrokerRunning br = BrokerRunning.isRunning();
@Autowired
private CachingConnectionFactory channelCf;

42
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.amqp.rabbit.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.anyString;
@@ -42,12 +43,10 @@ import javax.net.SocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.logging.log4j.Level;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpApplicationContextClosedException;
import org.springframework.amqp.AmqpAuthenticationException;
@@ -59,9 +58,10 @@ import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LogLevelAdjuster;
import org.springframework.amqp.rabbit.junit.LogLevels;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.ApplicationContext;
@@ -79,9 +79,12 @@ import com.rabbitmq.client.DefaultConsumer;
* @since 1.0
*
*/
@RabbitAvailable(queues = CachingConnectionFactoryIntegrationTests.CF_INTEGRATION_TEST_QUEUE)
@LogLevels(classes = { CachingConnectionFactoryIntegrationTests.class,
CachingConnectionFactory.class }, categories = "com.rabbitmq", level = "DEBUG")
public class CachingConnectionFactoryIntegrationTests {
private static final String CF_INTEGRATION_TEST_QUEUE = "cfIntegrationTest";
public static final String CF_INTEGRATION_TEST_QUEUE = "cfIntegrationTest";
private static final String CF_INTEGRATION_CONNECTION_NAME = "cfIntegrationTestConnectionName";
@@ -89,15 +92,7 @@ public class CachingConnectionFactoryIntegrationTests {
private CachingConnectionFactory connectionFactory;
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(CF_INTEGRATION_TEST_QUEUE);
@Rule
public LogLevelAdjuster adjuster = new LogLevelAdjuster(Level.DEBUG,
CachingConnectionFactoryIntegrationTests.class, CachingConnectionFactory.class)
.categories("com.rabbitmq");
@Before
@BeforeEach
public void open() {
connectionFactory = new CachingConnectionFactory("localhost");
connectionFactory.setPort(BrokerTestUtils.getPort());
@@ -105,10 +100,10 @@ public class CachingConnectionFactoryIntegrationTests {
connectionFactory.setConnectionNameStrategy(cf -> CF_INTEGRATION_CONNECTION_NAME);
}
@After
@AfterEach
public void close() {
if (!this.connectionFactory.getVirtualHost().equals("non-existent")) {
this.brokerIsRunning.removeTestQueues();
RabbitAvailableCondition.getBrokerRunning().purgeTestQueues();
}
assertThat(connectionFactory.getRabbitConnectionFactory().getClientProperties().get("foo")).isEqualTo("bar");
connectionFactory.destroy();
@@ -404,7 +399,7 @@ public class CachingConnectionFactoryIntegrationTests {
}
@Test
@Ignore // Don't run this on the CI build server
@Disabled // Don't run this on the CI build server
public void hangOnClose() throws Exception {
final Socket proxy = SocketFactory.getDefault().createSocket("localhost", 5672);
final ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(2765);
@@ -461,12 +456,13 @@ public class CachingConnectionFactoryIntegrationTests {
factory.destroy();
}
@Test(expected = AmqpResourceNotAvailableException.class)
@Test
public void testChannelMax() {
this.connectionFactory.getRabbitConnectionFactory().setRequestedChannelMax(1);
Connection connection = this.connectionFactory.createConnection();
connection.createChannel(true);
connection.createChannel(false);
assertThatExceptionOfType(AmqpResourceNotAvailableException.class)
.isThrownBy(() -> connection.createChannel(false));
}
}

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
View File

@@ -65,8 +65,8 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.commons.logging.Log;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
@@ -1711,7 +1711,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
}
@Test
@Ignore // Test to verify log message is suppressed after patch to CCF
@Disabled // Test to verify log message is suppressed after patch to CCF
public void testReturnsNormalCloseDeferredClose() throws Exception {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java
View File

@@ -28,7 +28,7 @@ import static org.mockito.Mockito.when;
import java.util.concurrent.ExecutorService;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;

9
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryLifecycleTests.java
View File

@@ -23,15 +23,14 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpApplicationContextClosedException;
import org.springframework.amqp.core.AnonymousQueue;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.context.ApplicationListener;
import org.springframework.context.SmartLifecycle;
@@ -51,11 +50,9 @@ import com.rabbitmq.client.impl.AMQImpl;
* @since 1.5.3
*
*/
@RabbitAvailable
public class ConnectionFactoryLifecycleTests {
@Rule
public BrokerRunning brokerRunning = BrokerRunning.isRunning();
@Test
public void testConnectionFactoryAvailableDuringStop() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class);

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtilsTests.java
View File

@@ -19,7 +19,7 @@ package org.springframework.amqp.rabbit.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.transaction.support.TransactionSynchronizationManager;

17
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactoryIntegrationTests.java
View File

@@ -20,31 +20,28 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.UUID;
import org.junit.After;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
/**
*
* @author Gary Russell
*/
@RabbitAvailable(management = true)
public class LocalizedQueueConnectionFactoryIntegrationTests {
@ClassRule
public static BrokerRunning brokerRunning = BrokerRunning.isBrokerAndManagementRunning();
private LocalizedQueueConnectionFactory lqcf;
private CachingConnectionFactory defaultConnectionFactory;
@Before
@BeforeEach
public void setup() {
this.defaultConnectionFactory = new CachingConnectionFactory("localhost");
String[] addresses = new String[] { "localhost:9999", "localhost:5672" };
@@ -57,7 +54,7 @@ public class LocalizedQueueConnectionFactoryIntegrationTests {
adminUris, nodes, vhost, username, password, false, null);
}
@After
@AfterEach
public void tearDown() {
this.lqcf.destroy();
this.defaultConnectionFactory.destroy();

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactoryTests.java
View File

@@ -38,7 +38,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.internal.stubbing.answers.CallsRealMethods;

13
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RabbitReconnectProblemTests.java
View File

@@ -22,9 +22,8 @@ import java.util.Map;
import java.util.concurrent.Semaphore;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.AmqpTemplate;
@@ -35,8 +34,7 @@ import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import com.rabbitmq.client.ConnectionFactory;
@@ -46,9 +44,8 @@ import com.rabbitmq.client.ConnectionFactory;
* @since 1.5.6
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@Ignore("Requires user interaction")
@SpringJUnitConfig
@Disabled("Requires user interaction")
public class RabbitReconnectProblemTests {
@Autowired

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/RoutingConnectionFactoryTests.java
View File

@@ -37,7 +37,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SSLConnectionTests.java
View File

@@ -29,8 +29,8 @@ import java.util.Collections;
import javax.net.ssl.SSLContext;
import org.apache.commons.logging.Log;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
@@ -55,7 +55,7 @@ import com.rabbitmq.client.ConnectionFactory;
public class SSLConnectionTests {
@Test
@Ignore
@Disabled
public void test() throws Exception {
RabbitConnectionFactoryBean fb = new RabbitConnectionFactoryBean();
fb.setUseSSL(true);

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java
View File

@@ -30,7 +30,7 @@ import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConnectionFactory;

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java
View File

@@ -38,7 +38,7 @@ import org.springframework.amqp.core.QueueBuilder.MasterLocator;
import org.springframework.amqp.core.QueueBuilder.Overflow;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerRunningSupport;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
@@ -64,7 +64,7 @@ import com.rabbitmq.http.client.domain.QueueInfo;
@RabbitAvailable(management = true)
public class FixedReplyQueueDeadLetterTests {
private static BrokerRunning brokerRunning;
private static BrokerRunningSupport brokerRunning;
@Autowired
private RabbitTemplate rabbitTemplate;

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java
View File

@@ -30,10 +30,8 @@ import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer;
import org.springframework.amqp.rabbit.support.ActiveObjectCounter;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
@@ -55,8 +53,6 @@ public class RabbitBindingIntegrationTests {
private RabbitTemplate template;
public BrokerRunning brokerIsRunning = RabbitAvailableCondition.getBrokerRunning();
@BeforeEach
public void setup() {
connectionFactory = new CachingConnectionFactory(BrokerTestUtils.getPort());
@@ -66,7 +62,6 @@ public class RabbitBindingIntegrationTests {
@AfterEach
public void cleanUp() {
this.brokerIsRunning.purgeTestQueues();
this.template.stop();
this.connectionFactory.destroy();
}

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitGatewaySupportTests.java
View File

@@ -22,7 +22,7 @@ import static org.mockito.Mockito.mock;
import java.util.ArrayList;
import java.util.List;
import org.junit.Test;
import org.junit.jupiter.api.Test;
/**
* @author Mark Pollack

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitMessagingTemplateTests.java
View File

@@ -31,8 +31,8 @@ import java.io.Writer;
import java.util.HashMap;
import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
@@ -66,7 +66,7 @@ public class RabbitMessagingTemplateTests {
private RabbitMessagingTemplate messagingTemplate;
@Before
@BeforeEach
public void setup() {
MockitoAnnotations.initMocks(this);
messagingTemplate = new RabbitMessagingTemplate(rabbitTemplate);

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateHeaderTests.java
View File

@@ -29,7 +29,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.amqp.core.Message;

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java
View File

@@ -45,7 +45,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.amqp.AmqpAuthenticationException;

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/support/SimpleBatchStrategyTests.java
View File

@@ -18,8 +18,8 @@ package org.springframework.amqp.rabbit.core.support;
import java.nio.ByteBuffer;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.util.StopWatch;
@@ -31,7 +31,7 @@ import org.springframework.util.StopWatch;
public class SimpleBatchStrategyTests {
@Test
@Ignore
@Disabled
public void testBatchingPerf() { // used to compare ByteBuffer Vs. System.arrayCopy()
StopWatch watch = new StopWatch();
byte[] bbBuff = new byte[10000];

31
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerIntegrationTests.java
View File

@@ -25,18 +25,15 @@ import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.logging.log4j.Level;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LogLevelAdjuster;
import org.springframework.amqp.rabbit.junit.LogLevels;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
import org.springframework.amqp.rabbit.support.ActiveObjectCounter;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
@@ -48,24 +45,20 @@ import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter
* @since 1.0
*
*/
@RabbitAvailable(queues = { BlockingQueueConsumerIntegrationTests.QUEUE1_NAME,
BlockingQueueConsumerIntegrationTests.QUEUE2_NAME })
@LogLevels(classes = {RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class,
BlockingQueueConsumerIntegrationTests.class }, level = "INFO")
public class BlockingQueueConsumerIntegrationTests {
private static Queue queue1 = new Queue("test.queue1");
public static final String QUEUE1_NAME = "test.queue1.BlockingQueueConsumerIntegrationTests";
private static Queue queue2 = new Queue("test.queue2");
public static final String QUEUE2_NAME = "test.queue2.BlockingQueueConsumerIntegrationTests";
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue1.getName(), queue2.getName());
private static Queue queue1 = new Queue(QUEUE1_NAME);
@Rule
public LogLevelAdjuster logLevels = new LogLevelAdjuster(Level.INFO, RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class,
BlockingQueueConsumerIntegrationTests.class);
@After
public void tearDown() {
this.brokerIsRunning.removeTestQueues();
}
private static Queue queue2 = new Queue(QUEUE2_NAME);
@Test
public void testTransactionalLowLevel() throws Exception {

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumerTests.java
View File

@@ -45,9 +45,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.logging.log4j.Level;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
@@ -55,7 +53,7 @@ import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.ChannelProxy;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.junit.LogLevelAdjuster;
import org.springframework.amqp.rabbit.junit.LogLevels;
import org.springframework.amqp.rabbit.support.ActiveObjectCounter;
import org.springframework.amqp.rabbit.support.ConsumerCancelledException;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
@@ -78,11 +76,9 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringChannel;
* @since 1.0.1
*
*/
@LogLevels(classes = BlockingQueueConsumer.class, level = "ERROR")
public class BlockingQueueConsumerTests {
@Rule
public LogLevelAdjuster adjuster = new LogLevelAdjuster(Level.ERROR, BlockingQueueConsumer.class);
@Test
public void testRequeue() throws Exception {
Exception ex = new RuntimeException();

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java
View File

@@ -22,7 +22,6 @@ import static org.assertj.core.api.Assertions.fail;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Message;
@@ -32,9 +31,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.connection.ShutDownChannelListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
import org.springframework.context.ApplicationContext;
@@ -55,13 +52,6 @@ public class ContainerInitializationTests {
public static final String TEST_MISMATCH2 = "test.mismatch2";
public BrokerRunning brokerRunning = RabbitAvailableCondition.getBrokerRunning();
@AfterEach
public void tearDown() {
brokerRunning.purgeTestQueues();
}
@Test
public void testNoAdmin() {
try {

79
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java
View File

@@ -37,12 +37,11 @@ import java.util.concurrent.atomic.AtomicReference;
import org.aopalliance.intercept.MethodInterceptor;
import org.apache.commons.logging.LogFactory;
import org.apache.logging.log4j.Level;
import org.junit.AfterClass;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.mockito.ArgumentCaptor;
import org.springframework.amqp.core.Queue;
@@ -52,7 +51,10 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.LogLevelAdjuster;
import org.springframework.amqp.rabbit.junit.BrokerRunningSupport;
import org.springframework.amqp.rabbit.junit.LogLevels;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener;
@@ -79,41 +81,48 @@ import com.rabbitmq.client.Consumer;
* @since 2.0
*
*/
@RabbitAvailable(queues = { DirectMessageListenerContainerIntegrationTests.Q1,
DirectMessageListenerContainerIntegrationTests.Q2,
DirectMessageListenerContainerIntegrationTests.EQ1,
DirectMessageListenerContainerIntegrationTests.EQ2,
DirectMessageListenerContainerIntegrationTests.DLQ1 })
@LogLevels(classes = { CachingConnectionFactory.class, DirectReplyToMessageListenerContainer.class,
DirectMessageListenerContainer.class, DirectMessageListenerContainerIntegrationTests.class,
BrokerRunning.class }, level = "DEBUG")
public class DirectMessageListenerContainerIntegrationTests {
private static final String Q1 = "testQ1";
public static final String Q1 = "testQ1.DirectMessageListenerContainerIntegrationTests";
private static final String Q2 = "testQ2";
public static final String Q2 = "testQ2.DirectMessageListenerContainerIntegrationTests";
private static final String EQ1 = "eventTestQ1";
public static final String EQ1 = "eventTestQ1.DirectMessageListenerContainerIntegrationTests";
private static final String EQ2 = "eventTestQ2";
public static final String EQ2 = "eventTestQ2.DirectMessageListenerContainerIntegrationTests";
private static final String DLQ1 = "testDLQ1";
public static final String DLQ1 = "testDLQ1.DirectMessageListenerContainerIntegrationTests";
@ClassRule
public static BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues(Q1, Q2, EQ1, EQ2, DLQ1);
private static CachingConnectionFactory adminCf;
private static CachingConnectionFactory adminCf =
new CachingConnectionFactory(brokerRunning.getConnectionFactory());
private static RabbitAdmin admin;
private static RabbitAdmin admin = new RabbitAdmin(adminCf);
private String testName;
@Rule
public LogLevelAdjuster adjuster = new LogLevelAdjuster(Level.DEBUG,
CachingConnectionFactory.class, DirectReplyToMessageListenerContainer.class,
DirectMessageListenerContainer.class, DirectMessageListenerContainerIntegrationTests.class,
BrokerRunning.class);
@BeforeAll
public static void setUp() {
adminCf = new CachingConnectionFactory(RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
admin = new RabbitAdmin(adminCf);
}
@Rule
public TestName testName = new TestName();
@AfterClass
@AfterAll
public static void tearDown() {
brokerRunning.removeTestQueues();
adminCf.destroy();
}
@BeforeEach
public void captureTestName(TestInfo info) {
this.testName = info.getDisplayName();
}
@SuppressWarnings("unchecked")
@Test
public void testSimple() throws Exception {
@@ -342,7 +351,7 @@ public class DirectMessageListenerContainerIntegrationTests {
}
@Test
public void testEvents() throws Exception {
public void testEvents(BrokerRunningSupport brokerRunning) throws Exception {
CachingConnectionFactory cf = new CachingConnectionFactory("localhost");
DirectMessageListenerContainer container = new DirectMessageListenerContainer(cf);
container.setQueueNames(EQ1, EQ2);
@@ -374,7 +383,7 @@ public class DirectMessageListenerContainerIntegrationTests {
@SuppressWarnings("unchecked")
@Test
public void testErrorHandler() throws Exception {
public void testErrorHandler(BrokerRunningSupport brokerRunning) throws Exception {
brokerRunning.deleteQueues(Q1);
Queue q1 = new Queue(Q1, true, false, false, new ArgumentBuilder()
.put("x-dead-letter-exchange", "")
@@ -503,17 +512,17 @@ public class DirectMessageListenerContainerIntegrationTests {
}
@Test
public void testRecoverDeletedQueueAutoDeclare() throws Exception {
testRecoverDeletedQueueGuts(true);
public void testRecoverDeletedQueueAutoDeclare(BrokerRunningSupport brokerRunning) throws Exception {
testRecoverDeletedQueueGuts(true, brokerRunning);
}
@Test
public void testRecoverDeletedQueueNoAutoDeclare() throws Exception {
testRecoverDeletedQueueGuts(false);
public void testRecoverDeletedQueueNoAutoDeclare(BrokerRunningSupport brokerRunning) throws Exception {
testRecoverDeletedQueueGuts(false, brokerRunning);
}
@SuppressWarnings("unchecked")
private void testRecoverDeletedQueueGuts(boolean autoDeclare) throws Exception {
private void testRecoverDeletedQueueGuts(boolean autoDeclare, BrokerRunningSupport brokerRunning) throws Exception {
CachingConnectionFactory cf = new CachingConnectionFactory("localhost");
DirectMessageListenerContainer container = new DirectMessageListenerContainer(cf);
if (autoDeclare) {
@@ -719,7 +728,7 @@ public class DirectMessageListenerContainerIntegrationTests {
@Override
public String createConsumerTag(String queue) {
return queue + "/" + DirectMessageListenerContainerIntegrationTests.this.testName.getMethodName() + n++;
return queue + "/" + DirectMessageListenerContainerIntegrationTests.this.testName + n++;
}
}

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerMockTests.java
View File

@@ -38,7 +38,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.amqp.core.AcknowledgeMode;

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.java
View File

@@ -22,14 +22,11 @@ import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Address;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
import org.springframework.amqp.rabbit.listener.DirectReplyToMessageListenerContainer.ChannelHolder;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.utils.test.TestUtils;
@@ -51,13 +48,6 @@ public class DirectReplyToMessageListenerContainerTests {
public static final String TEST_RELEASE_CONSUMER_Q = "test.release.consumer";
public BrokerRunning brokerRunning = RabbitAvailableCondition.getBrokerRunning();
@AfterEach
public void tearDown() {
this.brokerRunning.purgeTestQueues();
}
@Test
public void testReleaseConsumerRace() throws Exception {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ErrorHandlerTests.java
View File

@@ -24,7 +24,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.MessageProperties;

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ExternalTxManagerTests.java
View File

@@ -37,7 +37,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.mockito.stubbing.Answer;

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/LocallyTransactedTests.java
View File

@@ -36,7 +36,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java
View File

@@ -34,7 +34,6 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -52,10 +51,8 @@ import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
@@ -90,8 +87,6 @@ public class MessageListenerContainerErrorHandlerIntegrationTests {
private volatile CountDownLatch errorsHandled;
public BrokerRunning brokerIsRunning = RabbitAvailableCondition.getBrokerRunning();
@BeforeEach
public void setUp() {
doAnswer(invocation -> {
@@ -100,11 +95,6 @@ public class MessageListenerContainerErrorHandlerIntegrationTests {
}).when(errorHandler).handleError(any(Throwable.class));
}
@AfterEach
public void tearDown() {
this.brokerIsRunning.purgeTestQueues();
}
@Test // AMQP-385
public void testErrorHandlerThrowsARADRE() throws Exception {
RabbitTemplate template = this.createTemplate(1);

82
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java
View File

@@ -17,11 +17,12 @@
package org.springframework.amqp.rabbit.listener;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
import java.net.UnknownHostException;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
@@ -31,11 +32,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.logging.log4j.Level;
import org.junit.After;
import org.junit.Assume;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.amqp.AmqpIllegalStateException;
@@ -45,10 +42,10 @@ import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LogLevelAdjuster;
import org.springframework.amqp.rabbit.junit.LongRunningIntegrationTest;
import org.springframework.amqp.rabbit.junit.LogLevels;
import org.springframework.amqp.rabbit.junit.LongRunning;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
import org.springframework.amqp.rabbit.support.ActiveObjectCounter;
@@ -59,6 +56,7 @@ import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.junit.jupiter.DisabledIf;
import com.rabbitmq.client.DnsRecordIpAddressResolver;
@@ -70,11 +68,18 @@ import com.rabbitmq.client.DnsRecordIpAddressResolver;
* @since 1.0
*
*/
@RabbitAvailable(queues = MessageListenerContainerLifecycleIntegrationTests.TEST_QUEUE)
@LongRunning
@LogLevels(classes = { RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class,
MessageListenerContainerLifecycleIntegrationTests.class }, level = "INFO")
public class MessageListenerContainerLifecycleIntegrationTests {
public static final String TEST_QUEUE = "test.queue.MessageListenerContainerLifecycleIntegrationTests";
private static Log logger = LogFactory.getLog(MessageListenerContainerLifecycleIntegrationTests.class);
private static Queue queue = new Queue("test.queue");
private static Queue queue = new Queue(TEST_QUEUE);
private enum TransactionMode {
ON, OFF, PREFETCH, PREFETCH_NO_TX;
@@ -121,17 +126,6 @@ public class MessageListenerContainerLifecycleIntegrationTests {
}
}
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue.getName());
@Rule
public LogLevelAdjuster logLevels = new LogLevelAdjuster(Level.INFO, RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class,
MessageListenerContainerLifecycleIntegrationTests.class);
private RabbitTemplate createTemplate(int concurrentConsumers) {
RabbitTemplate template = new RabbitTemplate();
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
@@ -142,11 +136,6 @@ public class MessageListenerContainerLifecycleIntegrationTests {
return template;
}
@After
public void tearDown() {
this.brokerIsRunning.removeTestQueues();
}
@Test
public void testTransactionalLowLevel() throws Exception {
doTest(MessageCount.MEDIUM, Concurrency.LOW, TransactionMode.ON);
@@ -187,34 +176,31 @@ public class MessageListenerContainerLifecycleIntegrationTests {
doTest(MessageCount.HIGH, Concurrency.HIGH, TransactionMode.PREFETCH_NO_TX);
}
@Test
public void testBadCredentials() throws Exception {
/**
* If localhost also resolves to an IPv6 address the client will try that
* after a failure due to bad credentials and, if Rabbit is not listening there
* we won't get a fatal startup exception because a connect exception is not
* considered fatal.
* @throws UnknownHostException unknown host
*/
public static boolean checkIpV6() throws UnknownHostException {
DnsRecordIpAddressResolver resolver = new DnsRecordIpAddressResolver("localhost");
if (resolver.getAddresses().size() > 1) {
/*
* If localhost also resolves to an IPv6 address the client will try that
* after a failure due to bad credentials and, if Rabbit is not listening there
* we won't get a fatal startup exception because a connect exception is not
* considered fatal.
*/
Assume.assumeNoException(
new RuntimeException("Resolver returned multiple addresses for localhost, ignoring test"));
}
return resolver.getAddresses().size() > 1;
}
@Test
@DisabledIf("#{T(org.springframework.amqp.rabbit.listener.MessageListenerContainerLifecycleIntegrationTests)"
+ ".checkIpV6()}")
public void testBadCredentials() throws Exception {
RabbitTemplate template = createTemplate(1);
com.rabbitmq.client.ConnectionFactory cf = new com.rabbitmq.client.ConnectionFactory();
cf.setAutomaticRecoveryEnabled(false);
cf.setUsername("foo");
final CachingConnectionFactory connectionFactory = new CachingConnectionFactory(cf);
try {
doTest(MessageCount.LOW, Concurrency.LOW, TransactionMode.OFF, template, connectionFactory);
fail("expected exception");
}
catch (AmqpIllegalStateException e) {
assertThat(e.getCause() instanceof FatalListenerStartupException).as("Expected FatalListenerStartupException").isTrue();
}
finally {
((DisposableBean) template.getConnectionFactory()).destroy();
}
assertThatExceptionOfType(AmqpIllegalStateException.class).isThrownBy(() ->
doTest(MessageCount.LOW, Concurrency.LOW, TransactionMode.OFF, template, connectionFactory))
.withCauseExactlyInstanceOf(FatalListenerStartupException.class);
((DisposableBean) template.getConnectionFactory()).destroy();
}
private void doTest(MessageCount level, Concurrency concurrency, TransactionMode transactionMode) throws Exception {

33
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerMultipleQueueIntegrationTests.java
View File

@@ -24,18 +24,15 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.logging.log4j.Level;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LogLevelAdjuster;
import org.springframework.amqp.rabbit.junit.LogLevels;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
@@ -44,25 +41,21 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
* @author Gunnar Hillert
* @author Gary Russell
*/
@RabbitAvailable(queues = { MessageListenerContainerMultipleQueueIntegrationTests.TEST_QUEUE_1,
MessageListenerContainerMultipleQueueIntegrationTests.TEST_QUEUE_2 })
@LogLevels(level = "INFO", classes = { RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class })
public class MessageListenerContainerMultipleQueueIntegrationTests {
public static final String TEST_QUEUE_1 = "test.queue.1.MessageListenerContainerMultipleQueueIntegrationTests";
public static final String TEST_QUEUE_2 = "test.queue.2.MessageListenerContainerMultipleQueueIntegrationTests";
private static Log logger = LogFactory.getLog(MessageListenerContainerMultipleQueueIntegrationTests.class);
private static Queue queue1 = new Queue("test.queue.1");
private static Queue queue1 = new Queue(TEST_QUEUE_1);
private static Queue queue2 = new Queue("test.queue.2");
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue1.getName(), queue2.getName());
@Rule
public LogLevelAdjuster logLevels = new LogLevelAdjuster(Level.INFO, RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class);
@After
public void tearDown() {
this.brokerIsRunning.removeTestQueues();
}
private static Queue queue2 = new Queue(TEST_QUEUE_2);
@Test
public void testMultipleQueues() {

31
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java
View File

@@ -23,20 +23,18 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.logging.log4j.Level;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LogLevelAdjuster;
import org.springframework.amqp.rabbit.junit.LogLevels;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.DisposableBean;
@@ -51,11 +49,16 @@ import com.rabbitmq.client.Channel;
* @since 1.0
*
*/
@RabbitAvailable(queues = MessageListenerManualAckIntegrationTests.TEST_QUEUE)
@LogLevels(level = "ERROR", classes = { RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class })
public class MessageListenerManualAckIntegrationTests {
public static final String TEST_QUEUE = "test.queue.MessageListenerManualAckIntegrationTests";
private static Log logger = LogFactory.getLog(MessageListenerManualAckIntegrationTests.class);
private final Queue queue = new Queue("test.queue");
private final Queue queue = new Queue(TEST_QUEUE);
private final RabbitTemplate template = new RabbitTemplate();
@@ -69,14 +72,7 @@ public class MessageListenerManualAckIntegrationTests {
private SimpleMessageListenerContainer container;
@Rule
public LogLevelAdjuster logLevels = new LogLevelAdjuster(Level.ERROR, RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class);
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue.getName());
@Before
@BeforeEach
public void createConnectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
@@ -85,7 +81,7 @@ public class MessageListenerManualAckIntegrationTests {
template.setConnectionFactory(connectionFactory);
}
@After
@AfterEach
public void clear() throws Exception {
// Wait for broker communication to finish before trying to stop container
Thread.sleep(300L);
@@ -93,7 +89,6 @@ public class MessageListenerManualAckIntegrationTests {
if (container != null) {
container.shutdown();
}
this.brokerIsRunning.removeTestQueues();
((DisposableBean) template.getConnectionFactory()).destroy();
}

62
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.amqp.rabbit.listener;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import java.util.Collections;
import java.util.HashSet;
@@ -28,10 +29,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.logging.log4j.Level;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.AmqpIllegalStateException;
import org.springframework.amqp.core.AcknowledgeMode;
@@ -43,10 +42,11 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionProxy;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LogLevelAdjuster;
import org.springframework.amqp.rabbit.junit.LongRunningIntegrationTest;
import org.springframework.amqp.rabbit.junit.LogLevels;
import org.springframework.amqp.rabbit.junit.LongRunning;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.listener.MessageListenerRecoveryCachingConnectionIntegrationTests.ManualAckListener;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.utils.test.TestUtils;
@@ -65,13 +65,22 @@ import com.rabbitmq.client.Channel;
* @since 1.0
*
*/
@RabbitAvailable(queues = { MessageListenerRecoveryCachingConnectionIntegrationTests.TEST_QUEUE,
MessageListenerRecoveryCachingConnectionIntegrationTests.TEST_SEND })
@LongRunning
@LogLevels(level = "DEBUG", classes = { RabbitTemplate.class, ManualAckListener.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class, CachingConnectionFactory.class })
public class MessageListenerRecoveryCachingConnectionIntegrationTests {
public static final String TEST_QUEUE = "test.queue.MessageListenerRecoveryCachingConnectionIntegrationTests";
public static final String TEST_SEND = "test.send.MessageListenerRecoveryCachingConnectionIntegrationTests";
private static Log logger = LogFactory.getLog(MessageListenerRecoveryCachingConnectionIntegrationTests.class);
private final Queue queue = new Queue("test.queue");
private final Queue queue = new Queue(TEST_QUEUE);
private final Queue sendQueue = new Queue("test.send");
private final Queue sendQueue = new Queue(TEST_SEND);
private int concurrentConsumers = 1;
@@ -83,16 +92,6 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
private SimpleMessageListenerContainer container;
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@Rule
public LogLevelAdjuster logLevels = new LogLevelAdjuster(Level.DEBUG, RabbitTemplate.class, ManualAckListener.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class, CachingConnectionFactory.class);
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue.getName(), sendQueue.getName());
protected CachingConnectionFactory createConnectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
@@ -101,7 +100,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
return connectionFactory;
}
@After
@AfterEach
public void clear() throws Exception {
// Wait for broker communication to finish before trying to stop container
Thread.sleep(300L);
@@ -109,7 +108,6 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
if (container != null) {
container.shutdown();
}
this.brokerIsRunning.removeTestQueues();
}
@Test
@@ -322,7 +320,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
}
@Test(expected = AmqpIllegalStateException.class)
@Test
public void testListenerDoesNotRecoverFromMissingQueue() throws Exception {
concurrentConsumers = 3;
@@ -332,15 +330,12 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
RabbitAdmin admin = new RabbitAdmin(connectionFactory);
admin.deleteQueue("nonexistent");
try {
container = createContainer("nonexistent", new VanillaListener(latch), connectionFactory);
}
finally {
((DisposableBean) connectionFactory).destroy();
}
assertThatExceptionOfType(AmqpIllegalStateException.class).isThrownBy(() ->
container = createContainer("nonexistent", new VanillaListener(latch), connectionFactory));
((DisposableBean) connectionFactory).destroy();
}
@Test(expected = AmqpIllegalStateException.class)
@Test
public void testSingleListenerDoesNotRecoverFromMissingQueue() throws Exception {
/*
* A single listener sometimes doesn't have time to attempt to start before we ask it if it has failed, so this
@@ -352,12 +347,9 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
ConnectionFactory connectionFactory = createConnectionFactory();
RabbitAdmin admin = new RabbitAdmin(connectionFactory);
admin.deleteQueue("nonexistent");
try {
container = createContainer("nonexistent", new VanillaListener(latch), connectionFactory);
}
finally {
((DisposableBean) connectionFactory).destroy();
}
assertThatExceptionOfType(AmqpIllegalStateException.class).isThrownBy(() ->
container = createContainer("nonexistent", new VanillaListener(latch), connectionFactory));
((DisposableBean) connectionFactory).destroy();
}
@Test

31
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerTxSizeIntegrationTests.java
View File

@@ -23,20 +23,18 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.logging.log4j.Level;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LogLevelAdjuster;
import org.springframework.amqp.rabbit.junit.LogLevels;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.DisposableBean;
@@ -51,11 +49,16 @@ import com.rabbitmq.client.Channel;
* @since 1.0
*
*/
@RabbitAvailable(queues = MessageListenerTxSizeIntegrationTests.TEST_QUEUE)
@LogLevels(level = "ERROR", classes = { RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class })
public class MessageListenerTxSizeIntegrationTests {
public static final String TEST_QUEUE = "test.queue.MessageListenerTxSizeIntegrationTests";
private static Log logger = LogFactory.getLog(MessageListenerTxSizeIntegrationTests.class);
private final Queue queue = new Queue("test.queue");
private final Queue queue = new Queue(TEST_QUEUE);
private final RabbitTemplate template = new RabbitTemplate();
@@ -69,14 +72,7 @@ public class MessageListenerTxSizeIntegrationTests {
private SimpleMessageListenerContainer container;
@Rule
public LogLevelAdjuster logLevels = new LogLevelAdjuster(Level.ERROR, RabbitTemplate.class,
SimpleMessageListenerContainer.class, BlockingQueueConsumer.class);
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue.getName());
@Before
@BeforeEach
public void createConnectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
@@ -85,7 +81,7 @@ public class MessageListenerTxSizeIntegrationTests {
template.setConnectionFactory(connectionFactory);
}
@After
@AfterEach
public void clear() throws Exception {
// Wait for broker communication to finish before trying to stop container
Thread.sleep(300L);
@@ -95,7 +91,6 @@ public class MessageListenerTxSizeIntegrationTests {
}
((DisposableBean) template.getConnectionFactory()).destroy();
this.brokerIsRunning.removeTestQueues();
}
@Test

32
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpointTests.java
View File

@@ -31,10 +31,9 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.mockito.ArgumentCaptor;
import org.springframework.amqp.core.Address;
@@ -72,36 +71,35 @@ import com.rabbitmq.client.Channel;
*/
public class MethodRabbitListenerEndpointTests {
@Rule
public final TestName name = new TestName();
private final DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
private final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
private final RabbitEndpointSampleBean sample = new RabbitEndpointSampleBean();
public String testName;
@Before
public void setup() {
@BeforeEach
public void setup(TestInfo info) {
initializeFactory(factory);
this.testName = info.getTestMethod().get().getName();
}
@Test
public void createMessageListenerNoFactory() {
public void createMessageListenerNoFactory(TestInfo info) {
MethodRabbitListenerEndpoint endpoint = new MethodRabbitListenerEndpoint();
endpoint.setBean(this);
endpoint.setMethod(getTestMethod());
endpoint.setMethod(info.getTestMethod().get());
assertThatIllegalStateException()
.isThrownBy(() -> endpoint.createMessageListener(container));
}
@Test
public void createMessageListener() {
public void createMessageListener(TestInfo info) {
MethodRabbitListenerEndpoint endpoint = new MethodRabbitListenerEndpoint();
endpoint.setBean(this);
endpoint.setMethod(getTestMethod());
endpoint.setMethod(info.getTestMethod().get());
endpoint.setMessageHandlerMethodFactory(factory);
assertThat(endpoint.createMessageListener(container)).isNotNull();
@@ -418,11 +416,11 @@ public class MethodRabbitListenerEndpointTests {
}
private Method getDefaultListenerMethod(Class<?>... parameterTypes) {
return getListenerMethod(name.getMethodName(), parameterTypes);
return getListenerMethod(this.testName, parameterTypes);
}
private void assertDefaultListenerMethodInvocation() {
assertListenerMethodInvocation(sample, name.getMethodName());
assertListenerMethodInvocation(this.sample, this.testName);
}
private void assertListenerMethodInvocation(RabbitEndpointSampleBean bean, String methodName) {
@@ -451,10 +449,6 @@ public class MethodRabbitListenerEndpointTests {
};
}
private Method getTestMethod() {
return ReflectionUtils.findMethod(MethodRabbitListenerEndpointTests.class, name.getMethodName());
}
static class RabbitEndpointSampleBean {
private final Map<String, Boolean> invocations = new HashMap<String, Boolean>();

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistrarTests.java
View File

@@ -20,8 +20,8 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.config.RabbitListenerContainerTestFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint;
@@ -41,7 +41,7 @@ public class RabbitListenerEndpointRegistrarTests {
private final RabbitListenerContainerTestFactory containerFactory = new RabbitListenerContainerTestFactory();
@Before
@BeforeEach
public void setup() {
registrar.setEndpointRegistry(registry);
registrar.setBeanFactory(new StaticListableBeanFactory());

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpointRegistryTests.java
View File

@@ -19,7 +19,7 @@ package org.springframework.amqp.rabbit.listener;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.config.RabbitListenerContainerTestFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint;

38
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java
View File

@@ -41,10 +41,9 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.amqp.AmqpIOException;
@@ -61,9 +60,9 @@ import org.springframework.amqp.rabbit.connection.PublisherCallbackChannelImpl;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.junit.LongRunningIntegrationTest;
import org.springframework.amqp.rabbit.junit.LongRunning;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
@@ -88,50 +87,45 @@ import com.rabbitmq.client.Channel;
* @since 1.3
*
*/
@RabbitAvailable(queues = { SimpleMessageListenerContainerIntegration2Tests.TEST_QUEUE,
SimpleMessageListenerContainerIntegration2Tests.TEST_QUEUE_1 })
@LongRunning
public class SimpleMessageListenerContainerIntegration2Tests {
public static final String TEST_QUEUE = "test.queue.SimpleMessageListenerContainerIntegration2Tests";
public static final String TEST_QUEUE_1 = "test.queue.1.SimpleMessageListenerContainerIntegration2Tests";
private static Log logger = LogFactory.getLog(SimpleMessageListenerContainerIntegration2Tests.class);
private final ExecutorService executorService = Executors.newSingleThreadExecutor();
private final Queue queue = new Queue("test.queue");
private final Queue queue = new Queue(TEST_QUEUE);
private final Queue queue1 = new Queue("test.queue.1");
private final Queue queue1 = new Queue(TEST_QUEUE_1);
private final RabbitTemplate template = new RabbitTemplate();
private RabbitAdmin admin;
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue.getName(), queue1.getName());
@Rule
public LongRunningIntegrationTest longRunningIntegrationTest = new LongRunningIntegrationTest();
private SimpleMessageListenerContainer container;
@Before
@BeforeEach
public void declareQueues() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
connectionFactory.setPort(BrokerTestUtils.getPort());
template.setConnectionFactory(connectionFactory);
admin = new RabbitAdmin(connectionFactory);
admin.deleteQueue(queue.getName());
admin.declareQueue(queue);
admin.deleteQueue(queue1.getName());
admin.declareQueue(queue1);
}
@After
@AfterEach
public void clear() throws Exception {
logger.debug("Shutting down at end of test");
if (container != null) {
container.shutdown();
}
((DisposableBean) template.getConnectionFactory()).destroy();
this.brokerIsRunning.removeTestQueues();
this.executorService.shutdown();
}

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java
View File

@@ -55,7 +55,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.AmqpAuthenticationException;

14
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java
View File

@@ -23,10 +23,10 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.junit.BrokerTestUtils;
import org.springframework.amqp.rabbit.support.Delivery;
@@ -45,7 +45,7 @@ import com.rabbitmq.client.GetResponse;
* @author Dave Syer
*
*/
@Ignore
@Disabled
public class UnackedRawIntegrationTests {
private final ConnectionFactory factory = new ConnectionFactory();
@@ -53,7 +53,7 @@ public class UnackedRawIntegrationTests {
private Channel noTxChannel;
private Channel txChannel;
@Before
@BeforeEach
public void init() throws Exception {
factory.setHost("localhost");
@@ -74,7 +74,7 @@ public class UnackedRawIntegrationTests {
}
@After
@AfterEach
public void clear() throws Exception {
if (txChannel != null) {
try {

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java
View File

@@ -30,8 +30,8 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Address;
@@ -65,7 +65,7 @@ public class MessageListenerAdapterTests {
private final SimpleService simpleService = new SimpleService();
@Before
@BeforeEach
public void init() {
this.messageProperties = new MessageProperties();
this.messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java
View File

@@ -28,8 +28,8 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
@@ -59,7 +59,7 @@ public class MessagingMessageListenerAdapterTests {
private final SampleBean sample = new SampleBean();
@Before
@BeforeEach
public void setup() {
initializeFactory(factory);
}

15
src/reference/asciidoc/testing.adoc
View File

@@ -522,10 +522,11 @@ Version 2.0.2 introduced support for JUnit5.
This class-level annotation is similar to the `BrokerRunning` `@Rule` discussed in <<junit-rules>>.
It is processed by the `RabbitAvailableCondition`.
The annotation has two properties:
The annotation has three properties:
* `queues`: An array of queues that are declared (and purged) before each test and deleted when all tests are complete.
* `management`: Set this to `true` if your tests also require the management plugin installed on the broker.
* `purgeAfterEach`: (Since version 2.2) when `true` (default), the `queues` will be purged between tests.
It is used to check whether the broker is available and skip the tests if not.
As discussed in <<brokerRunning-configure>>, the environment variable called `RABBITMQ_SERVER_REQUIRED`, if `true`, causes the tests to fail fast if there is no broker.
@@ -534,8 +535,8 @@ You can configure the condition by using environment variables as discussed in <
In addition, the `RabbitAvailableCondition` supports argument resolution for parameterized test constructors and methods.
Two argument types are supported:
* `BrokerRunning`: The instance
* `ConnectionFactory`: The `BrokerRunning` instance's RabbitMQ connection factory
* `BrokerRunningSupport`: The instance (before 2.2, this was a JUnit 4 `BrokerRunning` instance)
* `ConnectionFactory`: The `BrokerRunningSupport` instance's RabbitMQ connection factory
The following example shows both:
@@ -547,7 +548,7 @@ public class RabbitAvailableCTORInjectionTests {
private final ConnectionFactory connectionFactory;
public RabbitAvailableCTORInjectionTests(BrokerRunning brokerRunning) {
public RabbitAvailableCTORInjectionTests(BrokerRunningSupport brokerRunning) {
this.connectionFactory = brokerRunning.getConnectionFactory();
}
@@ -578,7 +579,7 @@ public class RabbitAvailableCTORInjectionTests {
private final CachingConnectionFactory connectionFactory;
public RabbitAvailableCTORInjectionTests(BrokerRunning brokerRunning) {
public RabbitAvailableCTORInjectionTests(BrokerRunningSupport brokerRunning) {
this.connectionFactory =
new CachingConnectionFactory(brokerRunning.getConnectionFactory());
}
@@ -593,6 +594,10 @@ public class RabbitAvailableCTORInjectionTests {
====
When you use a Spring annotation application context within a test class, you can get a reference to the condition's connection factory through a static method called `RabbitAvailableCondition.getBrokerRunning()`.
IMPORTANT: Starting with version 2.2, `getBrokerRunning()` returns a `BrokerRunningSupport` object; previously, the JUnit 4 `BrokerRunnning` instance was returned.
The new class has the same API as `BrokerRunning`.
The following test comes from the framework and demonstrates the usage:
====

6
src/reference/asciidoc/whats-new.adoc
View File

@@ -19,6 +19,12 @@ In addition, `ListenerExecutionFailedException` has been moved from `org.springf
JUnit (4) is now an optional dependency and will no longer appear as a transitive dependency.
===== "Breaking" API Changes
the JUnit (5) `RabbitAvailableCondition.getBrokerRunning()` now returns a `BrokerRunningSupport` instance instead of a `BrokerRunning`, which depends on JUnit 4.
It has the same API so it's just a matter of changing the class name of any references.
See <<junit5-conditions>> for more information.
===== ListenerContainer Changes
Messages with fatal exceptions are now rejected and NOT requeued, by default, even if the acknowledge mode is manual.