25621442036

Committed 10 May 2026 06:06AM UTC coverage: 83.337% (-0.06%) from 83.396%

Build # 25621442036

Build Type

Pull #297

github

Committed by

web-flow

Commit Message

Merge 151ec0fb3 into a868dcde0

Pull Request Pull Request #297: Nats scheduling fix

Coverage Stats

2625 of 3487 branches covered (75.28%)

Branch coverage included in aggregate %.

129 of 180 new or added lines in 12 files covered. (71.67%)

11 existing lines in 3 files now uncovered.

7833 of 9062 relevant lines covered (86.44%)

0.86 hits per line

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

0.0

/rqueue-core/src/main/java/com/github/sonus21/rqueue/metrics/RqueueQueueMetricsProvider.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.metrics;

/**
 * Backend-agnostic provider of queue-depth metrics. Each backend (Redis, NATS, ...) supplies its
 * own implementation; consumers like {@link RqueueMetrics} read sizes through this interface
 * instead of reaching into a Redis-shaped DAO.
 *
 * <p>The {@code queueName} argument is the user-facing queue name (the value bound on
 * {@code @RqueueListener(value="...")}), not an internal storage key. Implementations are
 * responsible for mapping it to the appropriate backend-specific key(s).
 *
 * <p>Implementations must return {@code 0} when a queue has no messages of the requested kind
 * (rather than throwing) so callers can use the values directly as gauge readings.
 */
public interface RqueueQueueMetricsProvider {

  /**
   * Number of messages waiting to be consumed from {@code queueName} — i.e. enqueued and ready for
   * a worker to pick up, excluding messages already in-flight (processing) or scheduled for a
   * future delivery time.
   */
  long getPendingMessageCount(String queueName);

  /**
   * Number of messages enqueued to {@code queueName} with a future delivery time that has not yet
   * elapsed. Backends that don't support delayed delivery return {@code 0}.
   */
  long getScheduledMessageCount(String queueName);

  /**
   * Number of messages currently in-flight for {@code queueName} — handed to a worker but not yet
   * acked or nacked. Backends without an explicit in-flight set return {@code 0}.
   */
  long getProcessingMessageCount(String queueName);

  /**
   * Number of messages in the dead-letter queue associated with {@code queueName}. Returns
   * {@code 0} when no DLQ is configured for the queue or the backend does not surface DLQ depth.
   */
  long getDeadLetterMessageCount(String queueName);

  /**
   * Priority-aware variant of {@link #getPendingMessageCount(String)}. The default implementation
   * ignores priority and returns the parent queue depth, which is the right behaviour for backends
   * that don't model per-priority sub-queues.
   */
  default long getPendingMessageCount(String queueName, String priority) {
    return getPendingMessageCount(queueName);
  }

  /** Priority-aware variant of {@link #getScheduledMessageCount(String)}. */
  default long getScheduledMessageCount(String queueName, String priority) {
    return getScheduledMessageCount(queueName);
  }

  /** Priority-aware variant of {@link #getProcessingMessageCount(String)}. */
  default long getProcessingMessageCount(String queueName, String priority) {
    return getProcessingMessageCount(queueName);
  }

  /**
   * Per-consumer variant of {@link #getPendingMessageCount(String)}. When two
   * {@code @RqueueListener} methods on the same queue declare different {@code consumerName}
   * overrides, each gets its own QueueDetail and its own metric registration; backends that can
   * report per-consumer pending depth (e.g. NATS JetStream Limits/Interest streams or any
   * fan-out broker) should override this. The default delegates to the queue-level call so
   * single-consumer queues, and backends without per-consumer state, behave unchanged.
   *
   * @param queueName    user-facing queue name
   * @param consumerName consumer-name override from {@code @RqueueListener(consumerName=...)};
   *                     {@code null} or empty when no override is set
   */
  default long getPendingMessageCountByConsumer(String queueName, String consumerName) {
    return getPendingMessageCount(queueName);
  }

  /** Per-consumer variant of {@link #getProcessingMessageCount(String)}. See related javadoc. */
  default long getProcessingMessageCountByConsumer(String queueName, String consumerName) {
    return getProcessingMessageCount(queueName);
  }
}

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.metrics;
17
18	/**
19	* Backend-agnostic provider of queue-depth metrics. Each backend (Redis, NATS, ...) supplies its
20	* own implementation; consumers like {@link RqueueMetrics} read sizes through this interface
21	* instead of reaching into a Redis-shaped DAO.
22	*
23	* <p>The {@code queueName} argument is the user-facing queue name (the value bound on
24	* {@code @RqueueListener(value="...")}), not an internal storage key. Implementations are
25	* responsible for mapping it to the appropriate backend-specific key(s).
26	*
27	* <p>Implementations must return {@code 0} when a queue has no messages of the requested kind
28	* (rather than throwing) so callers can use the values directly as gauge readings.
29	*/
30	public interface RqueueQueueMetricsProvider {
31
32	/**
33	* Number of messages waiting to be consumed from {@code queueName} — i.e. enqueued and ready for
34	* a worker to pick up, excluding messages already in-flight (processing) or scheduled for a
35	* future delivery time.
36	*/
37	long getPendingMessageCount(String queueName);
38
39	/**
40	* Number of messages enqueued to {@code queueName} with a future delivery time that has not yet
41	* elapsed. Backends that don't support delayed delivery return {@code 0}.
42	*/
43	long getScheduledMessageCount(String queueName);
44
45	/**
46	* Number of messages currently in-flight for {@code queueName} — handed to a worker but not yet
47	* acked or nacked. Backends without an explicit in-flight set return {@code 0}.
48	*/
49	long getProcessingMessageCount(String queueName);
50
51	/**
52	* Number of messages in the dead-letter queue associated with {@code queueName}. Returns
53	* {@code 0} when no DLQ is configured for the queue or the backend does not surface DLQ depth.
54	*/
55	long getDeadLetterMessageCount(String queueName);
56
57	/**
58	* Priority-aware variant of {@link #getPendingMessageCount(String)}. The default implementation
59	* ignores priority and returns the parent queue depth, which is the right behaviour for backends
60	* that don't model per-priority sub-queues.
61	*/
62	default long getPendingMessageCount(String queueName, String priority) {
63	return getPendingMessageCount(queueName);	×
64	}
65
66	/** Priority-aware variant of {@link #getScheduledMessageCount(String)}. */
67	default long getScheduledMessageCount(String queueName, String priority) {
68	return getScheduledMessageCount(queueName);	×
69	}
70
71	/** Priority-aware variant of {@link #getProcessingMessageCount(String)}. */
72	default long getProcessingMessageCount(String queueName, String priority) {
73	return getProcessingMessageCount(queueName);	×
74	}
75
76	/**
77	* Per-consumer variant of {@link #getPendingMessageCount(String)}. When two
78	* {@code @RqueueListener} methods on the same queue declare different {@code consumerName}
79	* overrides, each gets its own QueueDetail and its own metric registration; backends that can
80	* report per-consumer pending depth (e.g. NATS JetStream Limits/Interest streams or any
81	* fan-out broker) should override this. The default delegates to the queue-level call so
82	* single-consumer queues, and backends without per-consumer state, behave unchanged.
83	*
84	* @param queueName user-facing queue name
85	* @param consumerName consumer-name override from {@code @RqueueListener(consumerName=...)};
86	* {@code null} or empty when no override is set
87	*/
88	default long getPendingMessageCountByConsumer(String queueName, String consumerName) {
NEW 89	return getPendingMessageCount(queueName);	×
90	}
91
92	/** Per-consumer variant of {@link #getProcessingMessageCount(String)}. See related javadoc. */
93	default long getProcessingMessageCountByConsumer(String queueName, String consumerName) {
NEW 94	return getProcessingMessageCount(queueName);	×
95	}
96	}

sonus21 / rqueue / 25621442036

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