This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 56cb7e3f0496cd757726cadbdb3065ab9729daea
Author: Andrea Cosentino <anco...@gmail.com>
AuthorDate: Fri Sep 20 12:37:37 2019 +0200
CAMEL-14001 - Camel-Pulsar: Add MessageRoutingMode and MessageRouter option
to producer, messageRouter option
---
.../src/main/docs/pulsar-component.adoc | 3 ++-
.../camel/component/pulsar/PulsarProducer.java | 9 ++++++--
.../pulsar/configuration/PulsarConfiguration.java | 14 +++++++++++++
.../endpoint/dsl/PulsarEndpointBuilderFactory.java | 24 ++++++++++++++++++++++
4 files changed, 47 insertions(+), 3 deletions(-)
diff --git a/components/camel-pulsar/src/main/docs/pulsar-component.adoc
b/components/camel-pulsar/src/main/docs/pulsar-component.adoc
index fa40b34..3e19465 100644
--- a/components/camel-pulsar/src/main/docs/pulsar-component.adoc
+++ b/components/camel-pulsar/src/main/docs/pulsar-component.adoc
@@ -69,7 +69,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (26 parameters):
+=== Query Parameters (27 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -96,6 +96,7 @@ with the following path and query parameters:
| *lazyStartProducer* (producer) | Whether the producer should be started lazy
(on the first message). By starting lazy you can use this to allow CamelContext
and routes to startup in situations where a producer may otherwise fail during
starting and cause the route to fail being started. By deferring this startup
to be lazy then the startup failure can be handled during routing messages via
Camel's routing error handlers. Beware that when the first message is processed
then creating and [...]
| *maxPendingMessages* (producer) | Set the max size of the queue holding the
messages pending to receive an acknowledgment from the broker. Default is 1000.
| 1000 | int
| *maxPendingMessagesAcross Partitions* (producer) | Set the number of max
pending messages across all the partitions. Default is 50000. | 50000 | int
+| *messageRouter* (producer) | Set a custom Message Router. | | MessageRouter
| *messageRoutingMode* (producer) | Set the message routing mode for the
producer. | RoundRobinPartition | MessageRoutingMode
| *producerName* (producer) | Name of the producer | default-producer | String
| *sendTimeoutMs* (producer) | Send timeout in milliseconds. Defaults to
30,000ms (30 seconds) | 30000 | int
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 f4a5646..d7c8b0a 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
@@ -25,6 +25,7 @@ import org.apache.camel.TypeConversionException;
import org.apache.camel.component.pulsar.configuration.PulsarConfiguration;
import org.apache.camel.component.pulsar.utils.message.PulsarMessageUtils;
import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
@@ -73,8 +74,12 @@ public class PulsarProducer extends DefaultProducer {
.batchingMaxMessages(configuration.getMaxPendingMessages())
.enableBatching(configuration.isBatchingEnabled())
.initialSequenceId(configuration.getInitialSequenceId())
- .compressionType(configuration.getCompressionType())
- .messageRoutingMode(configuration.getMessageRoutingMode());
+ .compressionType(configuration.getCompressionType());
+ if (ObjectHelper.isNotEmpty(configuration.getMessageRouter())) {
+
producerBuilder.messageRouter(configuration.getMessageRouter());
+ } else {
+
producerBuilder.messageRoutingMode(configuration.getMessageRoutingMode());
+ }
producer = producerBuilder.create();
}
}
diff --git
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/configuration/PulsarConfiguration.java
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/configuration/PulsarConfiguration.java
index 4366166..ea34312 100644
---
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/configuration/PulsarConfiguration.java
+++
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/configuration/PulsarConfiguration.java
@@ -23,6 +23,7 @@ import
org.apache.camel.component.pulsar.utils.consumers.SubscriptionType;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
import org.apache.pulsar.client.api.CompressionType;
+import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.MessageRoutingMode;
import static
org.apache.camel.component.pulsar.utils.consumers.SubscriptionType.EXCLUSIVE;
@@ -72,6 +73,8 @@ public class PulsarConfiguration {
private CompressionType compressionType = CompressionType.NONE;
@UriParam(label = "producer", description = "MessageRoutingMode",
defaultValue = "RoundRobinPartition")
private MessageRoutingMode messageRoutingMode =
MessageRoutingMode.RoundRobinPartition;
+ @UriParam(label = "producer", description = "Custom Message Router")
+ private MessageRouter messageRouter;
public String getSubscriptionName() {
return subscriptionName;
@@ -319,6 +322,17 @@ public class PulsarConfiguration {
public void setMessageRoutingMode(MessageRoutingMode
messageRoutingMode) {
this.messageRoutingMode = messageRoutingMode;
}
+
+ /**
+ * Set a custom Message Router.
+ */
+ public MessageRouter getMessageRouter() {
+ return messageRouter;
+ }
+
+ public void setMessageRouter(MessageRouter messageRouter) {
+ this.messageRouter = messageRouter;
+ }
}
diff --git
a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/PulsarEndpointBuilderFactory.java
b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/PulsarEndpointBuilderFactory.java
index 943e2f1..ace4f65 100644
---
a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/PulsarEndpointBuilderFactory.java
+++
b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/PulsarEndpointBuilderFactory.java
@@ -667,6 +667,30 @@ public interface PulsarEndpointBuilderFactory {
return this;
}
/**
+ * Set a custom Message Router.
+ *
+ * The option is a:
+ * <code>org.apache.pulsar.client.api.MessageRouter</code> type.
+ *
+ * Group: producer
+ */
+ default PulsarEndpointProducerBuilder messageRouter(Object
messageRouter) {
+ doSetProperty("messageRouter", messageRouter);
+ return this;
+ }
+ /**
+ * Set a custom Message Router.
+ *
+ * The option will be converted to a
+ * <code>org.apache.pulsar.client.api.MessageRouter</code> type.
+ *
+ * Group: producer
+ */
+ default PulsarEndpointProducerBuilder messageRouter(String
messageRouter) {
+ doSetProperty("messageRouter", messageRouter);
+ return this;
+ }
+ /**
* Set the message routing mode for the producer.
*
* The option is a:
