• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

sonus21 / rqueue / 25621809822

10 May 2026 06:27AM UTC coverage: 83.425% (+0.03%) from 83.396%
25621809822

push

github

web-flow
Nats scheduling fix (#297)

* ci: compile main sources in coverage_report job

The coverage_report job was producing an effectively empty
jacocoTestReport.xml (3.4KB vs ~1.1MB locally) because no .class files
existed when coverageReportOnly ran — the job checked out source code
and downloaded .exec artifacts, but never compiled. JaCoCo's report
generator skips packages/classes it cannot resolve, so the merged XML
ended up with only <sessioninfo> entries and no <package> elements.

That made coverallsJacoco silently no-op via the
"source file set empty, skipping" branch in CoverallsReporter, so
"Push coverage to Coveralls" reported success without uploading.

Verified by downloading the coverage-report artifact from a recent run
and comparing its XML structure against a local build's report.

Assisted-By: Claude Code

* nats-web: implement pause / soft-delete admin ops and capability-aware Q-detail

Replace the all-stub `NatsRqueueUtilityService` with real impls for the operations
JetStream can model: `pauseUnpauseQueue` persists the `paused` flag on `QueueConfig`
in the queue-config KV bucket and notifies the local listener container so the poller
stops dispatching; `deleteMessage` is a soft delete via `MessageMetadataService`
(stream message persists, dashboard hides via the metadata flag); `getDataType`
reports `STREAM`. `moveMessage`, `enqueueMessage`, and `makeEmpty` deliberately
remain "not supported" — there is no JetStream primitive for those.

Update `RqueueQDetailServiceImpl.getRunningTasks` / `getScheduledTasks` to return
header-only tables when the broker capabilities suppress those sections, instead of
emitting zero rows or 501s on NATS.

20 new unit tests cover the pause/delete paths and lock in the still-unsupported
operations. Updates `nats-task.md` / `nats-task-v2.md` to reflect what landed.

Assisted-By: Claude Code

* nats-web: capability-aware nav / charts and stream-based peek

End-to-end browser-tested the NATS dashboard and shipp... (continued)

2628 of 3487 branches covered (75.37%)

Branch coverage included in aggregate %.

128 of 179 new or added lines in 12 files covered. (71.51%)

1 existing line in 1 file now uncovered.

7841 of 9062 relevant lines covered (86.53%)

0.87 hits per line

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

70.83
/rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/js/NatsDeadLetterBridgeRegistrar.java
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 implements SmartInitializingSingleton, DisposableBean {
60

61
  private static final Logger log = Logger.getLogger(NatsDeadLetterBridgeRegistrar.class.getName());
1✔
62

63
  private final MessageBroker broker;
64
  private final RqueueConfig rqueueConfig;
65
  private final List<AutoCloseable> bridges = new ArrayList<>();
1✔
66

67
  public NatsDeadLetterBridgeRegistrar(MessageBroker broker, RqueueConfig rqueueConfig) {
1✔
68
    this.broker = broker;
1✔
69
    this.rqueueConfig = rqueueConfig;
1✔
70
  }
1✔
71

72
  @Override
73
  public void afterSingletonsInstantiated() {
74
    if (rqueueConfig != null && rqueueConfig.isProducer()) {
1!
NEW
75
      log.log(
×
76
          Level.FINE,
77
          "NatsDeadLetterBridgeRegistrar: producer-only mode — skipping bridge installation");
NEW
78
      return;
×
79
    }
80
    if (!(broker instanceof JetStreamMessageBroker)) {
1!
81
      // Defensive — the bean is wired only by the NATS auto-config, but other backends could
82
      // theoretically substitute a different MessageBroker via @Primary.
NEW
83
      return;
×
84
    }
85
    JetStreamMessageBroker nb = (JetStreamMessageBroker) broker;
1✔
86
    List<QueueDetail> queues = EndpointRegistry.getActiveQueueDetails();
1✔
87
    if (queues.isEmpty()) {
1✔
88
      return;
1✔
89
    }
90
    int installed = 0;
1✔
91
    for (QueueDetail q : queues) {
1✔
92
      String consumerName = q.resolvedConsumerName();
1✔
93
      try {
94
        bridges.add(nb.installDeadLetterBridge(q, consumerName));
1✔
95
        installed++;
1✔
NEW
96
      } catch (RuntimeException e) {
×
97
        // Best-effort: a single failure must not abort listener startup. The rqueue-level DLQ
98
        // path (PostProcessingHandler.moveToDlq) still works regardless.
NEW
99
        log.log(
×
100
            Level.WARNING,
NEW
101
            "Failed to install dead-letter advisory bridge for queue " + q.getName() + " consumer "
×
NEW
102
                + consumerName + ": " + e.getMessage(),
×
103
            e);
104
      }
1✔
105
    }
1✔
106
    log.log(
1✔
107
        Level.INFO,
108
        "NatsDeadLetterBridgeRegistrar: installed {0} advisory bridge(s) across {1} queue(s)",
109
        new Object[] {installed, queues.size()});
1✔
110
  }
1✔
111

112
  @Override
113
  public void destroy() {
114
    for (AutoCloseable c : bridges) {
1!
115
      try {
NEW
116
        c.close();
×
NEW
117
      } catch (Exception ignore) {
×
118
        // best-effort close; we are shutting down anyway
NEW
119
      }
×
NEW
120
    }
×
121
    bridges.clear();
1✔
122
  }
1✔
123
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc