Delay Rqueue startup for Boot web apps

Author: sonus21

rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/RqueueMessageListenerContainer.java

@@ -517,10 +517,14 @@ private List<QueueDetail> getQueueDetail(String queue, MappingInformation mappin
 
   @Override
   public void start() {
-    log.info("Starting Rqueue Message container {}", RqueueConfig.getBrokerId());
     synchronized (lifecycleMgr) {
-      running = true;
+      if (running) {
+        log.debug("Rqueue Message container {} is already running", RqueueConfig.getBrokerId());
+        return;
+      }
+      log.info("Starting Rqueue Message container {}", RqueueConfig.getBrokerId());
       doStart();
+      running = true;
       rqueueBeanProvider
           .getApplicationEventPublisher()
           .publishEvent(new RqueueBootstrapEvent(EVENT_SOURCE, true));

rqueue-spring-boot-starter/src/main/java/com/github/sonus21/rqueue/spring/boot/RqueueAutoStartupLifecycle.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2026 Sonu Kumar
+ *
+ * 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 com.github.sonus21.rqueue.spring.boot;
+
+import com.github.sonus21.rqueue.listener.RqueueMessageListenerContainer;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.ApplicationListener;
+
+/**
+ * Delays Rqueue poller startup in Boot web applications until the servlet or reactive web server
+ * is ready and Spring Boot has published {@link ApplicationReadyEvent}.
+ */
+public class RqueueAutoStartupLifecycle
+    implements BeanPostProcessor, ApplicationListener<ApplicationReadyEvent> {
+
+  private final Set<RqueueMessageListenerContainer> delayedContainers =
+      ConcurrentHashMap.newKeySet();
+
+  @Override
+  public Object postProcessBeforeInitialization(Object bean, String beanName)
+      throws BeansException {
+    if (bean instanceof RqueueMessageListenerContainer) {
+      RqueueMessageListenerContainer container = (RqueueMessageListenerContainer) bean;
+      if (container.isAutoStartup()) {
+        container.setAutoStartup(false);
+        delayedContainers.add(container);
+      }
+    }
+    return bean;
+  }
+
+  @Override
+  public void onApplicationEvent(ApplicationReadyEvent event) {
+    for (RqueueMessageListenerContainer container : delayedContainers) {
+      if (!container.isRunning()) {
+        container.start();
+      }
+    }
+  }
+}

rqueue-spring-boot-starter/src/main/java/com/github/sonus21/rqueue/spring/boot/RqueueListenerAutoConfig.java

@@ -33,15 +33,18 @@
 import com.github.sonus21.rqueue.listener.RqueueMessageListenerContainer;
 import com.github.sonus21.rqueue.utils.condition.ReactiveEnabled;
 import com.github.sonus21.rqueue.utils.condition.RqueueEnabled;
+import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
 import org.springframework.boot.data.redis.autoconfigure.DataRedisAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Role;
 
 @Configuration
 @AutoConfigureAfter(DataRedisAutoConfiguration.class)
@@ -54,6 +57,13 @@
 @Import(RqueueRedisConfigImportSelector.class)
 public class RqueueListenerAutoConfig extends RqueueListenerBaseConfig {
 
+  @Bean
+  @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+  @ConditionalOnWebApplication
+  public static RqueueAutoStartupLifecycle rqueueAutoStartupLifecycle() {
+    return new RqueueAutoStartupLifecycle();
+  }
+
   @Bean
   @ConditionalOnMissingBean
   public RqueueMessageHandler rqueueMessageHandler(MessageBroker messageBroker) {

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/unit/RqueueListenerAutoConfigTest.java

@@ -17,6 +17,7 @@
 package com.github.sonus21.rqueue.spring.boot.tests.unit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -35,6 +36,8 @@
 import com.github.sonus21.rqueue.core.spi.Capabilities;
 import com.github.sonus21.rqueue.core.spi.MessageBroker;
 import com.github.sonus21.rqueue.listener.RqueueMessageHandler;
+import com.github.sonus21.rqueue.listener.RqueueMessageListenerContainer;
+import com.github.sonus21.rqueue.spring.boot.RqueueAutoStartupLifecycle;
 import com.github.sonus21.rqueue.spring.boot.RqueueListenerAutoConfig;
 import com.github.sonus21.rqueue.spring.boot.tests.SpringBootUnitTest;
 import org.apache.commons.lang3.reflect.FieldUtils;
@@ -125,6 +128,33 @@ void rqueueMessageListenerContainer()
     assertSame(messageBroker, factory.getMessageBroker());
   }
 
+  @Test
+  void autoStartupLifecycleDelaysOnlyAutoStartupContainers() {
+    RqueueAutoStartupLifecycle lifecycle = new RqueueAutoStartupLifecycle();
+    TestRqueueMessageListenerContainer autoStartupContainer =
+        new TestRqueueMessageListenerContainer();
+    TestRqueueMessageListenerContainer disabledContainer =
+        new TestRqueueMessageListenerContainer();
+    TestRqueueMessageListenerContainer alreadyRunningContainer =
+        new TestRqueueMessageListenerContainer();
+    disabledContainer.setAutoStartup(false);
+
+    lifecycle.postProcessBeforeInitialization(autoStartupContainer, "autoStartupContainer");
+    lifecycle.postProcessBeforeInitialization(disabledContainer, "disabledContainer");
+    lifecycle.postProcessBeforeInitialization(alreadyRunningContainer, "alreadyRunningContainer");
+    alreadyRunningContainer.running = true;
+
+    assertFalse(autoStartupContainer.isAutoStartup());
+    assertFalse(disabledContainer.isAutoStartup());
+    assertFalse(alreadyRunningContainer.isAutoStartup());
+
+    lifecycle.onApplicationEvent(null);
+
+    assertEquals(1, autoStartupContainer.startCount);
+    assertEquals(0, disabledContainer.startCount);
+    assertEquals(0, alreadyRunningContainer.startCount);
+  }
+
   @Test
   void rqueueMessageEnqueuerWiresBroker() throws IllegalAccessException {
     SimpleRqueueListenerContainerFactory factory = new SimpleRqueueListenerContainerFactory();
@@ -158,4 +188,25 @@ void rqueueMessageSenderUsesConfiguredMessageConverter() throws IllegalAccessExc
     MessageConverter converter = messageSender.getMessageConverter();
     assertTrue(converter.hashCode() == messageConverter.hashCode());
   }
+
+  private class TestRqueueMessageListenerContainer extends RqueueMessageListenerContainer {
+
+    private int startCount;
+    private boolean running;
+
+    private TestRqueueMessageListenerContainer() {
+      super(rqueueMessageHandler, messageTemplate);
+    }
+
+    @Override
+    public void start() {
+      startCount++;
+      running = true;
+    }
+
+    @Override
+    public boolean isRunning() {
+      return running;
+    }
+  }
 }