38966765975

Committed 16 Oct 2026 09:04PM UTC coverage: 89.313% (-0.2%) from 89.5%

Build # 38966765975

Build Type

push

github

Committed by

scottf

Commit Message

Move push and pull subscribe out of JetStream into JsPushSubscribe and JsPullSubscribe

- The 15 pushSubscribe and 15 pullSubscribe overloads move to JsPushSubscribe and JsPullSubscribe (io.synadia.client.impl), each as an instance method (the constructor takes the JetStream) and as a static method (JetStream first parameter).
- KV/OS watch, OS get and the tests call the static form.
- README: new "Consuming from JetStream" section; the simplified API (ConsumerContext) is the preferred way to consume, the two classes exist only to support legacy behavior. jetstream README and migration guide updated.

Coverage Stats

4689 of 5499 branches covered (85.27%)

Branch coverage included in aggregate %.

71 of 110 new or added lines in 5 files covered. (64.55%)

1 existing line in 1 file now uncovered.

11649 of 12794 relevant lines covered (91.05%)

4.03 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 = JsPushSubscribe.pushSubscribe(js, streamName, creator);
        }
        else {
            PushConsumerCreator creator = new PushConsumerCreator()
                .ackPolicy(AckPolicy.None)
                .deliverPolicy(deliverPolicy)
                .headersOnly(headersOnly)
                .filterSubjects(subjects);
            sub = JsPushSubscribe.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 = JsPushSubscribe.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 = JsPushSubscribe.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 / 38966765975

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