You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2019/07/05 10:20:49 UTC
[camel] 01/02: CAMEL-13718 - fixed wrong syntax for pulsar endpoints
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-2.24.x
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 530b7c0d01b494e258ecfda48b2d9d6c3b94d142
Author: Lars Heinemann <lh...@gmail.com>
AuthorDate: Tue Jul 2 17:54:27 2019 +0200
CAMEL-13718 - fixed wrong syntax for pulsar endpoints
Signed-off-by: Lars Heinemann <lh...@gmail.com>
---
.../java/org/apache/camel/component/pulsar/PulsarEndpoint.java | 10 +++++-----
.../java/org/apache/camel/component/pulsar/PulsarProducer.java | 6 +++---
.../pulsar/utils/consumers/CommonCreationStrategyImpl.java | 2 +-
3 files changed, 9 insertions(+), 9 deletions(-)
diff --git a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarEndpoint.java b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarEndpoint.java
index bc6b9c2..67574e9 100644
--- a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarEndpoint.java
+++ b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarEndpoint.java
@@ -28,18 +28,18 @@ import org.apache.camel.spi.UriPath;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
-@UriEndpoint(scheme = "pulsar", title = "Apache Pulsar", syntax = "pulsar:persistence://tenant/namespace/topic", label = "messaging")
+@UriEndpoint(scheme = "pulsar", firstVersion = "2.24.0", title = "Apache Pulsar", syntax = "pulsar:uri", label = "messaging")
public class PulsarEndpoint extends DefaultEndpoint {
private PulsarClient pulsarClient;
@UriParam
private PulsarConfiguration pulsarConfiguration;
@UriPath(label = "consumer,producer", description = "The Topic's full URI path including type, tenant and namespace")
- private final String topic;
+ private final String topicUri;
public PulsarEndpoint(String uri, String path, PulsarConfiguration pulsarConfiguration, PulsarComponent component, PulsarClient pulsarClient) throws PulsarClientException {
super(uri, component);
- this.topic = path;
+ this.topicUri = path;
this.pulsarConfiguration = pulsarConfiguration;
this.pulsarClient = pulsarClient;
}
@@ -85,8 +85,8 @@ public class PulsarEndpoint extends DefaultEndpoint {
return pulsarConfiguration;
}
- public String getTopic() {
- return topic;
+ public String getTopicUri() {
+ return topicUri;
}
}
diff --git a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarProducer.java b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarProducer.java
index 370890d..06ac5e7 100644
--- a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarProducer.java
+++ b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/PulsarProducer.java
@@ -54,16 +54,16 @@ public class PulsarProducer extends DefaultProducer {
private synchronized void createProducer() throws org.apache.pulsar.client.api.PulsarClientException {
if (producer == null) {
- final String topic = pulsarEndpoint.getTopic();
+ final String topicUri = pulsarEndpoint.getTopicUri();
String producerName = pulsarEndpoint.getPulsarConfiguration().getProducerName();
if (producerName == null) {
- producerName = topic + "-" + Thread.currentThread().getId();
+ producerName = topicUri + "-" + Thread.currentThread().getId();
}
final ProducerBuilder<byte[]> producerBuilder = pulsarEndpoint
.getPulsarClient()
.newProducer()
.producerName(producerName)
- .topic(topic);
+ .topic(topicUri);
producer = producerBuilder.create();
}
}
diff --git a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/CommonCreationStrategyImpl.java b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/CommonCreationStrategyImpl.java
index b54817f..f8e6ed1 100644
--- a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/CommonCreationStrategyImpl.java
+++ b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/CommonCreationStrategyImpl.java
@@ -30,7 +30,7 @@ public final class CommonCreationStrategyImpl {
public static ConsumerBuilder<byte[]> create(final String name, final PulsarEndpoint pulsarEndpoint, final PulsarConsumer pulsarConsumer) {
final PulsarConfiguration endpointConfiguration = pulsarEndpoint.getPulsarConfiguration();
- return pulsarEndpoint.getPulsarClient().newConsumer().topic(pulsarEndpoint.getTopic()).subscriptionName(endpointConfiguration.getSubscriptionName())
+ return pulsarEndpoint.getPulsarClient().newConsumer().topic(pulsarEndpoint.getTopicUri()).subscriptionName(endpointConfiguration.getSubscriptionName())
.receiverQueueSize(endpointConfiguration.getConsumerQueueSize()).consumerName(name)
.messageListener(new PulsarMessageListener(pulsarEndpoint, pulsarConsumer.getExceptionHandler(), pulsarConsumer.getProcessor()));
}