38985951354

Committed 17 Oct 2026 12:20AM UTC coverage: 89.51% (+0.2%) from 89.313%

Build # 38985951354

Build Type

push

github

Committed by

scottf

Commit Message

Rename JsPushSubscribe / JsPullSubscribe to JetStreamPushSubscribe / JetStreamPullSubscribe; test both call forms

- Classes renamed with git mv; every caller and doc updated.
- Redundant @NonNull removed; the classes are @NullMarked.
- New test enum JsSubscribeForm (INSTANCE / STATIC). The JetStreamSubscribeTests subscribe tests are parameterized on it, so every overload runs through the instance methods and the static methods; jacoco shows all 31 methods of each class covered.

Coverage Stats

4689 of 5499 branches covered (85.27%)

Branch coverage included in aggregate %.

3 of 6 new or added lines in 5 files covered. (50.0%)

11685 of 12794 relevant lines covered (91.33%)

4.04 hits per line

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

0.0

/jetstream/src/main/java/io/synadia/client/impl/AbstractBucketFeature.java

package io.synadia.client.impl;

import io.synadia.client.Message;
import io.synadia.client.MessageHandler;
import io.synadia.client.api.*;
import io.synadia.client.utils.JsValidator;

import java.util.Collections;
import java.util.List;

import static io.synadia.client.impl.JetStreamConstants.JS_NO_MESSAGE_FOUND_ERR;

/**
 * Base class for the features that are built on top of a stream and present it as a named bucket,
 * such as key value and object store. Holds the connection and JetStream contexts they all need
 * and works out the stream name that backs the bucket.
 */
public abstract class AbstractBucketFeature {

    /** The connection the feature works over. */
    public final NatsConnection nc;

    /** The options the feature was created with, null if it was created with defaults. */
    public final FeatureOptions fo;

    /** JetStream context used for the feature's message traffic. */
    public final JetStream js;

    /** JetStream management context used for the feature's stream and consumer administration. */
    public final JetStreamManagement jsm;

    /** The bucket name, already validated. */
    public final String bucketName;

    /** The name of the stream backing the bucket, derived from the bucket name. */
    public final String streamName;

    protected AbstractBucketFeature(String bucketName, NatsConnection connection, FeatureOptions fo) {
        this.nc = connection;
        this.fo = fo;
        if (fo == null) {
            this.js = new JetStream(connection, null);
            this.jsm = new JetStreamManagement(connection, null);
        }
        else {
            this.js = new JetStream(connection, fo.getJetStreamOptions());
            this.jsm = new JetStreamManagement(connection, fo.getJetStreamOptions());
        }
        this.bucketName = JsValidator.validateBucketName(bucketName, true);
        this.streamName = toStreamName(bucketName);
    }

    protected AbstractBucketFeature(String bucketName, AbstractBucketFeature existing) {
        this.nc = existing.nc;
        this.fo = existing.fo;
        this.js = existing.js;
        this.jsm = existing.jsm;
        this.bucketName = JsValidator.validateBucketName(bucketName, true);
        this.streamName = toStreamName(bucketName);
    }

    abstract protected String toStreamName(String bucketName);

    /**
     * The name of the bucket, as supplied when the feature was created.
     * @return the bucket name
     */
    public String getBucketName() {
        return bucketName;
    }

    /**
     * The name of the stream that holds the bucket's data. The feature derives it from the bucket
     * name, so it is not the same string.
     * @return the stream name
     */
    public String getStreamName() {
        return streamName;
    }

    protected MessageInfo _getLast(String subject) throws JetStreamException, InterruptedException {
        try {
            return jsm.getLastMessage(streamName, subject);
        }
        catch (JetStreamApiException jsae) {
            if (jsae.getApiErrorCode() == JS_NO_MESSAGE_FOUND_ERR) {
                return null;
            }
            throw jsae;
        }
    }

    protected MessageInfo _getBySeq(long seq) throws JetStreamException, InterruptedException {
        try {
            return jsm.getMessage(streamName, seq);
        }
        catch (JetStreamApiException jsae) {
            if (jsae.getApiErrorCode() == JS_NO_MESSAGE_FOUND_ERR) {
                return null;
            }
            throw jsae;
        }
    }

    protected void visitSubject(String subject, DeliverPolicy deliverPolicy, boolean headersOnly, boolean ordered, MessageHandler handler) throws JetStreamException, InterruptedException {
        visitSubject(Collections.singletonList(subject), deliverPolicy, headersOnly, ordered, handler);
    }

    protected void visitSubject(List<String> subjects, DeliverPolicy deliverPolicy, boolean headersOnly, boolean ordered, MessageHandler handler) throws JetStreamException, InterruptedException {
        JetStreamPushSubscription sub;
        if (ordered) {
            PushOrderedConsumerCreator creator = new PushOrderedConsumerCreator()
                .deliverPolicy(deliverPolicy)
                .headersOnly(headersOnly)
                .filterSubjects(subjects);
            sub = JetStreamPushSubscribe.pushSubscribe(js, streamName, creator);
        }
        else {
            PushConsumerCreator creator = new PushConsumerCreator()
                .ackPolicy(AckPolicy.None)
                .deliverPolicy(deliverPolicy)
                .headersOnly(headersOnly)
                .filterSubjects(subjects);
            sub = JetStreamPushSubscribe.pushSubscribe(js, streamName, creator);
        }

        long timeoutMillis = js.getTimeout();
        try {
            long pending = sub.getConsumerInfo().getCalculatedPending();
            while (pending > 0) { // no need to loop if nothing pending
                Message m = sub.nextMessage(timeoutMillis);
                if (m == null) {
                    return; // if there are no messages by the timeout, we are done.
                }
                handler.onMessage(m);
                if (--pending == 0) {
                    return;
                }
            }
        }
        finally {
            sub.unsubscribe();
        }
    }
}

1	package io.synadia.client.impl;
2
3	import io.synadia.client.Message;
4	import io.synadia.client.MessageHandler;
5	import io.synadia.client.api.*;
6	import io.synadia.client.utils.JsValidator;
7
8	import java.util.Collections;
9	import java.util.List;
10
11	import static io.synadia.client.impl.JetStreamConstants.JS_NO_MESSAGE_FOUND_ERR;
12
13	/**
14	* Base class for the features that are built on top of a stream and present it as a named bucket,
15	* such as key value and object store. Holds the connection and JetStream contexts they all need
16	* and works out the stream name that backs the bucket.
17	*/
18	public abstract class AbstractBucketFeature {
19
20	/** The connection the feature works over. */
21	public final NatsConnection nc;
22
23	/** The options the feature was created with, null if it was created with defaults. */
24	public final FeatureOptions fo;
25
26	/** JetStream context used for the feature's message traffic. */
27	public final JetStream js;
28
29	/** JetStream management context used for the feature's stream and consumer administration. */
30	public final JetStreamManagement jsm;
31
32	/** The bucket name, already validated. */
33	public final String bucketName;
34
35	/** The name of the stream backing the bucket, derived from the bucket name. */
36	public final String streamName;
37
38	protected AbstractBucketFeature(String bucketName, NatsConnection connection, FeatureOptions fo) {	×
39	this.nc = connection;	×
40	this.fo = fo;	×
41	if (fo == null) {	×
42	this.js = new JetStream(connection, null);	×
43	this.jsm = new JetStreamManagement(connection, null);	×
44	}
45	else {
46	this.js = new JetStream(connection, fo.getJetStreamOptions());	×
47	this.jsm = new JetStreamManagement(connection, fo.getJetStreamOptions());	×
48	}
49	this.bucketName = JsValidator.validateBucketName(bucketName, true);	×
50	this.streamName = toStreamName(bucketName);	×
51	}	×
52
53	protected AbstractBucketFeature(String bucketName, AbstractBucketFeature existing) {	×
54	this.nc = existing.nc;	×
55	this.fo = existing.fo;	×
56	this.js = existing.js;	×
57	this.jsm = existing.jsm;	×
58	this.bucketName = JsValidator.validateBucketName(bucketName, true);	×
59	this.streamName = toStreamName(bucketName);	×
60	}	×
61
62	abstract protected String toStreamName(String bucketName);
63
64	/**
65	* The name of the bucket, as supplied when the feature was created.
66	* @return the bucket name
67	*/
68	public String getBucketName() {
69	return bucketName;	×
70	}
71
72	/**
73	* The name of the stream that holds the bucket's data. The feature derives it from the bucket
74	* name, so it is not the same string.
75	* @return the stream name
76	*/
77	public String getStreamName() {
78	return streamName;	×
79	}
80
81	protected MessageInfo _getLast(String subject) throws JetStreamException, InterruptedException {
82	try {
83	return jsm.getLastMessage(streamName, subject);	×
84	}
85	catch (JetStreamApiException jsae) {	×
86	if (jsae.getApiErrorCode() == JS_NO_MESSAGE_FOUND_ERR) {	×
87	return null;	×
88	}
89	throw jsae;	×
90	}
91	}
92
93	protected MessageInfo _getBySeq(long seq) throws JetStreamException, InterruptedException {
94	try {
95	return jsm.getMessage(streamName, seq);	×
96	}
97	catch (JetStreamApiException jsae) {	×
98	if (jsae.getApiErrorCode() == JS_NO_MESSAGE_FOUND_ERR) {	×
99	return null;	×
100	}
101	throw jsae;	×
102	}
103	}
104
105	protected void visitSubject(String subject, DeliverPolicy deliverPolicy, boolean headersOnly, boolean ordered, MessageHandler handler) throws JetStreamException, InterruptedException {
106	visitSubject(Collections.singletonList(subject), deliverPolicy, headersOnly, ordered, handler);	×
107	}	×
108
109	protected void visitSubject(List<String> subjects, DeliverPolicy deliverPolicy, boolean headersOnly, boolean ordered, MessageHandler handler) throws JetStreamException, InterruptedException {
110	JetStreamPushSubscription sub;
111	if (ordered) {	×
112	PushOrderedConsumerCreator creator = new PushOrderedConsumerCreator()	×
113	.deliverPolicy(deliverPolicy)	×
114	.headersOnly(headersOnly)	×
115	.filterSubjects(subjects);	×
NEW 116	sub = JetStreamPushSubscribe.pushSubscribe(js, streamName, creator);	×
117	}	×
118	else {
119	PushConsumerCreator creator = new PushConsumerCreator()	×
120	.ackPolicy(AckPolicy.None)	×
121	.deliverPolicy(deliverPolicy)	×
122	.headersOnly(headersOnly)	×
123	.filterSubjects(subjects);	×
NEW 124	sub = JetStreamPushSubscribe.pushSubscribe(js, streamName, creator);	×
125	}
126
127	long timeoutMillis = js.getTimeout();	×
128	try {
129	long pending = sub.getConsumerInfo().getCalculatedPending();	×
130	while (pending > 0) { // no need to loop if nothing pending	×
131	Message m = sub.nextMessage(timeoutMillis);	×
132	if (m == null) {	×
133	return; // if there are no messages by the timeout, we are done.	×
134	}
135	handler.onMessage(m);	×
136	if (--pending == 0) {	×
137	return;	×
138	}
139	}	×
140	}
141	finally {
142	sub.unsubscribe();	×
143	}
144	}	×
145	}

synadia-io / nats.java.v3 / 38985951354

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