Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Configure QueueBrowser's pulsar reader with subscription name #118

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
Open

Configure QueueBrowser's pulsar reader with subscription name #118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
20f1f83
Configure QueueBrowser's pulsar reader with subscription name
michaeljmarshall Apr 5, 2024
e2fd831
Add test coverage
michaeljmarshall Apr 5, 2024
51154cf
Merge remote-tracking branch 'upstream/master' into configure-queue-b…
michaeljmarshall Apr 15, 2024
45f6ffb
Make test compile
michaeljmarshall Apr 15, 2024
20341cf
Add jms.queueBrowserSubscriptionName config; make test pass
michaeljmarshall Apr 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,8 @@ public class PulsarConnectionFactory
private transient String lastConnectUsername = null;
private transient String lastConnectPassword = null;
private transient String queueSubscriptionName = "jms-queue";
// The queue browser subscription name is generated as reader-XYZ where XYZ is 10 random characters.
private transient String queueBrowserSubscriptionName = null;
private transient SubscriptionType topicSharedSubscriptionType = SubscriptionType.Shared;
private transient long waitForServerStartupTimeout = 60000;
private transient boolean usePulsarAdmin = true;
Expand Down Expand Up @@ -325,6 +327,9 @@ private synchronized void ensureInitialized(String connectUsername, String conne
this.queueSubscriptionName =
getAndRemoveString("jms.queueSubscriptionName", "jms-queue", configurationCopy);

this.queueBrowserSubscriptionName =
getAndRemoveString("jms.queueBrowserSubscriptionName", null, configurationCopy);

this.usePulsarAdmin =
Boolean.parseBoolean(getAndRemoveString("jms.usePulsarAdmin", "true", configurationCopy));

Expand Down Expand Up @@ -1541,6 +1546,9 @@ private Reader<?> createReaderForBrowserForNonPartitionedTopic(
.startMessageId(seekMessageId)
.startMessageIdInclusive()
.topic(fullQualifiedTopicName);
if (queueBrowserSubscriptionName != null) {
builder.subscriptionName(queueBrowserSubscriptionName);
}
Reader<?> newReader = builder.create();
readers.add(newReader);
return newReader;
Expand Down
10 changes: 8 additions & 2 deletions pulsar-jms/src/test/java/com/datastax/oss/pulsar/jms/QueueTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,14 +135,16 @@ public void sendJMSRedeliveryCountTest() throws Exception {
@Test
public void testQueueBrowsers() throws Exception {
int numMessages = 20;
String expectedSubscriptionName = "my-queue-browser-subscription";
Map<String, Object> properties = pulsarContainer.buildJMSConnectionProperties();
properties.put("jms.enableClientSideEmulation", "false");
properties.put("jms.queueBrowserSubscriptionName", expectedSubscriptionName);
try (PulsarConnectionFactory factory = new PulsarConnectionFactory(properties); ) {
try (Connection connection = factory.createConnection()) {
connection.start();
try (Session session = connection.createSession(); ) {
Queue destination =
session.createQueue("persistent://public/default/test-" + UUID.randomUUID());
String topicName = "persistent://public/default/test-" + UUID.randomUUID();
Queue destination = session.createQueue(topicName);

try (MessageProducer producer = session.createProducer(destination); ) {
for (int i = 0; i < numMessages; i++) {
Expand Down Expand Up @@ -183,6 +185,10 @@ public void testQueueBrowsers() throws Exception {
count++;
}
assertEquals(1, count);
TopicStats stats = pulsarContainer.getAdmin().topics().getStats(topicName);
// Validate QueueBrowser is connected using subscription name
assertEquals(
stats.getSubscriptions().get(expectedSubscriptionName).getConsumers().size(), 1);
}

// scan again without calling hasMoreElements explicitly
Expand Down
Loading