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

GH-1443: Pull CCF.resetConnection() to CF

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

**Cherry-pick to `2.4.x`**
This commit is contained in:
Gary Russell
2022-03-29 12:41:45 -04:00
committed by Artem Bilan
parent 8843b1ade5
commit 447fa920d9
6 changed files with 34 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractRoutingConnectionFactory.java
View File

@@ -22,6 +22,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.amqp.AmqpException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -38,7 +39,7 @@ import org.springframework.util.Assert;
* @since 1.3
*/
public abstract class AbstractRoutingConnectionFactory implements ConnectionFactory, RoutingConnectionFactory,
InitializingBean {
InitializingBean, DisposableBean {
private final Map<Object, ConnectionFactory> targetConnectionFactories =
new ConcurrentHashMap<Object, ConnectionFactory>();
@@ -260,4 +261,15 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact
@Nullable
protected abstract Object determineCurrentLookupKey();
@Override
public void destroy() {
resetConnection();
}
@Override
public void resetConnection() {
this.targetConnectionFactories.values().forEach(factory -> factory.resetConnection());
this.defaultTargetConnectionFactory.resetConnection();
}
}

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2021 the original author or authors.
* Copyright 2002-2022 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.
@@ -882,6 +882,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
* used to force a reconnect to the primary broker after failing over to a secondary
* broker.
*/
@Override
public void resetConnection() {
synchronized (this.connectionMonitor) {
if (this.connection.target != null) {

10
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2022 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.
@@ -83,4 +83,12 @@ public interface ConnectionFactory {
return false;
}
/**
* Close any connection(s) that might be cached by this factory. This does not prevent
* new connections from being opened.
* @since 2.4.4
*/
default void resetConnection() {
}
}

9
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2019 the original author or authors.
* Copyright 2015-2022 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.
@@ -350,7 +350,7 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
}
@Override
public void destroy() {
public void resetConnection() {
Exception lastException = null;
for (ConnectionFactory connectionFactory : this.nodeFactories.values()) {
if (connectionFactory instanceof DisposableBean) {
@@ -367,4 +367,9 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
}
}
@Override
public void destroy() {
resetConnection();
}
}

1
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactory.java
View File

@@ -149,6 +149,7 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory im
* used to force a reconnect to the primary broker after failing over to a secondary
* broker.
*/
@Override
public void resetConnection() {
destroy();
}

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2021 the original author or authors.
* Copyright 2020-2022 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.
@@ -140,6 +140,7 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
* used to force a reconnect to the primary broker after failing over to a secondary
* broker.
*/
@Override
public void resetConnection() {
destroy();
}