25607335827

Committed 09 May 2026 05:31PM UTC coverage: 83.708% (+0.3%) from 83.396%

Build # 25607335827

Build Type

Pull #297

github

Committed by

web-flow

Commit Message

Merge 6ead9c27a into a868dcde0

Pull Request Pull Request #297: Nats scheduling fix

Coverage Stats

2611 of 3455 branches covered (75.57%)

Branch coverage included in aggregate %.

96 of 107 new or added lines in 5 files covered. (89.72%)

5 existing lines in 2 files now uncovered.

7814 of 8999 relevant lines covered (86.83%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.43

/rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/js/NatsDeadLetterBridgeRegistrar.java

/*
 * 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.nats.js;

import com.github.sonus21.rqueue.config.RqueueConfig;
import com.github.sonus21.rqueue.core.EndpointRegistry;
import com.github.sonus21.rqueue.core.spi.MessageBroker;
import com.github.sonus21.rqueue.listener.QueueDetail;
import java.util.ArrayList;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.SmartInitializingSingleton;

/**
 * Bootstrap-time installer for the NATS-native dead-letter advisory bridge. For every active queue
 * registered in {@link EndpointRegistry}, calls
 * {@link JetStreamMessageBroker#installDeadLetterBridge(QueueDetail, String)} so that messages
 * exceeding {@code maxDeliver} on the durable consumer are republished onto the queue's DLQ
 * stream via the {@code $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES} advisory subject.
 *
 * <p>This is the NATS-side equivalent of the Redis backend's {@code RqueueDeadLetterPublisher}:
 * the rqueue post-processing handler already routes failed deliveries to a configured Rqueue-level
 * DLQ ({@code @RqueueListener(deadLetterQueue=...)}); the advisory bridge registered here is an
 * additional, independent path that catches messages whose handler exhausted retries without the
 * post-processor noticing (e.g. listener restart, container shutdown mid-retry, JetStream-driven
 * redelivery exhaustion outside the rqueue retry counter).
 *
 * <p><b>Lifecycle.</b> Implements {@link SmartInitializingSingleton} so it runs after every
 * {@code @RqueueListener} bean has registered with {@link EndpointRegistry} and the
 * {@link com.github.sonus21.rqueue.nats.js.NatsStreamValidator} has provisioned the underlying
 * streams — but before {@code SmartLifecycle.start()} spawns the message pollers, so the bridge
 * is in place before the first delivery attempt. Implements {@link DisposableBean} so the
 * advisory dispatchers are torn down on context shutdown.
 *
 * <p><b>Producer-only mode.</b> When {@link RqueueConfig#isProducer()} is true the application
 * has no listeners and therefore no consumers that could exhaust retries; the registrar exits
 * early and installs nothing.
 *
 * <p><b>Backend gating.</b> Only does its work when the active broker is a
 * {@link JetStreamMessageBroker}; on Redis or other backends the bean simply no-ops, so it is safe
 * to wire unconditionally from the NATS auto-config (which is itself gated on
 * {@code rqueue.backend=nats}).
 */
public class NatsDeadLetterBridgeRegistrar
    implements SmartInitializingSingleton, DisposableBean {

  private static final Logger log =
      Logger.getLogger(NatsDeadLetterBridgeRegistrar.class.getName());

  private final MessageBroker broker;
  private final RqueueConfig rqueueConfig;
  private final List<AutoCloseable> bridges = new ArrayList<>();

  public NatsDeadLetterBridgeRegistrar(MessageBroker broker, RqueueConfig rqueueConfig) {
    this.broker = broker;
    this.rqueueConfig = rqueueConfig;
  }

  @Override
  public void afterSingletonsInstantiated() {
    if (rqueueConfig != null && rqueueConfig.isProducer()) {
      log.log(Level.FINE,
          "NatsDeadLetterBridgeRegistrar: producer-only mode — skipping bridge installation");
      return;
    }
    if (!(broker instanceof JetStreamMessageBroker)) {
      // Defensive — the bean is wired only by the NATS auto-config, but other backends could
      // theoretically substitute a different MessageBroker via @Primary.
      return;
    }
    JetStreamMessageBroker nb = (JetStreamMessageBroker) broker;
    List<QueueDetail> queues = EndpointRegistry.getActiveQueueDetails();
    if (queues.isEmpty()) {
      return;
    }
    int installed = 0;
    for (QueueDetail q : queues) {
      String consumerName = q.resolvedConsumerName();
      try {
        bridges.add(nb.installDeadLetterBridge(q, consumerName));
        installed++;
      } catch (RuntimeException e) {
        // Best-effort: a single failure must not abort listener startup. The rqueue-level DLQ
        // path (PostProcessingHandler.moveToDlq) still works regardless.
        log.log(Level.WARNING,
            "Failed to install dead-letter advisory bridge for queue " + q.getName()
                + " consumer " + consumerName + ": " + e.getMessage(), e);
      }
    }
    log.log(Level.INFO,
        "NatsDeadLetterBridgeRegistrar: installed {0} advisory bridge(s) across {1} queue(s)",
        new Object[] {installed, queues.size()});
  }

  @Override
  public void destroy() {
    for (AutoCloseable c : bridges) {
      try {
        c.close();
      } catch (Exception ignore) {
        // best-effort close; we are shutting down anyway
      }
    }
    bridges.clear();
  }
}

1	/*
2	* Copyright (c) 2026 Sonu Kumar
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* You may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* https://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and limitations under the License.
14	*
15	*/
16	package com.github.sonus21.rqueue.nats.js;
17
18	import com.github.sonus21.rqueue.config.RqueueConfig;
19	import com.github.sonus21.rqueue.core.EndpointRegistry;
20	import com.github.sonus21.rqueue.core.spi.MessageBroker;
21	import com.github.sonus21.rqueue.listener.QueueDetail;
22	import java.util.ArrayList;
23	import java.util.List;
24	import java.util.logging.Level;
25	import java.util.logging.Logger;
26	import org.springframework.beans.factory.DisposableBean;
27	import org.springframework.beans.factory.SmartInitializingSingleton;
28
29	/**
30	* Bootstrap-time installer for the NATS-native dead-letter advisory bridge. For every active queue
31	* registered in {@link EndpointRegistry}, calls
32	* {@link JetStreamMessageBroker#installDeadLetterBridge(QueueDetail, String)} so that messages
33	* exceeding {@code maxDeliver} on the durable consumer are republished onto the queue's DLQ
34	* stream via the {@code $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES} advisory subject.
35	*
36	* <p>This is the NATS-side equivalent of the Redis backend's {@code RqueueDeadLetterPublisher}:
37	* the rqueue post-processing handler already routes failed deliveries to a configured Rqueue-level
38	* DLQ ({@code @RqueueListener(deadLetterQueue=...)}); the advisory bridge registered here is an
39	* additional, independent path that catches messages whose handler exhausted retries without the
40	* post-processor noticing (e.g. listener restart, container shutdown mid-retry, JetStream-driven
41	* redelivery exhaustion outside the rqueue retry counter).
42	*
43	* <p><b>Lifecycle.</b> Implements {@link SmartInitializingSingleton} so it runs after every
44	* {@code @RqueueListener} bean has registered with {@link EndpointRegistry} and the
45	* {@link com.github.sonus21.rqueue.nats.js.NatsStreamValidator} has provisioned the underlying
46	* streams — but before {@code SmartLifecycle.start()} spawns the message pollers, so the bridge
47	* is in place before the first delivery attempt. Implements {@link DisposableBean} so the
48	* advisory dispatchers are torn down on context shutdown.
49	*
50	* <p><b>Producer-only mode.</b> When {@link RqueueConfig#isProducer()} is true the application
51	* has no listeners and therefore no consumers that could exhaust retries; the registrar exits
52	* early and installs nothing.
53	*
54	* <p><b>Backend gating.</b> Only does its work when the active broker is a
55	* {@link JetStreamMessageBroker}; on Redis or other backends the bean simply no-ops, so it is safe
56	* to wire unconditionally from the NATS auto-config (which is itself gated on
57	* {@code rqueue.backend=nats}).
58	*/
59	public class NatsDeadLetterBridgeRegistrar
60	implements SmartInitializingSingleton, DisposableBean {
61
62	private static final Logger log =	1✔
63	Logger.getLogger(NatsDeadLetterBridgeRegistrar.class.getName());	1✔
64
65	private final MessageBroker broker;
66	private final RqueueConfig rqueueConfig;
67	private final List<AutoCloseable> bridges = new ArrayList<>();	1✔
68
69	public NatsDeadLetterBridgeRegistrar(MessageBroker broker, RqueueConfig rqueueConfig) {	1✔
70	this.broker = broker;	1✔
71	this.rqueueConfig = rqueueConfig;	1✔
72	}	1✔
73
74	@Override
75	public void afterSingletonsInstantiated() {
76	if (rqueueConfig != null && rqueueConfig.isProducer()) {	1!
NEW 77	log.log(Level.FINE,	×
78	"NatsDeadLetterBridgeRegistrar: producer-only mode — skipping bridge installation");
NEW 79	return;	×
80	}
81	if (!(broker instanceof JetStreamMessageBroker)) {	1!
82	// Defensive — the bean is wired only by the NATS auto-config, but other backends could
83	// theoretically substitute a different MessageBroker via @Primary.
NEW 84	return;	×
85	}
86	JetStreamMessageBroker nb = (JetStreamMessageBroker) broker;	1✔
87	List<QueueDetail> queues = EndpointRegistry.getActiveQueueDetails();	1✔
88	if (queues.isEmpty()) {	1✔
89	return;	1✔
90	}
91	int installed = 0;	1✔
92	for (QueueDetail q : queues) {	1✔
93	String consumerName = q.resolvedConsumerName();	1✔
94	try {
95	bridges.add(nb.installDeadLetterBridge(q, consumerName));	1✔
96	installed++;	1✔
NEW 97	} catch (RuntimeException e) {	×
98	// Best-effort: a single failure must not abort listener startup. The rqueue-level DLQ
99	// path (PostProcessingHandler.moveToDlq) still works regardless.
NEW 100	log.log(Level.WARNING,	×
NEW 101	"Failed to install dead-letter advisory bridge for queue " + q.getName()	×
NEW 102	+ " consumer " + consumerName + ": " + e.getMessage(), e);	×
103	}	1✔
104	}	1✔
105	log.log(Level.INFO,	1✔
106	"NatsDeadLetterBridgeRegistrar: installed {0} advisory bridge(s) across {1} queue(s)",
107	new Object[] {installed, queues.size()});	1✔
108	}	1✔
109
110	@Override
111	public void destroy() {
112	for (AutoCloseable c : bridges) {	1!
113	try {
NEW 114	c.close();	×
NEW 115	} catch (Exception ignore) {	×
116	// best-effort close; we are shutting down anyway
NEW 117	}	×
NEW 118	}	×
119	bridges.clear();	1✔
120	}	1✔
121	}

sonus21 / rqueue / 25607335827

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous