fix: add channel shutdown listener to reconnect

Author: juancgalvis

.github/workflows/release.yml

@@ -1,18 +1,18 @@
 name: release
 on:
   release:
-    types: [ released ]
+    types: [ released, prereleased ]
 jobs:
   release:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
       - name: Set env
         run: echo "RELEASE_VERSION=${GITHUB_REF#refs/*/v}" >> $GITHUB_ENV
       - name: Upgrade Gradle.properties
         run: sed -i 's/version=.*/version=${{ env.RELEASE_VERSION }}/g' gradle.properties
       - name: Generate Changelog
-        uses: heinrichreimer/github-changelog-generator-action@v2.3
+        uses: heinrichreimer/github-changelog-generator-action@v2.4
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
           pullRequests: true
@@ -21,7 +21,7 @@ jobs:
           issuesWoLabels: true
           stripGeneratorNotice: true
       - name: Set up JDK 17
-        uses: actions/setup-java@v3
+        uses: actions/setup-java@v5
         with:
           distribution: temurin
           java-version: 17
@@ -35,13 +35,13 @@ jobs:
         env:
           ORG_GRADLE_PROJECT_sonatypeUsername: ${{ secrets.MAVEN_USERNAME }}
           ORG_GRADLE_PROJECT_sonatypePassword: ${{ secrets.MAVEN_PASSWORD }}
-      - name: Push Changelog
-        uses: github-actions-x/commit@v2.9
-        with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
-          push-branch: 'master'
-          commit-message: 'Automatic docs and changelog generation [skip ci]'
-          force-add: 'true'
-          files: CHANGELOG.md gradle.properties
-          name: ${{ github.actor }}
-          email: ${{ github.actor }}@users.noreply.github.com
+#      - name: Push Changelog
+#        uses: github-actions-x/commit@v2.9
+#        with:
+#          github-token: ${{ secrets.GITHUB_TOKEN }}
+#          push-branch: 'master'
+#          commit-message: 'Automatic docs and changelog generation [skip ci]'
+#          force-add: 'true'
+#          files: CHANGELOG.md gradle.properties
+#          name: ${{ github.actor }}
+#          email: ${{ github.actor }}@users.noreply.github.com

async/async-commons/async-commons.gradle

@@ -10,7 +10,7 @@ dependencies {
     compileOnly 'io.projectreactor:reactor-core'
     api 'io.projectreactor.rabbitmq:reactor-rabbitmq:1.5.6'
     api 'com.fasterxml.jackson.core:jackson-databind'
-    implementation 'commons-io:commons-io:2.16.1'
+    implementation 'commons-io:commons-io:2.21.0'
 
     testImplementation 'io.projectreactor:reactor-test'
 }

async/async-rabbit-starter/src/main/java/org/reactivecommons/async/rabbit/config/RabbitMqConfig.java

@@ -134,6 +134,7 @@ public BrokerConfig brokerConfig() {
     public ConnectionFactoryProvider rabbitRConnectionFactory(RabbitProperties properties)
             throws NoSuchAlgorithmException, KeyManagementException {
         final ConnectionFactory factory = new ConnectionFactory();
+        factory.setAutomaticRecoveryEnabled(true);
         PropertyMapper map = PropertyMapper.get();
         map.from(properties::determineHost).whenNonNull().to(factory::setHost);
         map.from(properties::determinePort).to(factory::setPort);

async/async-rabbit/src/main/java/org/reactivecommons/async/rabbit/listeners/GenericMessageListener.java

@@ -1,6 +1,8 @@
 package org.reactivecommons.async.rabbit.listeners;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.ShutdownSignalException;
 import lombok.extern.java.Log;
 import org.reactivecommons.async.commons.DiscardNotifier;
 import org.reactivecommons.async.commons.FallbackStrategy;
@@ -10,6 +12,7 @@
 import org.reactivecommons.async.rabbit.RabbitMessage;
 import org.reactivecommons.async.rabbit.communications.ReactiveMessageListener;
 import org.reactivecommons.async.rabbit.communications.TopologyCreator;
+import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
@@ -25,6 +28,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.logging.Level;
 
@@ -49,7 +53,8 @@ public abstract class GenericMessageListener {
     private final DiscardNotifier discardNotifier;
     private final String objectType;
     private final CustomReporter customReporter;
-    private volatile Flux<AcknowledgableDelivery> messageFlux;
+    private final AtomicReference<Channel> channelRef = new AtomicReference<>();
+    private Disposable listenerSubscription;
 
     public GenericMessageListener(String queueName, ReactiveMessageListener listener, boolean useDLQRetries,
                                   boolean createTopology, long maxRetries, long retryDelay, DiscardNotifier discardNotifier,
@@ -78,34 +83,45 @@ protected Mono<Void> setUpBindings(TopologyCreator creator) {
         return Mono.empty();
     }
 
-    public void startListener() {
-        log.log(Level.INFO, "Using max concurrency {0}, in queue: {1}", new Object[]{messageListener.getMaxConcurrency(), queueName});
-        if (useDLQRetries) {
-            log.log(Level.INFO, "ATTENTION! Using DLQ Strategy for retries with {0} + 1 Max Retries configured!", new Object[]{maxRetries});
-        } else {
-            log.log(Level.INFO, "ATTENTION! Using infinite fast retries as Retry Strategy");
+    public synchronized void startListener() {
+        Channel current = channelRef.get();
+        if (current != null && current.isOpen()) {
+            log.warning("Channel is already open, no need to restart listener");
+            return;
         }
+        stopListener();
+        var baseSubscriber = new LoggerSubscriber<>(getClass().getName());
+        listenerSubscription = baseSubscriber;
+        Flux.defer(this::buildConsumeFlux)
+                .subscribe(baseSubscriber);
+    }
 
-        ConsumeOptions consumeOptions = new ConsumeOptions();
-        consumeOptions.qos(messageListener.getPrefetchCount());
-
-        if (createTopology) {
-            this.messageFlux = setUpBindings(messageListener.getTopologyCreator())
-                    .thenMany(receiver.consumeManualAck(queueName, consumeOptions)
-                            .transform(this::consumeFaultTolerant));
-        } else {
-            this.messageFlux = receiver.consumeManualAck(queueName, consumeOptions)
-                    .doOnError(err -> log.log(Level.SEVERE, "Error listening queue", err))
-                    .transform(this::consumeFaultTolerant);
-        }
 
+    private Flux<AcknowledgableDelivery> buildConsumeFlux() {
+        ConsumeOptions options = new ConsumeOptions()
+                .qos(messageListener.getPrefetchCount())
+                .channelCallback(channel -> {
+                    channelRef.set(channel);
+                    channel.addShutdownListener(this::onChannelShutdown);
+                });
+
+        Flux<AcknowledgableDelivery> source = createTopology
+                ? setUpBindings(messageListener.getTopologyCreator())
+                  .thenMany(receiver.consumeManualAck(queueName, options))
+                : receiver.consumeManualAck(queueName, options);
 
-        onTerminate();
+        return source.transform(this::consumeFaultTolerant);
     }
 
-    private void onTerminate() {
-        messageFlux.doOnTerminate(this::onTerminate)
-                .subscribe(new LoggerSubscriber<>(getClass().getName()));
+    private void onChannelShutdown(ShutdownSignalException cause) {
+        log.log(Level.SEVERE, "Channel shutdown detected in listener", cause);
+        startListener();
+    }
+
+    public synchronized void stopListener() {
+        if (listenerSubscription != null && !listenerSubscription.isDisposed()) {
+            listenerSubscription.dispose();
+        }
     }
 
     protected Mono<AcknowledgableDelivery> handle(AcknowledgableDelivery msj, Instant initTime) {

build.gradle

@@ -13,7 +13,7 @@ buildscript {
 plugins {
     id 'jacoco'
     id 'org.sonarqube' version '5.1.0.4882'
-    id 'org.springframework.boot' version '3.3.1' apply false
+    id 'org.springframework.boot' version '3.5.13' apply false
     id 'io.github.gradle-nexus.publish-plugin' version '2.0.0'
     id 'co.com.bancolombia.cleanArchitecture' version '3.17.13'
 }

main.gradle

@@ -16,7 +16,10 @@ allprojects {
 
 nexusPublishing {
     repositories {
-        sonatype()
+        sonatype {
+            nexusUrl.set(uri("https://ossrh-staging-api.central.sonatype.com/service/local/"))
+            snapshotRepositoryUrl.set(uri("https://central.sonatype.com/repository/maven-snapshots/"))
+        }
     }
 }
 
@@ -32,6 +35,7 @@ subprojects {
 
     dependencies {
         testImplementation 'org.springframework.boot:spring-boot-starter-test'
+        testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
         testImplementation 'io.projectreactor:reactor-test'
         compileOnly 'org.projectlombok:lombok'
         annotationProcessor 'org.projectlombok:lombok'
@@ -55,7 +59,7 @@ subprojects {
 
     dependencyManagement {
         imports {
-            mavenBom 'org.springframework.boot:spring-boot-dependencies:3.3.1'
+            mavenBom 'org.springframework.boot:spring-boot-dependencies:3.5.13'
         }
     }