Skip to content

Commit

Permalink
Improve some code in the AbstractDeclarable
Browse files Browse the repository at this point in the history 
* Add `Declarable.setAdminsThatShouldDeclare(@nullable)`
* Improve `RabbitAdminDeclarationTests`
* Fix typo in the `AbstractMessageListenerContainer` JavaDoc
  • Loading branch information
@artembilan
artembilan committed Oct 15, 2024
1 parent ef23aa9 commit 347c96a
Show file tree
Hide file tree
Showing 4 changed files with 51 additions and 58 deletions.
15 changes: 7 additions & 8 deletions spring-amqp/src/main/java/org/springframework/amqp/core/AbstractDeclarable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


import org.springframework.lang.Nullable;
import org.springframework.util.Assert;

Expand All @@ -40,15 +39,15 @@
*/
public abstract class AbstractDeclarable implements Declarable {

private final Lock lock = new ReentrantLock();
private final Lock lock = new ReentrantLock();

private boolean shouldDeclare = true;
private final Map<String, Object> arguments;

private Collection<Object> declaringAdmins = new ArrayList<>();
private boolean shouldDeclare = true;

private boolean ignoreDeclarationExceptions;

private final Map<String, Object> arguments;
private Collection<Object> declaringAdmins = new ArrayList<>();

public AbstractDeclarable() {
this(null);
Expand All @@ -74,7 +73,7 @@ public boolean shouldDeclare() {
}

/**
* Whether or not this object should be automatically declared
* Whether this object should be automatically declared
* by any {@code AmqpAdmin}. Default is {@code true}.
* @param shouldDeclare true or false.
*/
Expand Down Expand Up @@ -102,14 +101,14 @@ public void setIgnoreDeclarationExceptions(boolean ignoreDeclarationExceptions)
}

@Override
public void setAdminsThatShouldDeclare(Object... adminArgs) {
public void setAdminsThatShouldDeclare(@Nullable Object... adminArgs) {
Collection<Object> admins = new ArrayList<>();
if (adminArgs != null) {
if (adminArgs.length > 1) {
Assert.noNullElements(adminArgs, "'admins' cannot contain null elements");
}
if (adminArgs.length > 0 && !(adminArgs.length == 1 && adminArgs[0] == null)) {
admins.addAll(Arrays.asList(adminArgs));
admins = Arrays.asList(adminArgs);
}
}
this.declaringAdmins = admins;
Expand Down
4 changes: 2 additions & 2 deletions spring-amqp/src/main/java/org/springframework/amqp/core/Declarable.java
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2024 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.
Expand Down Expand Up @@ -61,7 +61,7 @@ public interface Declarable {
* the behavior such that all admins will declare the object.
* @param adminArgs The admins.
*/
void setAdminsThatShouldDeclare(Object... adminArgs);
void setAdminsThatShouldDeclare(@Nullable Object... adminArgs);

/**
* Add an argument to the declarable.
Expand Down
2 changes: 1 addition & 1 deletion .../main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1490,7 +1490,7 @@ protected void invokeErrorHandler(Throwable ex) {
// -------------------------------------------------------------------------

/**
* Execute the specified listener, committing or rolling back the transaction afterwards (if necessary).
* Execute the specified listener, committing or rolling back the transaction afterward (if necessary).
* @param channel the Rabbit Channel to operate on
* @param data the received Rabbit Message
* @see #invokeListener
Expand Down
88 changes: 41 additions & 47 deletions ...abbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminDeclarationTests.java
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2024 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.
Expand All @@ -17,7 +17,7 @@
package org.springframework.amqp.rabbit.core;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyMap;
Expand Down Expand Up @@ -79,8 +79,8 @@ public void testUnconditional() throws Exception {
given(cf.createConnection()).willReturn(conn);
given(conn.createChannel(false)).willReturn(channel);
given(channel.queueDeclare("foo", true, false, false, new HashMap<>()))
.willReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
final AtomicReference<ConnectionListener> listener = new AtomicReference<ConnectionListener>();
.willReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
AtomicReference<ConnectionListener> listener = new AtomicReference<>();
willAnswer(invocation -> {
listener.set((ConnectionListener) invocation.getArguments()[0]);
return null;
Expand All @@ -100,15 +100,15 @@ public void testUnconditional() throws Exception {
listener.get().onCreate(conn);

verify(channel).queueDeclare("foo", true, false, false, new HashMap<>());
verify(channel).exchangeDeclare("bar", "direct", true, false, false, new HashMap<String, Object>());
verify(channel).exchangeDeclare("bar", "direct", true, false, false, new HashMap<>());
verify(channel).queueBind("foo", "bar", "foo", new HashMap<>());
}

@Test
public void testNoDeclareWithCachedConnections() throws Exception {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);

final List<Channel> mockChannels = new ArrayList<Channel>();
List<Channel> mockChannels = new ArrayList<>();

AtomicInteger connectionNumber = new AtomicInteger();
willAnswer(invocation -> {
Expand Down Expand Up @@ -153,8 +153,8 @@ public void testUnconditionalWithExplicitFactory() throws Exception {
given(cf.createConnection()).willReturn(conn);
given(conn.createChannel(false)).willReturn(channel);
given(channel.queueDeclare("foo", true, false, false, new HashMap<>()))
.willReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
final AtomicReference<ConnectionListener> listener = new AtomicReference<ConnectionListener>();
.willReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
AtomicReference<ConnectionListener> listener = new AtomicReference<>();
willAnswer(invocation -> {
listener.set(invocation.getArgument(0));
return null;
Expand All @@ -177,7 +177,7 @@ public void testUnconditionalWithExplicitFactory() throws Exception {
listener.get().onCreate(conn);

verify(channel).queueDeclare("foo", true, false, false, new HashMap<>());
verify(channel).exchangeDeclare("bar", "direct", true, false, false, new HashMap<String, Object>());
verify(channel).exchangeDeclare("bar", "direct", true, false, false, new HashMap<>());
verify(channel).queueBind("foo", "bar", "foo", new HashMap<>());
}

Expand All @@ -189,8 +189,9 @@ public void testSkipBecauseDifferentFactory() throws Exception {
Channel channel = mock(Channel.class);
given(cf.createConnection()).willReturn(conn);
given(conn.createChannel(false)).willReturn(channel);
given(channel.queueDeclare("foo", true, false, false, null)).willReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
final AtomicReference<ConnectionListener> listener = new AtomicReference<ConnectionListener>();
given(channel.queueDeclare("foo", true, false, false, null))
.willReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
AtomicReference<ConnectionListener> listener = new AtomicReference<>();
willAnswer(invocation -> {
listener.set(invocation.getArgument(0));
return null;
Expand All @@ -215,20 +216,21 @@ public void testSkipBecauseDifferentFactory() throws Exception {

verify(channel, never()).queueDeclare(eq("foo"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
verify(channel, never())
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
verify(channel, never()).queueBind(eq("foo"), eq("bar"), eq("foo"), any(Map.class));
}

@SuppressWarnings("unchecked")
@Test
public void testSkipBecauseShouldntDeclare() throws Exception {
public void testSkipBecauseShouldNotDeclare() throws Exception {
ConnectionFactory cf = mock(ConnectionFactory.class);
Connection conn = mock(Connection.class);
Channel channel = mock(Channel.class);
given(cf.createConnection()).willReturn(conn);
given(conn.createChannel(false)).willReturn(channel);
given(channel.queueDeclare("foo", true, false, false, null)).willReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
final AtomicReference<ConnectionListener> listener = new AtomicReference<ConnectionListener>();
given(channel.queueDeclare("foo", true, false, false, null))
.willReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
AtomicReference<ConnectionListener> listener = new AtomicReference<>();
willAnswer(invocation -> {
listener.set(invocation.getArgument(0));
return null;
Expand All @@ -252,7 +254,7 @@ public void testSkipBecauseShouldntDeclare() throws Exception {

verify(channel, never()).queueDeclare(eq("foo"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
verify(channel, never())
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class));
verify(channel, never()).queueBind(eq("foo"), eq("bar"), eq("foo"), any(Map.class));
}

Expand All @@ -263,19 +265,17 @@ public void testJavaConfig() throws Exception {
verify(Config.channel1).queueDeclare("foo", true, false, false, new HashMap<>());
verify(Config.channel1, never()).queueDeclare("baz", true, false, false, new HashMap<>());
verify(Config.channel1).queueDeclare("qux", true, false, false, new HashMap<>());
verify(Config.channel1).exchangeDeclare("bar", "direct", true, false, true, new HashMap<String, Object>());
verify(Config.channel1).exchangeDeclare("bar", "direct", true, false, true, new HashMap<>());
verify(Config.channel1).queueBind("foo", "bar", "foo", new HashMap<>());

Config.listener2.onCreate(Config.conn2);
verify(Config.channel2, never())
.queueDeclare(eq("foo"), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
verify(Config.channel1, never()).queueDeclare("baz", true, false, false, new HashMap<>());
verify(Config.channel2).queueDeclare("qux", true, false, false, new HashMap<>());
verify(Config.channel2, never())
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(),
anyBoolean(), anyMap());
anyBoolean(), anyMap());
verify(Config.channel2, never()).queueBind(eq("foo"), eq("bar"), eq("foo"), anyMap());

Config.listener3.onCreate(Config.conn3);
verify(Config.channel3, never())
.queueDeclare(eq("foo"), anyBoolean(), anyBoolean(), anyBoolean(), isNull());
Expand All @@ -286,7 +286,7 @@ public void testJavaConfig() throws Exception {
verify(Config.channel3, never()).queueDeclare("qux", true, false, false, new HashMap<>());
verify(Config.channel3, never())
.exchangeDeclare(eq("bar"), eq("direct"), anyBoolean(), anyBoolean(),
anyBoolean(), anyMap());
anyBoolean(), anyMap());
verify(Config.channel3, never()).queueBind(eq("foo"), eq("bar"), eq("foo"), anyMap());

context.close();
Expand Down Expand Up @@ -316,13 +316,9 @@ public void testAddRemove() {
assertThat(queue.getDeclaringAdmins()).hasSize(2);
queue.setAdminsThatShouldDeclare((Object[]) null);
assertThat(queue.getDeclaringAdmins()).hasSize(0);
try {
queue.setAdminsThatShouldDeclare(null, admin1);
fail("Expected Exception");
}
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains("'admins' cannot contain null elements");
}
assertThatIllegalArgumentException()
.isThrownBy(() -> queue.setAdminsThatShouldDeclare(null, admin1))
.withMessageContaining("'admins' cannot contain null elements");
}

@Test
Expand All @@ -348,17 +344,17 @@ public void testNoOpWhenNothingToDeclare() throws Exception {
@Configuration
public static class Config {

private static Connection conn1 = mock(Connection.class);
private static final Connection conn1 = mock();

private static Connection conn2 = mock(Connection.class);
private static final Connection conn2 = mock();

private static Connection conn3 = mock(Connection.class);
private static final Connection conn3 = mock();

private static Channel channel1 = mock(Channel.class);
private static final Channel channel1 = mock();

private static Channel channel2 = mock(Channel.class);
private static final Channel channel2 = mock();

private static Channel channel3 = mock(Channel.class);
private static final Channel channel3 = mock();

private static ConnectionListener listener1;

Expand All @@ -371,9 +367,9 @@ public ConnectionFactory cf1() throws IOException {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
given(connectionFactory.createConnection()).willReturn(conn1);
given(conn1.createChannel(false)).willReturn(channel1);
willAnswer(inv -> {
return new AMQImpl.Queue.DeclareOk(inv.getArgument(0), 0, 0);
}).given(channel1).queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), any());
willAnswer(inv -> new AMQImpl.Queue.DeclareOk(inv.getArgument(0), 0, 0))
.given(channel1)
.queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), any());
willAnswer(invocation -> {
listener1 = invocation.getArgument(0);
return null;
Expand All @@ -386,9 +382,9 @@ public ConnectionFactory cf2() throws IOException {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
given(connectionFactory.createConnection()).willReturn(conn2);
given(conn2.createChannel(false)).willReturn(channel2);
willAnswer(inv -> {
return new AMQImpl.Queue.DeclareOk(inv.getArgument(0), 0, 0);
}).given(channel2).queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), any());
willAnswer(inv -> new AMQImpl.Queue.DeclareOk(inv.getArgument(0), 0, 0))
.given(channel2)
.queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), any());
willAnswer(invocation -> {
listener2 = invocation.getArgument(0);
return null;
Expand All @@ -401,9 +397,9 @@ public ConnectionFactory cf3() throws IOException {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
given(connectionFactory.createConnection()).willReturn(conn3);
given(conn3.createChannel(false)).willReturn(channel3);
willAnswer(inv -> {
return new AMQImpl.Queue.DeclareOk(inv.getArgument(0), 0, 0);
}).given(channel3).queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), any());
willAnswer(inv -> new AMQImpl.Queue.DeclareOk(inv.getArgument(0), 0, 0))
.given(channel3)
.queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), any());
willAnswer(invocation -> {
listener3 = invocation.getArgument(0);
return null;
Expand All @@ -413,14 +409,12 @@ public ConnectionFactory cf3() throws IOException {

@Bean
public RabbitAdmin admin1() throws IOException {
RabbitAdmin rabbitAdmin = new RabbitAdmin(cf1());
return rabbitAdmin;
return new RabbitAdmin(cf1());
}

@Bean
public RabbitAdmin admin2() throws IOException {
RabbitAdmin rabbitAdmin = new RabbitAdmin(cf2());
return rabbitAdmin;
return new RabbitAdmin(cf2());
}

@Bean
Expand Down

0 comments on commit 347c96a

Please sign in to comment.