25600722838

Committed 09 May 2026 12:06PM UTC coverage: 83.396% (-5.3%) from 88.677%

Build # 25600722838

Build Type

push

github

Committed by

web-flow

Commit Message

Nats v2 web (#295)

* 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 shipped the t... (continued)

Coverage Stats

2566 of 3407 branches covered (75.32%)

Branch coverage included in aggregate %.

795 of 1072 new or added lines in 22 files covered. (74.16%)

312 existing lines in 38 files now uncovered.

7715 of 8921 relevant lines covered (86.48%)

0.86 hits per line

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

91.86

/rqueue-core/src/main/java/com/github/sonus21/rqueue/common/RqueueRedisTemplate.java

/*
 * Copyright (c) 2020-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.common;

import com.github.sonus21.rqueue.utils.RedisUtils;
import java.io.Serializable;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;

@Slf4j
public class RqueueRedisTemplate<V extends Serializable> {

  protected RedisTemplate<String, V> redisTemplate;

  public RqueueRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
    if (redisConnectionFactory == null) {
      // Permitted on the NATS backend path, where the template is constructed for type
      // satisfaction but never used for Redis operations. Leaving redisTemplate null fails
      // fast and obviously if anyone does try to use it.
      this.redisTemplate = null;
      return;
    }
    this.redisTemplate = RedisUtils.getRedisTemplate(redisConnectionFactory);
    this.redisTemplate.afterPropertiesSet();
  }

  public RedisTemplate<String, V> getRedisTemplate() {
    return this.redisTemplate;
  }

  public Long removeFromZset(String zsetName, V val) {
    return redisTemplate.opsForZSet().remove(zsetName, val);
  }

  public Long getListSize(String lName) {
    return redisTemplate.opsForList().size(lName);
  }

  public Long getZsetSize(String zsetName) {
    return redisTemplate.opsForZSet().size(zsetName);
  }

  public Double getZsetMemberScore(String zsetName, String key) {
    return redisTemplate.opsForZSet().score(zsetName, key);
  }

  public Long rpush(String listName, V val) {
    return redisTemplate.opsForList().rightPush(listName, val);
  }

  public Long lpush(String listName, V val) {
    return redisTemplate.opsForList().leftPush(listName, val);
  }

  public Long addToSet(String setName, V... values) {
    return redisTemplate.opsForSet().add(setName, values);
  }

  public void set(String key, V val) {
    redisTemplate.opsForValue().set(key, val);
  }

  public V get(String key) {
    return redisTemplate.opsForValue().get(key);
  }

  public boolean exist(String key) {
    return ttl(key) != -2;
  }

  public int ttl(String key) {
    return redisTemplate.getExpire(key, TimeUnit.SECONDS).intValue();
  }

  public List<V> mget(Collection<String> keys) {
    return redisTemplate.opsForValue().multiGet(keys);
  }

  public void mset(Map<String, V> map) {
    redisTemplate.opsForValue().multiSet(map);
  }

  public void set(String key, V val, Duration duration) {
    redisTemplate.opsForValue().set(key, val, duration.toMillis(), TimeUnit.MILLISECONDS);
  }

  public void putHashValue(String key, String hashKey, V val) {
    redisTemplate.opsForHash().put(key, hashKey, val);
  }

  @SuppressWarnings("unchecked")
  public Map<String, V> getHashEntries(String key) {
    return (Map<String, V>) (Map<?, ?>) redisTemplate.opsForHash().entries(key);
  }

  public Long deleteHashValues(String key, String... hashKeys) {
    return redisTemplate.opsForHash().delete(key, (Object[]) hashKeys);
  }

  public Boolean expire(String key, Duration duration) {
    return redisTemplate.expire(key, duration.toMillis(), TimeUnit.MILLISECONDS);
  }

  public Boolean setIfAbsent(String lockKey, V val, Duration duration) {
    boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, val);
    if (result) {
      redisTemplate.expire(lockKey, duration.toMillis(), TimeUnit.MILLISECONDS);
    }
    return result;
  }

  public Boolean delete(String key) {
    return redisTemplate.delete(key);
  }

  public Object delete(Collection<String> keys) {
    return redisTemplate.delete(keys);
  }

  public DataType type(String key) {
    return redisTemplate.type(key);
  }

  public List<V> lrange(String key, long start, long end) {
    return redisTemplate.opsForList().range(key, start, end);
  }

  public Set<V> zrange(String key, long start, long end) {
    return redisTemplate.opsForZSet().range(key, start, end);
  }

  public Set<TypedTuple<V>> zrangeWithScore(String key, long start, long end) {
    return redisTemplate.opsForZSet().rangeWithScores(key, start, end);
  }

  public void zremRangeByScore(String key, long min, long max) {
    redisTemplate.opsForZSet().removeRangeByScore(key, min, max);
  }

  public Set<V> getMembers(String key) {
    return redisTemplate.opsForSet().members(key);
  }

  public void ltrim(String key, Integer start, Integer end) {
    redisTemplate.opsForList().trim(key, start, end);
  }

  public Boolean zadd(String key, V val, long score) {
    return redisTemplate.opsForZSet().add(key, val, score);
  }

  public void rename(String oldKey, String newKey) {
    rename(Collections.singletonList(oldKey), Collections.singletonList(newKey));
  }

  public void rename(List<String> oldKeys, List<String> newKeys) {
    if (oldKeys.size() != newKeys.size()) {
      throw new IllegalArgumentException("Old key and new key space set is different");
    }
    List<String> srcKeys = new LinkedList<>();
    List<String> dstKeys = new LinkedList<>();
    for (int i = 0; i < oldKeys.size(); i++) {
      String key = oldKeys.get(i);
      if (exist(key)) {
        srcKeys.add(key);
        dstKeys.add(newKeys.get(i));
      }
    }
    log.debug(
        "Pipeline result: {}",
        RedisUtils.executePipeLine(redisTemplate, (connection, keySerializer, valueSerializer) -> {
          for (int i = 0; i < srcKeys.size(); i++) {
            // TODO fix cross slot error?
            byte[] src = keySerializer.serialize(srcKeys.get(i));
            byte[] dst = keySerializer.serialize(dstKeys.get(i));
            connection.rename(src, dst);
          }
        }));
    if (srcKeys.size() != oldKeys.size()) {
      List<String> diff = new LinkedList<>(oldKeys);
      diff.removeAll(srcKeys);
      log.info("Some keys does not exist diff: {}", diff);
    }
  }
}

1	/*
2	* Copyright (c) 2020-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
17	package com.github.sonus21.rqueue.common;
18
19	import com.github.sonus21.rqueue.utils.RedisUtils;
20	import java.io.Serializable;
21	import java.time.Duration;
22	import java.util.Collection;
23	import java.util.Collections;
24	import java.util.LinkedList;
25	import java.util.List;
26	import java.util.Map;
27	import java.util.Set;
28	import java.util.concurrent.TimeUnit;
29	import lombok.extern.slf4j.Slf4j;
30	import org.springframework.data.redis.connection.DataType;
31	import org.springframework.data.redis.connection.RedisConnectionFactory;
32	import org.springframework.data.redis.core.RedisTemplate;
33	import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
34
35	@Slf4j	1✔
36	public class RqueueRedisTemplate<V extends Serializable> {
37
38	protected RedisTemplate<String, V> redisTemplate;
39
40	public RqueueRedisTemplate(RedisConnectionFactory redisConnectionFactory) {	1✔
41	if (redisConnectionFactory == null) {	1✔
42	// Permitted on the NATS backend path, where the template is constructed for type
43	// satisfaction but never used for Redis operations. Leaving redisTemplate null fails
44	// fast and obviously if anyone does try to use it.
45	this.redisTemplate = null;	1✔
46	return;	1✔
47	}
48	this.redisTemplate = RedisUtils.getRedisTemplate(redisConnectionFactory);	1✔
49	this.redisTemplate.afterPropertiesSet();	1✔
50	}	1✔
51
52	public RedisTemplate<String, V> getRedisTemplate() {
53	return this.redisTemplate;	1✔
54	}
55
56	public Long removeFromZset(String zsetName, V val) {
57	return redisTemplate.opsForZSet().remove(zsetName, val);	1✔
58	}
59
60	public Long getListSize(String lName) {
61	return redisTemplate.opsForList().size(lName);	1✔
62	}
63
64	public Long getZsetSize(String zsetName) {
65	return redisTemplate.opsForZSet().size(zsetName);	1✔
66	}
67
68	public Double getZsetMemberScore(String zsetName, String key) {
UNCOV 69	return redisTemplate.opsForZSet().score(zsetName, key);	×
70	}
71
72	public Long rpush(String listName, V val) {
73	return redisTemplate.opsForList().rightPush(listName, val);	1✔
74	}
75
76	public Long lpush(String listName, V val) {
77	return redisTemplate.opsForList().leftPush(listName, val);	1✔
78	}
79
80	public Long addToSet(String setName, V... values) {
UNCOV 81	return redisTemplate.opsForSet().add(setName, values);	×
82	}
83
84	public void set(String key, V val) {
85	redisTemplate.opsForValue().set(key, val);	1✔
86	}	1✔
87
88	public V get(String key) {
89	return redisTemplate.opsForValue().get(key);	1✔
90	}
91
92	public boolean exist(String key) {
93	return ttl(key) != -2;	1✔
94	}
95
96	public int ttl(String key) {
97	return redisTemplate.getExpire(key, TimeUnit.SECONDS).intValue();	1✔
98	}
99
100	public List<V> mget(Collection<String> keys) {
101	return redisTemplate.opsForValue().multiGet(keys);	1✔
102	}
103
104	public void mset(Map<String, V> map) {
105	redisTemplate.opsForValue().multiSet(map);	1✔
106	}	1✔
107
108	public void set(String key, V val, Duration duration) {
109	redisTemplate.opsForValue().set(key, val, duration.toMillis(), TimeUnit.MILLISECONDS);	1✔
110	}	1✔
111
112	public void putHashValue(String key, String hashKey, V val) {
113	redisTemplate.opsForHash().put(key, hashKey, val);	1✔
114	}	1✔
115
116	@SuppressWarnings("unchecked")
117	public Map<String, V> getHashEntries(String key) {
118	return (Map<String, V>) (Map<?, ?>) redisTemplate.opsForHash().entries(key);	1✔
119	}
120
121	public Long deleteHashValues(String key, String... hashKeys) {
122	return redisTemplate.opsForHash().delete(key, (Object[]) hashKeys);	1✔
123	}
124
125	public Boolean expire(String key, Duration duration) {
126	return redisTemplate.expire(key, duration.toMillis(), TimeUnit.MILLISECONDS);	1✔
127	}
128
129	public Boolean setIfAbsent(String lockKey, V val, Duration duration) {
130	boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, val);	1✔
131	if (result) {	1✔
132	redisTemplate.expire(lockKey, duration.toMillis(), TimeUnit.MILLISECONDS);	1✔
133	}
134	return result;	1✔
135	}
136
137	public Boolean delete(String key) {
138	return redisTemplate.delete(key);	1✔
139	}
140
141	public Object delete(Collection<String> keys) {
142	return redisTemplate.delete(keys);	1✔
143	}
144
145	public DataType type(String key) {
146	return redisTemplate.type(key);	1✔
147	}
148
149	public List<V> lrange(String key, long start, long end) {
150	return redisTemplate.opsForList().range(key, start, end);	1✔
151	}
152
153	public Set<V> zrange(String key, long start, long end) {
154	return redisTemplate.opsForZSet().range(key, start, end);	1✔
155	}
156
157	public Set<TypedTuple<V>> zrangeWithScore(String key, long start, long end) {
158	return redisTemplate.opsForZSet().rangeWithScores(key, start, end);	1✔
159	}
160
161	public void zremRangeByScore(String key, long min, long max) {
162	redisTemplate.opsForZSet().removeRangeByScore(key, min, max);	1✔
163	}	1✔
164
165	public Set<V> getMembers(String key) {
UNCOV 166	return redisTemplate.opsForSet().members(key);	×
167	}
168
169	public void ltrim(String key, Integer start, Integer end) {
UNCOV 170	redisTemplate.opsForList().trim(key, start, end);	×
UNCOV 171	}	×
172
173	public Boolean zadd(String key, V val, long score) {
174	return redisTemplate.opsForZSet().add(key, val, score);	1✔
175	}
176
177	public void rename(String oldKey, String newKey) {
178	rename(Collections.singletonList(oldKey), Collections.singletonList(newKey));	1✔
179	}	1✔
180
181	public void rename(List<String> oldKeys, List<String> newKeys) {
182	if (oldKeys.size() != newKeys.size()) {	1!
UNCOV 183	throw new IllegalArgumentException("Old key and new key space set is different");	×
184	}
185	List<String> srcKeys = new LinkedList<>();	1✔
186	List<String> dstKeys = new LinkedList<>();	1✔
187	for (int i = 0; i < oldKeys.size(); i++) {	1✔
188	String key = oldKeys.get(i);	1✔
189	if (exist(key)) {	1✔
190	srcKeys.add(key);	1✔
191	dstKeys.add(newKeys.get(i));	1✔
192	}
193	}
194	log.debug(	1✔
195	"Pipeline result: {}",
196	RedisUtils.executePipeLine(redisTemplate, (connection, keySerializer, valueSerializer) -> {	1✔
197	for (int i = 0; i < srcKeys.size(); i++) {	1✔
198	// TODO fix cross slot error?
199	byte[] src = keySerializer.serialize(srcKeys.get(i));	1✔
200	byte[] dst = keySerializer.serialize(dstKeys.get(i));	1✔
201	connection.rename(src, dst);	1✔
202	}
203	}));	1✔
204	if (srcKeys.size() != oldKeys.size()) {	1✔
205	List<String> diff = new LinkedList<>(oldKeys);	1✔
206	diff.removeAll(srcKeys);	1✔
207	log.info("Some keys does not exist diff: {}", diff);	1✔
208	}
209	}	1✔
210	}

sonus21 / rqueue / 25600722838

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