25606310187

Committed 09 May 2026 04:43PM UTC coverage: 83.722% (+0.3%) from 83.396%

Build # 25606310187

Build Type

Pull #297

github

Committed by

web-flow

Commit Message

Merge bf71e7ec5 into a868dcde0

Pull Request Pull Request #297: Nats scheduling fix

Coverage Stats

2600 of 3443 branches covered (75.52%)

Branch coverage included in aggregate %.

66 of 68 new or added lines in 3 files covered. (97.06%)

8 existing lines in 1 file now uncovered.

7784 of 8960 relevant lines covered (86.88%)

0.87 hits per line

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

83.33

/rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/internal/NatsProvisioner.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
 */
package com.github.sonus21.rqueue.nats.internal;

import com.github.sonus21.rqueue.enums.QueueType;
import com.github.sonus21.rqueue.nats.RqueueNatsConfig;
import com.github.sonus21.rqueue.nats.RqueueNatsException;
import io.nats.client.Connection;
import io.nats.client.JetStreamApiException;
import io.nats.client.JetStreamManagement;
import io.nats.client.KeyValue;
import io.nats.client.KeyValueManagement;
import io.nats.client.api.AckPolicy;
import io.nats.client.api.CompressionOption;
import io.nats.client.api.ConsumerConfiguration;
import io.nats.client.api.ConsumerInfo;
import io.nats.client.api.DeliverPolicy;
import io.nats.client.api.KeyValueConfiguration;
import io.nats.client.api.KeyValueStatus;
import io.nats.client.api.RetentionPolicy;
import io.nats.client.api.StreamConfiguration;
import io.nats.client.api.StreamInfo;
import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Idempotent stream/consumer/KV provisioning helpers. All methods are safe to call repeatedly;
 * if the target object already exists with a compatible config, this class leaves it alone. If it
 * exists but with diverging config (e.g. ackWait, maxDeliver), this class logs a WARN and does
 * NOT auto-mutate so user customizations are preserved.
 */
public class NatsProvisioner {

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

  private final Connection connection;
  private final KeyValueManagement kvm;
  private final JetStreamManagement jsm;
  private final RqueueNatsConfig config;

  private final ConcurrentHashMap<String, KeyValue> kvCache = new ConcurrentHashMap<>();
  private final ConcurrentHashMap<String, Object> kvLocks = new ConcurrentHashMap<>();

  // stream name → schedulingEnabled (true if the stream was created/updated with
  // allowMessageSchedules)
  private final ConcurrentHashMap<String, Boolean> streamsDone = new ConcurrentHashMap<>();
  private final ConcurrentHashMap<String, Object> streamLocks = new ConcurrentHashMap<>();

  // "streamName/requestedConsumerName" → actual consumer name (may differ for stale-rebind)
  private final ConcurrentHashMap<String, String> consumerCache = new ConcurrentHashMap<>();
  private final ConcurrentHashMap<String, Object> consumerLocks = new ConcurrentHashMap<>();

  /**
   * Minimum NATS server version that supports server-side message scheduling via the
   * {@code Nats-Schedule} JetStream publish header (ADR-51).
   */
  public static final String SCHEDULING_MIN_VERSION = "2.12.0";

  private final boolean schedulingSupported;

  public NatsProvisioner(Connection connection, JetStreamManagement jsm, RqueueNatsConfig config)
      throws IOException {
    this.connection = connection;
    this.kvm = connection.keyValueManagement();
    this.jsm = jsm;
    this.config = config;
    io.nats.client.api.ServerInfo serverInfo = connection.getServerInfo();
    this.schedulingSupported =
        serverInfo != null && serverInfo.isSameOrNewerThanVersion(SCHEDULING_MIN_VERSION);
    log.log(
        Level.INFO,
        "NATS server version={0}; message scheduling (ADR-51) supported={1}",
        new Object[] {serverInfo != null ? serverInfo.getVersion() : "unknown", schedulingSupported
        });
  }

  /** Returns {@code true} when the connected NATS server supports message scheduling (>= 2.12). */
  public boolean isMessageSchedulingSupported() {
    return schedulingSupported;
  }

  // ---- KV provisioning --------------------------------------------------

  /**
   * Returns a {@link KeyValue} handle for {@code bucketName}, creating the bucket on first call.
   * All buckets are created with S2 compression. A bucket-level TTL is applied at creation time
   * only; existing buckets are reused as-is.
   *
   * @param bucketName NATS KV bucket name (e.g. {@code "rqueue-jobs"})
   * @param ttl        bucket-level max-age; {@code null} or non-positive = no TTL
   */
  public KeyValue ensureKv(String bucketName, Duration ttl)
      throws IOException, JetStreamApiException {
    KeyValue cached = kvCache.get(bucketName);
    if (cached != null) {
      return cached;
    }
    Object lock = kvLocks.computeIfAbsent(bucketName, k -> new Object());
    synchronized (lock) {
      cached = kvCache.get(bucketName);
      if (cached != null) {
        return cached;
      }
      try {
        KeyValueStatus status = kvm.getStatus(bucketName);
        if (status != null) {
          KeyValue kv = connection.keyValue(bucketName);
          kvCache.put(bucketName, kv);
          return kv;
        }
      } catch (JetStreamApiException missing) {
        // bucket absent — fall through to create
      }
      RqueueNatsConfig.StreamDefaults sd = config.getStreamDefaults();
      KeyValueConfiguration.Builder cfg = KeyValueConfiguration.builder()
          .name(bucketName)
          .compression(true)
          .replicas(sd.getReplicas())
          .storageType(sd.getStorage());
      if (ttl != null && !ttl.isZero() && !ttl.isNegative()) {
        cfg.ttl(ttl);
      }
      kvm.create(cfg.build());
      KeyValue kv = connection.keyValue(bucketName);
      kvCache.put(bucketName, kv);
      return kv;
    }
  }

  // ---- Stream provisioning ----------------------------------------------

  /**
   * Ensure a JetStream stream exists with the given subjects, using {@link QueueType#QUEUE}
   * (WorkQueue retention) as the default. Callers that have a {@link QueueType} available should
   * use {@link #ensureStream(String, List, QueueType)} instead.
   */
  public void ensureStream(String streamName, List<String> subjects) {
    ensureStream(streamName, subjects, QueueType.QUEUE, null, false);
  }

  /** See {@link #ensureStream(String, List, QueueType, String, boolean)}. */
  public void ensureStream(String streamName, List<String> subjects, QueueType queueType) {
    ensureStream(streamName, subjects, queueType, null, false);
  }

  /** See {@link #ensureStream(String, List, QueueType, String, boolean)}. */
  public void ensureStream(
      String streamName, List<String> subjects, QueueType queueType, String description) {
    ensureStream(streamName, subjects, queueType, description, false);
  }

  /**
   * Ensure a JetStream stream exists with the given subjects and retention policy derived from
   * {@code queueType}:
   * <ul>
   *   <li>{@link QueueType#QUEUE} — {@link io.nats.client.api.RetentionPolicy#WorkQueue}: each
   *       message is delivered to exactly one consumer; competing-consumer semantics.
   *   <li>{@link QueueType#STREAM} — {@link io.nats.client.api.RetentionPolicy#Limits}: every
   *       independent durable consumer group receives all messages; stream/fan-out semantics.
   * </ul>
   *
   * <p>{@code description} is forwarded to JetStream as the stream's description (visible via
   * {@code nats stream info}). Callers should pass the rqueue queue name so operators can map a
   * stream back to the queue that created it; pass {@code null} to skip.
   *
   * <p>{@code allowSchedules} must be {@code true} when the stream will receive messages published
   * with the {@code Nats-Schedule} header (ADR-51). Only callers that perform delayed
   * enqueue should pass {@code true}; regular enqueue callers should pass {@code false} (or use the
   * shorter overloads). Equivalent to the CLI flag {@code --allow-schedules}.
   *
   * <p>Hits the NATS backend at most once per stream name per process lifetime; subsequent calls
   * return immediately from the in-process cache. If {@code allowSchedules=true} is later requested
   * for a stream that was previously created without that flag, the stream is updated in place via
   * {@link JetStreamManagement#updateStream}. If the stream already exists with a different
   * retention policy, a WARNING is logged and the existing config is left untouched.
   */
  public void ensureStream(
      String streamName,
      List<String> subjects,
      QueueType queueType,
      String description,
      boolean allowSchedules) {
    // Fast-path: already provisioned with at least as many capabilities as requested.
    Boolean cached = streamsDone.get(streamName);
    if (cached != null && (cached || !allowSchedules)) {
      return;
    }
    Object lock = streamLocks.computeIfAbsent(streamName, k -> new Object());
    synchronized (lock) {
      cached = streamsDone.get(streamName);
      if (cached != null && (cached || !allowSchedules)) {
        return;
      }
      boolean enableSchedules = allowSchedules && schedulingSupported;
      try {
        StreamInfo existing = safeGetStreamInfo(streamName);
        RetentionPolicy desired =
            queueType == QueueType.STREAM ? RetentionPolicy.Limits : RetentionPolicy.WorkQueue;
        if (existing == null) {
          if (!config.isAutoCreateStreams()) {
            throw new RqueueNatsException(
                "Stream '" + streamName + "' does not exist and autoCreateStreams=false");
          }
          RqueueNatsConfig.StreamDefaults sd = config.getStreamDefaults();
          StreamConfiguration.Builder b = StreamConfiguration.builder()
              .name(streamName)
              .subjects(subjects)
              .replicas(sd.getReplicas())
              .storageType(sd.getStorage())
              .retentionPolicy(desired)
              .compressionOption(CompressionOption.S2);
          if (enableSchedules) {
            // Enable server-side message scheduling (ADR-51 / Nats-Schedule header).
            // Equivalent to: nats stream add MY_STREAM --allow-schedules
            b.allowMessageSchedules(true);
          }
          if (description != null && !description.isEmpty()) {
            b.description(description);
          }
          if (sd.getMaxMsgs() > 0) {
            b.maxMessages(sd.getMaxMsgs());
          }
          if (sd.getMaxBytes() > 0) {
            b.maxBytes(sd.getMaxBytes());
          }
          if (sd.getMaxAge() != null
              && !sd.getMaxAge().isZero()
              && !sd.getMaxAge().isNegative()) {
            b.maxAge(sd.getMaxAge());
          }
          jsm.addStream(b.build());
        } else {
          RetentionPolicy actual = existing.getConfiguration().getRetentionPolicy();
          if (actual != desired) {
            log.log(
                Level.WARNING,
                "Stream ''{0}'' exists with retention={1} but queueMode requires retention={2}"
                    + " — leaving existing config in place.",
                new Object[] {streamName, actual, desired});
          }
          // Check whether new subjects need to be merged in (e.g. the sched wildcard added by
          // enqueueWithDelay after the stream was originally created by a plain enqueue call).
          java.util.List<String> existingSubjects = existing.getConfiguration().getSubjects();
          java.util.Set<String> existingSet = existingSubjects != null
              ? new java.util.HashSet<>(existingSubjects) : new java.util.HashSet<>();
          boolean needsSubjectUpdate = subjects.stream().anyMatch(s -> !existingSet.contains(s));
          boolean needsFlagUpdate =
              enableSchedules && !existing.getConfiguration().getAllowMsgSchedules();

          if (needsFlagUpdate || needsSubjectUpdate) {
            // Merge: keep all existing subjects and append new ones (never remove).
            java.util.LinkedHashSet<String> merged = new java.util.LinkedHashSet<>(existingSet);
            merged.addAll(subjects);
            StreamConfiguration.Builder upd =
                StreamConfiguration.builder(existing.getConfiguration())
                    .subjects(new java.util.ArrayList<>(merged));
            if (needsFlagUpdate) {
              upd.allowMessageSchedules(true);
            }
            jsm.updateStream(upd.build());
            if (needsFlagUpdate) {
              log.log(Level.INFO,
                  "Stream ''{0}'' updated to enable message scheduling (ADR-51).", streamName);
            }
            if (needsSubjectUpdate) {
              log.log(Level.INFO,
                  "Stream ''{0}'' updated with additional subjects: {1}.",
                  new Object[] {streamName, subjects});
            }
          }
        }
      } catch (IOException | JetStreamApiException e) {
        throw new RqueueNatsException(
            "Failed to ensure stream '" + streamName + "' for subjects " + subjects, e);
      }
      streamsDone.put(streamName, enableSchedules);
    }
  }

  /**
   * Ensure a durable pull consumer exists, returning the consumer name.
   * Hits the NATS backend at most once per (stream, consumer) pair per process lifetime.
   *
   * <p>Overload without a filter subject: used when the stream has only the work subject so the
   * filter would be redundant, or when multiple independent consumer groups (fan-out) must coexist
   * on a Limits-retention stream (NATS rejects two consumers with the same filter subject, error
   * 10100). For WorkQueue streams that also carry scheduler subjects ({@code .sched.*}) a filter
   * subject MUST be supplied via
   * {@link #ensureConsumer(String, String, String, Duration, long, long)} so the consumer only
   * receives work-subject messages and does not accidentally pick up scheduler entries.
   */
  public String ensureConsumer(
      String streamName,
      String consumerName,
      Duration ackWait,
      long maxDeliver,
      long maxAckPending) {
    return ensureConsumer(streamName, consumerName, null, ackWait, maxDeliver, maxAckPending);
  }

  /**
   * Ensure a durable pull consumer exists with an optional subject filter, returning the consumer
   * name. Hits the NATS backend at most once per (stream, consumer) pair per process lifetime.
   *
   * <p>{@code filterSubject} — when non-null, sets the consumer's filter subject so it only
   * receives messages published to that subject. Required when the stream carries both work subjects
   * and scheduler subjects ({@code .sched.*}): without a filter the consumer reads scheduler
   * entries before the scheduled time, delivering the message early. Pass {@code null} for streams
   * that do not use NATS scheduling, or where fan-out across multiple consumers is needed (Limits
   * retention).
   */
  public String ensureConsumer(
      String streamName,
      String consumerName,
      String filterSubject,
      Duration ackWait,
      long maxDeliver,
      long maxAckPending) {
    String cacheKey = streamName + "/" + consumerName;
    String cached = consumerCache.get(cacheKey);
    if (cached != null) {
      return cached;
    }
    Object lock = consumerLocks.computeIfAbsent(cacheKey, k -> new Object());
    synchronized (lock) {
      cached = consumerCache.get(cacheKey);
      if (cached != null) {
        return cached;
      }
      String actual = doEnsureConsumer(
          streamName, consumerName, filterSubject, ackWait, maxDeliver, maxAckPending);
      consumerCache.put(cacheKey, actual);
      return actual;
    }
  }

  private String doEnsureConsumer(
      String streamName,
      String consumerName,
      String filterSubject,
      Duration ackWait,
      long maxDeliver,
      long maxAckPending) {
    try {
      ConsumerInfo info = safeGetConsumerInfo(streamName, consumerName);
      if (info != null) {
        ConsumerConfiguration cc = info.getConsumerConfiguration();
        if (cc.getAckWait() != null && !cc.getAckWait().equals(ackWait)) {
          log.log(
              Level.WARNING,
              "Consumer " + streamName + "/" + consumerName
                  + " ackWait differs (existing=" + cc.getAckWait()
                  + ", desired=" + ackWait + ") - leaving existing config in place.");
        }
        if (cc.getMaxDeliver() != maxDeliver) {
          log.log(
              Level.WARNING,
              "Consumer " + streamName + "/" + consumerName
                  + " maxDeliver differs (existing=" + cc.getMaxDeliver()
                  + ", desired=" + maxDeliver + ") - leaving existing config in place.");
        }
        return consumerName;
      }
      if (!config.isAutoCreateConsumers()) {
        throw new RqueueNatsException("Consumer '" + consumerName + "' on stream '" + streamName
            + "' does not exist and autoCreateConsumers=false");
      }
      ConsumerConfiguration.Builder ccBuilder = ConsumerConfiguration.builder()
          .durable(consumerName)
          .ackPolicy(AckPolicy.Explicit)
          .deliverPolicy(DeliverPolicy.All)
          .ackWait(ackWait)
          .maxDeliver(maxDeliver)
          .maxAckPending(maxAckPending);
      if (filterSubject != null && !filterSubject.isEmpty()) {
        // Filter to the work subject only so that scheduler entries (published to
        // <workSubject>.sched.*) are not delivered to this consumer before the scheduled time.
        // The NATS scheduler fires the triggered message to workSubject when the time arrives.
        ccBuilder.filterSubject(filterSubject);
      }
      jsm.addOrUpdateConsumer(streamName, ccBuilder.build());
      return consumerName;
    } catch (JetStreamApiException e) {
      throw new RqueueNatsException(
          "Failed to ensure consumer '" + consumerName + "' on stream '" + streamName + "'", e);
    } catch (IOException e) {
      throw new RqueueNatsException(
          "Failed to ensure consumer '" + consumerName + "' on stream '" + streamName + "'", e);
    }
  }

  /** Ensure a DLQ stream exists capturing dead-letter subjects (e.g. "rqueue.js.*.dlq"). */
  public void ensureDlqStream(String dlqStreamName, List<String> dlqSubjects) {
    if (!config.isAutoCreateDlqStream()) {
      return;
    }
    ensureStream(dlqStreamName, dlqSubjects);
  }

  // ---- private helpers --------------------------------------------------

  private StreamInfo safeGetStreamInfo(String streamName)
      throws IOException, JetStreamApiException {
    try {
      return jsm.getStreamInfo(streamName);
    } catch (JetStreamApiException e) {
      // 10059 = stream not found
      if (e.getApiErrorCode() == 10059 || e.getErrorCode() == 404) {
        return null;
      }
      throw e;
    }
  }

  private ConsumerInfo safeGetConsumerInfo(String streamName, String consumerName)
      throws IOException, JetStreamApiException {
    try {
      return jsm.getConsumerInfo(streamName, consumerName);
    } catch (JetStreamApiException e) {
      // 10014 = consumer not found, 10059 = stream not found
      if (e.getApiErrorCode() == 10014 || e.getApiErrorCode() == 10059 || e.getErrorCode() == 404) {
        return null;
      }
      throw e;
    }
  }
}

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	package com.github.sonus21.rqueue.nats.internal;
11
12	import com.github.sonus21.rqueue.enums.QueueType;
13	import com.github.sonus21.rqueue.nats.RqueueNatsConfig;
14	import com.github.sonus21.rqueue.nats.RqueueNatsException;
15	import io.nats.client.Connection;
16	import io.nats.client.JetStreamApiException;
17	import io.nats.client.JetStreamManagement;
18	import io.nats.client.KeyValue;
19	import io.nats.client.KeyValueManagement;
20	import io.nats.client.api.AckPolicy;
21	import io.nats.client.api.CompressionOption;
22	import io.nats.client.api.ConsumerConfiguration;
23	import io.nats.client.api.ConsumerInfo;
24	import io.nats.client.api.DeliverPolicy;
25	import io.nats.client.api.KeyValueConfiguration;
26	import io.nats.client.api.KeyValueStatus;
27	import io.nats.client.api.RetentionPolicy;
28	import io.nats.client.api.StreamConfiguration;
29	import io.nats.client.api.StreamInfo;
30	import java.io.IOException;
31	import java.time.Duration;
32	import java.util.List;
33	import java.util.concurrent.ConcurrentHashMap;
34	import java.util.logging.Level;
35	import java.util.logging.Logger;
36
37	/**
38	* Idempotent stream/consumer/KV provisioning helpers. All methods are safe to call repeatedly;
39	* if the target object already exists with a compatible config, this class leaves it alone. If it
40	* exists but with diverging config (e.g. ackWait, maxDeliver), this class logs a WARN and does
41	* NOT auto-mutate so user customizations are preserved.
42	*/
43	public class NatsProvisioner {
44
45	private static final Logger log = Logger.getLogger(NatsProvisioner.class.getName());	1✔
46
47	private final Connection connection;
48	private final KeyValueManagement kvm;
49	private final JetStreamManagement jsm;
50	private final RqueueNatsConfig config;
51
52	private final ConcurrentHashMap<String, KeyValue> kvCache = new ConcurrentHashMap<>();	1✔
53	private final ConcurrentHashMap<String, Object> kvLocks = new ConcurrentHashMap<>();	1✔
54
55	// stream name → schedulingEnabled (true if the stream was created/updated with
56	// allowMessageSchedules)
57	private final ConcurrentHashMap<String, Boolean> streamsDone = new ConcurrentHashMap<>();	1✔
58	private final ConcurrentHashMap<String, Object> streamLocks = new ConcurrentHashMap<>();	1✔
59
60	// "streamName/requestedConsumerName" → actual consumer name (may differ for stale-rebind)
61	private final ConcurrentHashMap<String, String> consumerCache = new ConcurrentHashMap<>();	1✔
62	private final ConcurrentHashMap<String, Object> consumerLocks = new ConcurrentHashMap<>();	1✔
63
64	/**
65	* Minimum NATS server version that supports server-side message scheduling via the
66	* {@code Nats-Schedule} JetStream publish header (ADR-51).
67	*/
68	public static final String SCHEDULING_MIN_VERSION = "2.12.0";
69
70	private final boolean schedulingSupported;
71
72	public NatsProvisioner(Connection connection, JetStreamManagement jsm, RqueueNatsConfig config)
73	throws IOException {	1✔
74	this.connection = connection;	1✔
75	this.kvm = connection.keyValueManagement();	1✔
76	this.jsm = jsm;	1✔
77	this.config = config;	1✔
78	io.nats.client.api.ServerInfo serverInfo = connection.getServerInfo();	1✔
79	this.schedulingSupported =	1✔
80	serverInfo != null && serverInfo.isSameOrNewerThanVersion(SCHEDULING_MIN_VERSION);	1✔
81	log.log(	1✔
82	Level.INFO,
83	"NATS server version={0}; message scheduling (ADR-51) supported={1}",
84	new Object[] {serverInfo != null ? serverInfo.getVersion() : "unknown", schedulingSupported	1✔
85	});
86	}	1✔
87
88	/** Returns {@code true} when the connected NATS server supports message scheduling (>= 2.12). */
89	public boolean isMessageSchedulingSupported() {
90	return schedulingSupported;	1✔
91	}
92
93	// ---- KV provisioning --------------------------------------------------
94
95	/**
96	* Returns a {@link KeyValue} handle for {@code bucketName}, creating the bucket on first call.
97	* All buckets are created with S2 compression. A bucket-level TTL is applied at creation time
98	* only; existing buckets are reused as-is.
99	*
100	* @param bucketName NATS KV bucket name (e.g. {@code "rqueue-jobs"})
101	* @param ttl bucket-level max-age; {@code null} or non-positive = no TTL
102	*/
103	public KeyValue ensureKv(String bucketName, Duration ttl)
104	throws IOException, JetStreamApiException {
105	KeyValue cached = kvCache.get(bucketName);	1✔
106	if (cached != null) {	1✔
107	return cached;	1✔
108	}
109	Object lock = kvLocks.computeIfAbsent(bucketName, k -> new Object());	1✔
110	synchronized (lock) {	1✔
111	cached = kvCache.get(bucketName);	1✔
112	if (cached != null) {	1✔
113	return cached;	1✔
114	}
115	try {
116	KeyValueStatus status = kvm.getStatus(bucketName);	1✔
117	if (status != null) {	1!
118	KeyValue kv = connection.keyValue(bucketName);	1✔
119	kvCache.put(bucketName, kv);	1✔
120	return kv;	1✔
121	}
122	} catch (JetStreamApiException missing) {	1✔
123	// bucket absent — fall through to create
124	}	×
125	RqueueNatsConfig.StreamDefaults sd = config.getStreamDefaults();	1✔
126	KeyValueConfiguration.Builder cfg = KeyValueConfiguration.builder()	1✔
127	.name(bucketName)	1✔
128	.compression(true)	1✔
129	.replicas(sd.getReplicas())	1✔
130	.storageType(sd.getStorage());	1✔
131	if (ttl != null && !ttl.isZero() && !ttl.isNegative()) {	1!
132	cfg.ttl(ttl);	1✔
133	}
134	kvm.create(cfg.build());	1✔
135	KeyValue kv = connection.keyValue(bucketName);	1✔
136	kvCache.put(bucketName, kv);	1✔
137	return kv;	1✔
138	}
139	}
140
141	// ---- Stream provisioning ----------------------------------------------
142
143	/**
144	* Ensure a JetStream stream exists with the given subjects, using {@link QueueType#QUEUE}
145	* (WorkQueue retention) as the default. Callers that have a {@link QueueType} available should
146	* use {@link #ensureStream(String, List, QueueType)} instead.
147	*/
148	public void ensureStream(String streamName, List<String> subjects) {
149	ensureStream(streamName, subjects, QueueType.QUEUE, null, false);	1✔
150	}	1✔
151
152	/** See {@link #ensureStream(String, List, QueueType, String, boolean)}. */
153	public void ensureStream(String streamName, List<String> subjects, QueueType queueType) {
154	ensureStream(streamName, subjects, queueType, null, false);	1✔
155	}	1✔
156
157	/** See {@link #ensureStream(String, List, QueueType, String, boolean)}. */
158	public void ensureStream(
159	String streamName, List<String> subjects, QueueType queueType, String description) {
160	ensureStream(streamName, subjects, queueType, description, false);	1✔
161	}	1✔
162
163	/**
164	* Ensure a JetStream stream exists with the given subjects and retention policy derived from
165	* {@code queueType}:
166	* <ul>
167	* <li>{@link QueueType#QUEUE} — {@link io.nats.client.api.RetentionPolicy#WorkQueue}: each
168	* message is delivered to exactly one consumer; competing-consumer semantics.
169	* <li>{@link QueueType#STREAM} — {@link io.nats.client.api.RetentionPolicy#Limits}: every
170	* independent durable consumer group receives all messages; stream/fan-out semantics.
171	* </ul>
172	*
173	* <p>{@code description} is forwarded to JetStream as the stream's description (visible via
174	* {@code nats stream info}). Callers should pass the rqueue queue name so operators can map a
175	* stream back to the queue that created it; pass {@code null} to skip.
176	*
177	* <p>{@code allowSchedules} must be {@code true} when the stream will receive messages published
178	* with the {@code Nats-Schedule} header (ADR-51). Only callers that perform delayed
179	* enqueue should pass {@code true}; regular enqueue callers should pass {@code false} (or use the
180	* shorter overloads). Equivalent to the CLI flag {@code --allow-schedules}.
181	*
182	* <p>Hits the NATS backend at most once per stream name per process lifetime; subsequent calls
183	* return immediately from the in-process cache. If {@code allowSchedules=true} is later requested
184	* for a stream that was previously created without that flag, the stream is updated in place via
185	* {@link JetStreamManagement#updateStream}. If the stream already exists with a different
186	* retention policy, a WARNING is logged and the existing config is left untouched.
187	*/
188	public void ensureStream(
189	String streamName,
190	List<String> subjects,
191	QueueType queueType,
192	String description,
193	boolean allowSchedules) {
194	// Fast-path: already provisioned with at least as many capabilities as requested.
195	Boolean cached = streamsDone.get(streamName);	1✔
196	if (cached != null && (cached \|\| !allowSchedules)) {	1✔
197	return;	1✔
198	}
199	Object lock = streamLocks.computeIfAbsent(streamName, k -> new Object());	1✔
200	synchronized (lock) {	1✔
201	cached = streamsDone.get(streamName);	1✔
202	if (cached != null && (cached \|\| !allowSchedules)) {	1!
UNCOV 203	return;	×
204	}
205	boolean enableSchedules = allowSchedules && schedulingSupported;	1✔
206	try {
207	StreamInfo existing = safeGetStreamInfo(streamName);	1✔
208	RetentionPolicy desired =
209	queueType == QueueType.STREAM ? RetentionPolicy.Limits : RetentionPolicy.WorkQueue;	1✔
210	if (existing == null) {	1✔
211	if (!config.isAutoCreateStreams()) {	1✔
212	throw new RqueueNatsException(	1✔
213	"Stream '" + streamName + "' does not exist and autoCreateStreams=false");
214	}
215	RqueueNatsConfig.StreamDefaults sd = config.getStreamDefaults();	1✔
216	StreamConfiguration.Builder b = StreamConfiguration.builder()	1✔
217	.name(streamName)	1✔
218	.subjects(subjects)	1✔
219	.replicas(sd.getReplicas())	1✔
220	.storageType(sd.getStorage())	1✔
221	.retentionPolicy(desired)	1✔
222	.compressionOption(CompressionOption.S2);	1✔
223	if (enableSchedules) {	1✔
224	// Enable server-side message scheduling (ADR-51 / Nats-Schedule header).
225	// Equivalent to: nats stream add MY_STREAM --allow-schedules
226	b.allowMessageSchedules(true);	1✔
227	}
228	if (description != null && !description.isEmpty()) {	1!
229	b.description(description);	1✔
230	}
231	if (sd.getMaxMsgs() > 0) {	1!
232	b.maxMessages(sd.getMaxMsgs());	×
233	}
234	if (sd.getMaxBytes() > 0) {	1!
235	b.maxBytes(sd.getMaxBytes());	×
236	}
237	if (sd.getMaxAge() != null	1✔
238	&& !sd.getMaxAge().isZero()	1!
239	&& !sd.getMaxAge().isNegative()) {	1!
240	b.maxAge(sd.getMaxAge());	1✔
241	}
242	jsm.addStream(b.build());	1✔
243	} else {	1✔
244	RetentionPolicy actual = existing.getConfiguration().getRetentionPolicy();	1✔
245	if (actual != desired) {	1✔
246	log.log(	1✔
247	Level.WARNING,
248	"Stream ''{0}'' exists with retention={1} but queueMode requires retention={2}"
249	+ " — leaving existing config in place.",
250	new Object[] {streamName, actual, desired});
251	}
252	// Check whether new subjects need to be merged in (e.g. the sched wildcard added by
253	// enqueueWithDelay after the stream was originally created by a plain enqueue call).
254	java.util.List<String> existingSubjects = existing.getConfiguration().getSubjects();	1✔
255	java.util.Set<String> existingSet = existingSubjects != null	1!
256	? new java.util.HashSet<>(existingSubjects) : new java.util.HashSet<>();	1✔
257	boolean needsSubjectUpdate = subjects.stream().anyMatch(s -> !existingSet.contains(s));	1✔
258	boolean needsFlagUpdate =	1✔
259	enableSchedules && !existing.getConfiguration().getAllowMsgSchedules();	1!
260
261	if (needsFlagUpdate \|\| needsSubjectUpdate) {	1!
262	// Merge: keep all existing subjects and append new ones (never remove).
263	java.util.LinkedHashSet<String> merged = new java.util.LinkedHashSet<>(existingSet);	1✔
264	merged.addAll(subjects);	1✔
265	StreamConfiguration.Builder upd =	1✔
266	StreamConfiguration.builder(existing.getConfiguration())	1✔
267	.subjects(new java.util.ArrayList<>(merged));	1✔
268	if (needsFlagUpdate) {	1!
269	upd.allowMessageSchedules(true);	1✔
270	}
271	jsm.updateStream(upd.build());	1✔
272	if (needsFlagUpdate) {	1!
273	log.log(Level.INFO,	1✔
274	"Stream ''{0}'' updated to enable message scheduling (ADR-51).", streamName);
275	}
276	if (needsSubjectUpdate) {	1✔
277	log.log(Level.INFO,	1✔
278	"Stream ''{0}'' updated with additional subjects: {1}.",
279	new Object[] {streamName, subjects});
280	}
281	}
282	}
283	} catch (IOException \| JetStreamApiException e) {	1✔
284	throw new RqueueNatsException(	1✔
285	"Failed to ensure stream '" + streamName + "' for subjects " + subjects, e);
286	}	1✔
287	streamsDone.put(streamName, enableSchedules);	1✔
288	}	1✔
289	}	1✔
290
291	/**
292	* Ensure a durable pull consumer exists, returning the consumer name.
293	* Hits the NATS backend at most once per (stream, consumer) pair per process lifetime.
294	*
295	* <p>Overload without a filter subject: used when the stream has only the work subject so the
296	* filter would be redundant, or when multiple independent consumer groups (fan-out) must coexist
297	* on a Limits-retention stream (NATS rejects two consumers with the same filter subject, error
298	* 10100). For WorkQueue streams that also carry scheduler subjects ({@code .sched.*}) a filter
299	* subject MUST be supplied via
300	* {@link #ensureConsumer(String, String, String, Duration, long, long)} so the consumer only
301	* receives work-subject messages and does not accidentally pick up scheduler entries.
302	*/
303	public String ensureConsumer(
304	String streamName,
305	String consumerName,
306	Duration ackWait,
307	long maxDeliver,
308	long maxAckPending) {
309	return ensureConsumer(streamName, consumerName, null, ackWait, maxDeliver, maxAckPending);	1✔
310	}
311
312	/**
313	* Ensure a durable pull consumer exists with an optional subject filter, returning the consumer
314	* name. Hits the NATS backend at most once per (stream, consumer) pair per process lifetime.
315	*
316	* <p>{@code filterSubject} — when non-null, sets the consumer's filter subject so it only
317	* receives messages published to that subject. Required when the stream carries both work subjects
318	* and scheduler subjects ({@code .sched.*}): without a filter the consumer reads scheduler
319	* entries before the scheduled time, delivering the message early. Pass {@code null} for streams
320	* that do not use NATS scheduling, or where fan-out across multiple consumers is needed (Limits
321	* retention).
322	*/
323	public String ensureConsumer(
324	String streamName,
325	String consumerName,
326	String filterSubject,
327	Duration ackWait,
328	long maxDeliver,
329	long maxAckPending) {
330	String cacheKey = streamName + "/" + consumerName;	1✔
331	String cached = consumerCache.get(cacheKey);	1✔
332	if (cached != null) {	1✔
333	return cached;	1✔
334	}
335	Object lock = consumerLocks.computeIfAbsent(cacheKey, k -> new Object());	1✔
336	synchronized (lock) {	1✔
337	cached = consumerCache.get(cacheKey);	1✔
338	if (cached != null) {	1!
UNCOV 339	return cached;	×
340	}
341	String actual = doEnsureConsumer(	1✔
342	streamName, consumerName, filterSubject, ackWait, maxDeliver, maxAckPending);
343	consumerCache.put(cacheKey, actual);	1✔
344	return actual;	1✔
345	}
346	}
347
348	private String doEnsureConsumer(
349	String streamName,
350	String consumerName,
351	String filterSubject,
352	Duration ackWait,
353	long maxDeliver,
354	long maxAckPending) {
355	try {
356	ConsumerInfo info = safeGetConsumerInfo(streamName, consumerName);	1✔
357	if (info != null) {	1✔
358	ConsumerConfiguration cc = info.getConsumerConfiguration();	1✔
359	if (cc.getAckWait() != null && !cc.getAckWait().equals(ackWait)) {	1!
UNCOV 360	log.log(	×
361	Level.WARNING,
362	"Consumer " + streamName + "/" + consumerName
UNCOV 363	+ " ackWait differs (existing=" + cc.getAckWait()	×
364	+ ", desired=" + ackWait + ") - leaving existing config in place.");
365	}
366	if (cc.getMaxDeliver() != maxDeliver) {	1!
367	log.log(	×
368	Level.WARNING,
369	"Consumer " + streamName + "/" + consumerName
UNCOV 370	+ " maxDeliver differs (existing=" + cc.getMaxDeliver()	×
371	+ ", desired=" + maxDeliver + ") - leaving existing config in place.");
372	}
373	return consumerName;	1✔
374	}
375	if (!config.isAutoCreateConsumers()) {	1!
UNCOV 376	throw new RqueueNatsException("Consumer '" + consumerName + "' on stream '" + streamName	×
377	+ "' does not exist and autoCreateConsumers=false");
378	}
379	ConsumerConfiguration.Builder ccBuilder = ConsumerConfiguration.builder()	1✔
380	.durable(consumerName)	1✔
381	.ackPolicy(AckPolicy.Explicit)	1✔
382	.deliverPolicy(DeliverPolicy.All)	1✔
383	.ackWait(ackWait)	1✔
384	.maxDeliver(maxDeliver)	1✔
385	.maxAckPending(maxAckPending);	1✔
386	if (filterSubject != null && !filterSubject.isEmpty()) {	1!
387	// Filter to the work subject only so that scheduler entries (published to
388	// <workSubject>.sched.*) are not delivered to this consumer before the scheduled time.
389	// The NATS scheduler fires the triggered message to workSubject when the time arrives.
390	ccBuilder.filterSubject(filterSubject);	1✔
391	}
392	jsm.addOrUpdateConsumer(streamName, ccBuilder.build());	1✔
393	return consumerName;	1✔
NEW 394	} catch (JetStreamApiException e) {	×
NEW 395	throw new RqueueNatsException(	×
396	"Failed to ensure consumer '" + consumerName + "' on stream '" + streamName + "'", e);
397	} catch (IOException e) {	×
398	throw new RqueueNatsException(	×
399	"Failed to ensure consumer '" + consumerName + "' on stream '" + streamName + "'", e);
400	}
401	}
402
403	/** Ensure a DLQ stream exists capturing dead-letter subjects (e.g. "rqueue.js..dlq"). /
404	public void ensureDlqStream(String dlqStreamName, List<String> dlqSubjects) {
405	if (!config.isAutoCreateDlqStream()) {	1✔
406	return;	1✔
407	}
408	ensureStream(dlqStreamName, dlqSubjects);	1✔
409	}	1✔
410
411	// ---- private helpers --------------------------------------------------
412
413	private StreamInfo safeGetStreamInfo(String streamName)
414	throws IOException, JetStreamApiException {
415	try {
416	return jsm.getStreamInfo(streamName);	1✔
417	} catch (JetStreamApiException e) {	1✔
418	// 10059 = stream not found
419	if (e.getApiErrorCode() == 10059 \|\| e.getErrorCode() == 404) {	1!
420	return null;	1✔
421	}
UNCOV 422	throw e;	×
423	}
424	}
425
426	private ConsumerInfo safeGetConsumerInfo(String streamName, String consumerName)
427	throws IOException, JetStreamApiException {
428	try {
429	return jsm.getConsumerInfo(streamName, consumerName);	1✔
430	} catch (JetStreamApiException e) {	1✔
431	// 10014 = consumer not found, 10059 = stream not found
432	if (e.getApiErrorCode() == 10014 \|\| e.getApiErrorCode() == 10059 \|\| e.getErrorCode() == 404) {	1!
433	return null;	1✔
434	}
UNCOV 435	throw e;	×
436	}
437	}
438	}

sonus21 / rqueue / 25606310187

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