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/12/04 09:12:36 UTC
[camel] branch master updated (199b167 -> 1755297)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 199b167 camel-nats - Let producer use byte[] if possible instead of string as first choice
new 6f2ad84 CAMEL-14253: camel-nats - Configure brokers on component level
new ec71087 Regen
new 725fed6 CAMEL-14253: camel-nats - Configure brokers on component level
new 1755297 CAMEL-14253: camel-nats - Configure brokers on component level
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../camel-nats/src/main/docs/nats-component.adoc | 87 +++++++---
.../apache/camel/component/nats/NatsComponent.java | 18 +-
.../camel/component/nats/NatsConfiguration.java | 20 +--
.../apache/camel/component/nats/NatsConsumer.java | 1 -
.../apache/camel/component/nats/NatsEndpoint.java | 6 +-
.../apache/camel/component/nats/NatsProducer.java | 1 +
.../component/nats/NatsAuthConsumerLoadTest.java | 4 +-
.../camel/component/nats/NatsAuthProducerTest.java | 3 +-
.../camel/component/nats/NatsAuthTestSupport.java | 13 +-
.../nats/NatsAuthTokenConsumerLoadTest.java | 4 +-
.../component/nats/NatsAuthTokenProducerTest.java | 3 +-
.../component/nats/NatsAuthTokenTestSupport.java | 14 +-
.../camel/component/nats/NatsConsumerLoadTest.java | 9 +-
.../nats/NatsConsumerMaxMessagesQueueTest.java | 8 +-
.../nats/NatsConsumerMaxMessagesTest.java | 5 +-
.../component/nats/NatsConsumerReplyToTest.java | 6 +-
.../camel/component/nats/NatsConsumerTest.java | 5 +-
.../nats/NatsConsumerWithConnectionLoadTest.java | 15 +-
.../nats/NatsConsumerWithRedeliveryTest.java | 17 +-
.../camel/component/nats/NatsProducerTLSTest.java | 3 +-
.../camel/component/nats/NatsProducerTest.java | 3 +-
.../camel/component/nats/NatsTestSupport.java | 13 +-
.../opentracing/decorators/NatsSpanDecorator.java | 10 --
.../endpoint/dsl/NatsEndpointBuilderFactory.java | 185 ++++++++++-----------
.../modules/ROOT/pages/nats-component.adoc | 107 +++++++++---
.../modules/ROOT/pages/rest-swagger-component.adoc | 3 +-
.../springboot/NatsComponentConfiguration.java | 13 ++
27 files changed, 357 insertions(+), 219 deletions(-)
[camel] 02/04: Regen
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit ec710873d604d364947f56573830e3449151ab56
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Dec 4 10:00:33 2019 +0100
Regen
---
docs/components/modules/ROOT/pages/rest-swagger-component.adoc | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/docs/components/modules/ROOT/pages/rest-swagger-component.adoc b/docs/components/modules/ROOT/pages/rest-swagger-component.adoc
index 683f809..725b58b 100644
--- a/docs/components/modules/ROOT/pages/rest-swagger-component.adoc
+++ b/docs/components/modules/ROOT/pages/rest-swagger-component.adoc
@@ -127,7 +127,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (8 parameters):
+=== Query Parameters (9 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -141,6 +141,7 @@ with the following path and query parameters:
| *produces* (producer) | What payload type this component is producing. For example application/json according to the RFC7231. This equates to the value of Content-Type HTTP header. If set overrides any value present in the Swagger specification. Overrides all other configuration. | | String
| *basicPropertyBinding* (advanced) | Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+| *sslContextParameters* (security) | To configure security using SSLContextParameters. | | SSLContextParameters
|===
// endpoint options: END
// spring-boot-auto-configure options: START
[camel] 04/04: CAMEL-14253: camel-nats - Configure brokers on
component level
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 17552976c0692a4b34536c484b2e3d24be1b3f9c
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Dec 4 10:12:14 2019 +0100
CAMEL-14253: camel-nats - Configure brokers on component level
---
.../apache/camel/opentracing/decorators/NatsSpanDecorator.java | 10 ----------
1 file changed, 10 deletions(-)
diff --git a/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/decorators/NatsSpanDecorator.java b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/decorators/NatsSpanDecorator.java
index dfa3262..0882576 100644
--- a/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/decorators/NatsSpanDecorator.java
+++ b/components/camel-opentracing/src/main/java/org/apache/camel/opentracing/decorators/NatsSpanDecorator.java
@@ -16,9 +16,6 @@
*/
package org.apache.camel.opentracing.decorators;
-import java.util.Map;
-
-import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
public class NatsSpanDecorator extends AbstractMessagingSpanDecorator {
@@ -36,13 +33,6 @@ public class NatsSpanDecorator extends AbstractMessagingSpanDecorator {
}
@Override
- public String getDestination(Exchange exchange, Endpoint endpoint) {
- Map<String, String> queryParameters = toQueryParameters(endpoint.getEndpointUri());
- String topic = queryParameters.get("topic");
- return topic != null ? topic : super.getDestination(exchange, endpoint);
- }
-
- @Override
protected String getMessageId(Exchange exchange) {
return exchange.getIn().getHeader(NATS_SID, String.class);
}
[camel] 01/04: CAMEL-14253: camel-nats - Configure brokers on
component level
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 6f2ad84616da2ba30c43b998b5f3b57fc86acc89
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Dec 4 10:00:27 2019 +0100
CAMEL-14253: camel-nats - Configure brokers on component level
---
.../camel-nats/src/main/docs/nats-component.adoc | 15 +-
.../apache/camel/component/nats/NatsComponent.java | 18 +-
.../camel/component/nats/NatsConfiguration.java | 20 +--
.../apache/camel/component/nats/NatsConsumer.java | 1 -
.../apache/camel/component/nats/NatsEndpoint.java | 6 +-
.../apache/camel/component/nats/NatsProducer.java | 1 +
.../component/nats/NatsAuthConsumerLoadTest.java | 4 +-
.../camel/component/nats/NatsAuthProducerTest.java | 3 +-
.../camel/component/nats/NatsAuthTestSupport.java | 11 +-
.../nats/NatsAuthTokenConsumerLoadTest.java | 4 +-
.../component/nats/NatsAuthTokenProducerTest.java | 3 +-
.../component/nats/NatsAuthTokenTestSupport.java | 12 +-
.../camel/component/nats/NatsConsumerLoadTest.java | 9 +-
.../nats/NatsConsumerMaxMessagesQueueTest.java | 8 +-
.../nats/NatsConsumerMaxMessagesTest.java | 5 +-
.../component/nats/NatsConsumerReplyToTest.java | 6 +-
.../camel/component/nats/NatsConsumerTest.java | 5 +-
.../nats/NatsConsumerWithConnectionLoadTest.java | 15 +-
.../nats/NatsConsumerWithRedeliveryTest.java | 17 +-
.../camel/component/nats/NatsProducerTLSTest.java | 3 +-
.../camel/component/nats/NatsProducerTest.java | 3 +-
.../camel/component/nats/NatsTestSupport.java | 11 +-
.../endpoint/dsl/NatsEndpointBuilderFactory.java | 185 ++++++++++-----------
.../modules/ROOT/pages/nats-component.adoc | 35 ++--
.../springboot/NatsComponentConfiguration.java | 13 ++
25 files changed, 238 insertions(+), 175 deletions(-)
diff --git a/components/camel-nats/src/main/docs/nats-component.adoc b/components/camel-nats/src/main/docs/nats-component.adoc
index ff05266..8ee8984 100644
--- a/components/camel-nats/src/main/docs/nats-component.adoc
+++ b/components/camel-nats/src/main/docs/nats-component.adoc
@@ -35,13 +35,14 @@ Where *servers* represents the list of NATS servers.
// component options: START
-The Nats component supports 4 options, which are listed below.
+The Nats component supports 5 options, which are listed below.
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
+| *servers* (common) | URLs to one or more NAT servers. Use comma to separate URLs when specifying multiple servers. | | String
| *useGlobalSslContext Parameters* (security) | Enable usage of global SSL context parameters. | false | boolean
| *basicPropertyBinding* (advanced) | Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
| *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 [...]
@@ -57,7 +58,7 @@ The Nats component supports 4 options, which are listed below.
The Nats endpoint is configured using URI syntax:
----
-nats:servers
+nats:topic
----
with the following path and query parameters:
@@ -68,7 +69,7 @@ with the following path and query parameters:
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
-| *servers* | *Required* URLs to one or more NAT servers. Use comma to separate URLs when specifying multiple servers. | | String
+| *topic* | *Required* The name of topic we want to use | | String
|===
@@ -78,9 +79,8 @@ with the following path and query parameters:
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
-| *connection* (common) | Reference an already instantiated connection to Nats server | | Connection
| *connectionTimeout* (common) | Timeout for connection attempts. (in milliseconds) | 2000 | int
-| *flushConnection* (common) | Define if we want to flush connection or not | false | boolean
+| *flushConnection* (common) | Define if we want to flush connection when stopping or not | false | boolean
| *flushTimeout* (common) | Set the flush timeout (in milliseconds) | 1000 | int
| *maxPingsOut* (common) | maximum number of pings have not received a response allowed by the client | 2 | int
| *maxReconnectAttempts* (common) | Max reconnection attempts | 60 | int
@@ -91,11 +91,11 @@ with the following path and query parameters:
| *reconnect* (common) | Whether or not using reconnection feature | true | boolean
| *reconnectTimeWait* (common) | Waiting time before attempts reconnection (in milliseconds) | 2000 | int
| *requestCleanupInterval* (common) | Interval to clean up cancelled/timed out requests. | 5000 | int
-| *topic* (common) | *Required* The name of topic we want to use | | String
+| *servers* (common) | URLs to one or more NAT servers. Use comma to separate URLs when specifying multiple servers. | | String
| *verbose* (common) | Whether or not running in verbose mode | false | boolean
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *maxMessages* (consumer) | Stop receiving messages from a topic we are subscribing to after maxMessages | | String
-| *poolSize* (consumer) | Consumer pool size | 10 | int
+| *poolSize* (consumer) | Consumer thread pool size (default is 10) | 10 | int
| *queueName* (consumer) | The Queue name if we are using nats for a queue configuration | | String
| *replyToDisabled* (consumer) | Can be used to turn off sending back reply message in the consumer. | false | boolean
| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
@@ -103,6 +103,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 [...]
| *replySubject* (producer) | the subject to which subscribers should send response | | String
| *basicPropertyBinding* (advanced) | Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
+| *connection* (advanced) | Reference an already instantiated connection to Nats server | | Connection
| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
| *secure* (security) | Set secure option indicating TLS is required | false | boolean
| *sslContextParameters* (security) | To configure security using SSLContextParameters | | SSLContextParameters
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsComponent.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsComponent.java
index 26938a2..1829e6e 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsComponent.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsComponent.java
@@ -21,20 +21,24 @@ import java.util.Map;
import org.apache.camel.Endpoint;
import org.apache.camel.SSLContextParametersAware;
import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriPath;
import org.apache.camel.spi.annotations.Component;
import org.apache.camel.support.DefaultComponent;
@Component("nats")
public class NatsComponent extends DefaultComponent implements SSLContextParametersAware {
+ @Metadata
+ private String servers;
@Metadata(label = "security", defaultValue = "false")
private boolean useGlobalSslContextParameters;
@Override
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
NatsConfiguration config = new NatsConfiguration();
+ config.setTopic(remaining);
+ config.setServers(servers);
setProperties(config, parameters);
- config.setServers(remaining);
if (config.getSslContextParameters() == null) {
config.setSslContextParameters(retrieveGlobalSslContextParameters());
@@ -44,6 +48,18 @@ public class NatsComponent extends DefaultComponent implements SSLContextParamet
return endpoint;
}
+ /**
+ * URLs to one or more NAT servers. Use comma to separate URLs when
+ * specifying multiple servers.
+ */
+ public String getServers() {
+ return servers;
+ }
+
+ public void setServers(String servers) {
+ this.servers = servers;
+ }
+
@Override
public boolean isUseGlobalSslContextParameters() {
return this.useGlobalSslContextParameters;
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
index e053b96..742375b 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConfiguration.java
@@ -33,20 +33,19 @@ public class NatsConfiguration {
@UriPath
@Metadata(required = true)
- private String servers;
- @UriParam
- @Metadata(required = true)
private String topic;
- @UriParam
+ @UriParam(label = "common")
+ private String servers;
+ @UriParam(label = "advanced")
private Connection connection;
- @UriParam(defaultValue = "true")
+ @UriParam(label = "common", defaultValue = "true")
private boolean reconnect = true;
- @UriParam
+ @UriParam(label = "common", defaultValue = "2000")
+ private int reconnectTimeWait = 2000;
+ @UriParam(label = "common")
private boolean pedantic;
@UriParam
private boolean verbose;
- @UriParam(defaultValue = "2000")
- private int reconnectTimeWait = 2000;
@UriParam(defaultValue = "60")
private int maxReconnectAttempts = Options.DEFAULT_MAX_RECONNECT;
@UriParam(defaultValue = "120000")
@@ -79,6 +78,7 @@ public class NatsConfiguration {
private boolean secure;
@UriParam(label = "security")
private SSLContextParameters sslContextParameters;
+
/**
* URLs to one or more NAT servers. Use comma to separate URLs when
* specifying multiple servers.
@@ -284,7 +284,7 @@ public class NatsConfiguration {
}
/**
- * Consumer pool size
+ * Consumer thread pool size (default is 10)
*/
public int getPoolSize() {
return poolSize;
@@ -299,7 +299,7 @@ public class NatsConfiguration {
}
/**
- * Define if we want to flush connection or not
+ * Define if we want to flush connection when stopping or not
*/
public void setFlushConnection(boolean flushConnection) {
this.flushConnection = flushConnection;
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
index 5c905aa..41d6e8f 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java
@@ -62,7 +62,6 @@ public class NatsConsumer extends DefaultConsumer {
@Override
protected void doStop() throws Exception {
-
if (getEndpoint().getConfiguration().isFlushConnection()) {
log.debug("Flushing Messages before stopping");
connection.flush(Duration.ofMillis(getEndpoint().getConfiguration().getFlushTimeout()));
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsEndpoint.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsEndpoint.java
index d07b1fc..8f74250 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsEndpoint.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsEndpoint.java
@@ -36,7 +36,7 @@ import org.apache.camel.support.DefaultEndpoint;
/**
* The nats component allows you produce and consume messages from <a href="http://nats.io/">NATS</a>.
*/
-@UriEndpoint(firstVersion = "2.17.0", scheme = "nats", title = "Nats", syntax = "nats:servers", label = "messaging")
+@UriEndpoint(firstVersion = "2.17.0", scheme = "nats", title = "Nats", syntax = "nats:topic", label = "messaging")
public class NatsEndpoint extends DefaultEndpoint {
@UriParam
@@ -54,7 +54,9 @@ public class NatsEndpoint extends DefaultEndpoint {
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- return new NatsConsumer(this, processor);
+ NatsConsumer consumer = new NatsConsumer(this, processor);
+ configureConsumer(consumer);
+ return consumer;
}
public ExecutorService createExecutor() {
diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
index 764a7a5..438a99d 100644
--- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
+++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsProducer.java
@@ -22,6 +22,7 @@ import io.nats.client.Connection;
import io.nats.client.Connection.Status;
import org.apache.camel.Exchange;
import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.IOHelper;
import org.apache.camel.util.ObjectHelper;
public class NatsProducer extends DefaultProducer {
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthConsumerLoadTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthConsumerLoadTest.java
index 5d7c9fd..d1af70c 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthConsumerLoadTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthConsumerLoadTest.java
@@ -35,7 +35,7 @@ public class NatsAuthConsumerLoadTest extends NatsAuthTestSupport {
@Test
public void testLoadConsumer() throws InterruptedException, IOException, TimeoutException {
mockResultEndpoint.setExpectedMessageCount(100);
- Options options = new Options.Builder().server("nats://" + getNatsUrl()).build();
+ Options options = new Options.Builder().server("nats://" + getNatsBrokerUrl()).build();
Connection connection = Nats.connect(options);
for (int i = 0; i < 100; i++) {
@@ -50,7 +50,7 @@ public class NatsAuthConsumerLoadTest extends NatsAuthTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("nats://" + getNatsUrl() + "?topic=test").to(mockResultEndpoint);
+ from("nats:test").to(mockResultEndpoint);
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthProducerTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthProducerTest.java
index 63039d4..1a34bc2 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthProducerTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthProducerTest.java
@@ -23,7 +23,6 @@ public class NatsAuthProducerTest extends NatsAuthTestSupport {
@Test
public void sendTest() throws Exception {
-
template.sendBody("direct:send", "pippo");
}
@@ -32,7 +31,7 @@ public class NatsAuthProducerTest extends NatsAuthTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:send").to("nats://" + getNatsUrl() + "?topic=test");
+ from("direct:send").to("nats:test");
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTestSupport.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTestSupport.java
index 06d2b54..cfdc929 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTestSupport.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTestSupport.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.nats;
+import org.apache.camel.CamelContext;
import org.apache.camel.test.testcontainers.ContainerAwareTestSupport;
import org.apache.camel.test.testcontainers.Wait;
import org.testcontainers.containers.GenericContainer;
@@ -45,7 +46,7 @@ public class NatsAuthTestSupport extends ContainerAwareTestSupport {
);
}
- public String getNatsUrl() {
+ public String getNatsBrokerUrl() {
return String.format(
"%s:%s@%s:%d",
USERNAME,
@@ -54,4 +55,12 @@ public class NatsAuthTestSupport extends ContainerAwareTestSupport {
getContainerPort(CONTAINER_NAME, 4222)
);
}
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ NatsComponent nats = context.getComponent("nats", NatsComponent.class);
+ nats.setServers(getNatsBrokerUrl());
+ return context;
+ }
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenConsumerLoadTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenConsumerLoadTest.java
index a2156b5..bb38064 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenConsumerLoadTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenConsumerLoadTest.java
@@ -35,7 +35,7 @@ public class NatsAuthTokenConsumerLoadTest extends NatsAuthTokenTestSupport {
@Test
public void testLoadConsumer() throws InterruptedException, IOException, TimeoutException {
mockResultEndpoint.setExpectedMessageCount(100);
- Options options = new Options.Builder().server("nats://" + getNatsUrl()).build();
+ Options options = new Options.Builder().server("nats://" + getNatsBrokerUrl()).build();
Connection connection = Nats.connect(options);
for (int i = 0; i < 100; i++) {
@@ -50,7 +50,7 @@ public class NatsAuthTokenConsumerLoadTest extends NatsAuthTokenTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("nats://" + getNatsUrl() + "?topic=test").to(mockResultEndpoint);
+ from("nats:test").to(mockResultEndpoint);
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenProducerTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenProducerTest.java
index f32fc3b..c087b16 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenProducerTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenProducerTest.java
@@ -23,7 +23,6 @@ public class NatsAuthTokenProducerTest extends NatsAuthTestSupport {
@Test
public void sendTest() throws Exception {
-
template.sendBody("direct:send", "pippo");
}
@@ -32,7 +31,7 @@ public class NatsAuthTokenProducerTest extends NatsAuthTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:send").to("nats://" + getNatsUrl() + "?topic=test");
+ from("direct:send").to("nats:test");
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenTestSupport.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenTestSupport.java
index 84c8db5..2a19a28 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenTestSupport.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenTestSupport.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.nats;
+import org.apache.camel.CamelContext;
import org.apache.camel.test.testcontainers.ContainerAwareTestSupport;
import org.apache.camel.test.testcontainers.Wait;
import org.testcontainers.containers.GenericContainer;
@@ -42,7 +43,7 @@ public class NatsAuthTokenTestSupport extends ContainerAwareTestSupport {
);
}
- public String getNatsUrl() {
+ public String getNatsBrokerUrl() {
return String.format(
"%s@%s:%d",
TOKEN,
@@ -50,4 +51,13 @@ public class NatsAuthTokenTestSupport extends ContainerAwareTestSupport {
getContainerPort(CONTAINER_NAME, 4222)
);
}
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ NatsComponent nats = context.getComponent("nats", NatsComponent.class);
+ nats.setServers(getNatsBrokerUrl());
+ return context;
+ }
+
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerLoadTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerLoadTest.java
index 3c0fc66..3a016b0 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerLoadTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerLoadTest.java
@@ -16,9 +16,6 @@
*/
package org.apache.camel.component.nats;
-import java.io.IOException;
-import java.util.concurrent.TimeoutException;
-
import io.nats.client.Connection;
import io.nats.client.Nats;
import io.nats.client.Options;
@@ -33,9 +30,9 @@ public class NatsConsumerLoadTest extends NatsTestSupport {
protected MockEndpoint mockResultEndpoint;
@Test
- public void testLoadConsumer() throws InterruptedException, IOException, TimeoutException {
+ public void testLoadConsumer() throws Exception {
mockResultEndpoint.setExpectedMessageCount(10000);
- Options options = new Options.Builder().server("nats://" + getNatsUrl()).build();
+ Options options = new Options.Builder().server("nats://" + getNatsBrokerUrl()).build();
Connection connection = Nats.connect(options);
for (int i = 0; i < 10000; i++) {
@@ -50,7 +47,7 @@ public class NatsConsumerLoadTest extends NatsTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("nats://" + getNatsUrl() + "?topic=test").to(mockResultEndpoint);
+ from("nats:test").to(mockResultEndpoint);
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerMaxMessagesQueueTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerMaxMessagesQueueTest.java
index 94508aa..6fa1f02 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerMaxMessagesQueueTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerMaxMessagesQueueTest.java
@@ -43,9 +43,11 @@ public class NatsConsumerMaxMessagesQueueTest extends NatsTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:send").to("nats://" + getNatsUrl() + "?topic=test");
- from("nats://" + getNatsUrl() + "?topic=test&maxMessages=5&queueName=test").routeId("cons1").to(mockResultEndpoint);
- from("nats://" + getNatsUrl() + "?topic=test&maxMessages=6&queueName=test").routeId("cons2").to(mockResultEndpoint);
+ from("direct:send").to("nats:test");
+
+ from("nats:test?maxMessages=5&queueName=test").routeId("cons1").to(mockResultEndpoint);
+
+ from("nats:test?maxMessages=6&queueName=test").routeId("cons2").to(mockResultEndpoint);
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerMaxMessagesTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerMaxMessagesTest.java
index 57be8a6..2892757 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerMaxMessagesTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerMaxMessagesTest.java
@@ -51,8 +51,9 @@ public class NatsConsumerMaxMessagesTest extends NatsTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:send").to("nats://" + getNatsUrl() + "?topic=test");
- from("nats://" + getNatsUrl() + "?topic=test&maxMessages=5").to(mockResultEndpoint);
+ from("direct:send").to("nats:test");
+
+ from("nats:test?maxMessages=5").to(mockResultEndpoint);
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerReplyToTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerReplyToTest.java
index 33a5d5b..5572854 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerReplyToTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerReplyToTest.java
@@ -36,7 +36,7 @@ public class NatsConsumerReplyToTest extends NatsTestSupport {
mockResultEndpoint.assertIsSatisfied();
// grab reply message from the reply queue
- String out = consumer.receiveBody("nats://" + getNatsUrl() + "?topic=myReplyQueue", 5000, String.class);
+ String out = consumer.receiveBody("nats:myReplyQueue", 5000, String.class);
assertEquals("Bye World", out);
}
@@ -46,9 +46,9 @@ public class NatsConsumerReplyToTest extends NatsTestSupport {
@Override
public void configure() throws Exception {
from("direct:send")
- .to("nats://" + getNatsUrl() + "?topic=test&replySubject=myReplyQueue&flushConnection=true");
+ .to("nats:test?replySubject=myReplyQueue&flushConnection=true");
- from("nats://" + getNatsUrl() + "?topic=test&flushConnection=true")
+ from("nats:test?flushConnection=true")
.to(mockResultEndpoint)
.convertBodyTo(String.class)
.setBody().simple("Bye ${body}");
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerTest.java
index 85676a4..2954b55 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerTest.java
@@ -41,8 +41,9 @@ public class NatsConsumerTest extends NatsTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:send").to("nats://" + getNatsUrl() + "?topic=test&flushConnection=true");
- from("nats://" + getNatsUrl() + "?topic=test&flushConnection=true").to(mockResultEndpoint);
+ from("direct:send").to("nats:test?flushConnection=true");
+
+ from("nats:test?flushConnection=true").to(mockResultEndpoint);
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerWithConnectionLoadTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerWithConnectionLoadTest.java
index 61be2ba..1b712a3 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerWithConnectionLoadTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerWithConnectionLoadTest.java
@@ -16,9 +16,6 @@
*/
package org.apache.camel.component.nats;
-import java.io.IOException;
-import java.util.concurrent.TimeoutException;
-
import io.nats.client.Connection;
import io.nats.client.Nats;
import io.nats.client.Options;
@@ -40,18 +37,18 @@ public class NatsConsumerWithConnectionLoadTest extends NatsTestSupport {
private Connection connection;
@BindToRegistry("connection")
- public Connection connection() throws IllegalStateException, IOException, InterruptedException {
+ public Connection connection() throws Exception {
Builder options = new Options.Builder();
- options.server("nats://" + getNatsUrl());
+ options.server("nats://" + getNatsBrokerUrl());
connection = Nats.connect(options.build());
return connection;
}
@Test
- public void testLoadConsumer() throws InterruptedException, IOException, TimeoutException {
+ public void testLoadConsumer() throws Exception {
mockResultEndpoint.setExpectedMessageCount(100);
mockResultEndpoint1.setExpectedMessageCount(0);
- Options options = new Options.Builder().server("nats://" + getNatsUrl()).build();
+ Options options = new Options.Builder().server("nats://" + getNatsBrokerUrl()).build();
Connection connection = Nats.connect(options);
for (int i = 0; i < 100; i++) {
@@ -67,8 +64,8 @@ public class NatsConsumerWithConnectionLoadTest extends NatsTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("nats://thisismytest?topic=test&connection=#connection").to(mockResultEndpoint);
- from("nats://thisismytest?topic=test1&connection=#connection").to(mockResultEndpoint1);
+ from("nats:test?connection=#connection").to(mockResultEndpoint);
+ from("nats:test1?connection=#connection").to(mockResultEndpoint1);
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerWithRedeliveryTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerWithRedeliveryTest.java
index afea0dc..c418744 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerWithRedeliveryTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerWithRedeliveryTest.java
@@ -38,7 +38,7 @@ public class NatsConsumerWithRedeliveryTest extends NatsTestSupport {
private MockEndpoint exception;
@Test
- public void testConsumer() throws InterruptedException, IOException {
+ public void testConsumer() throws Exception {
mockResultEndpoint.setExpectedMessageCount(1);
mockResultEndpoint.setAssertPeriod(1000);
@@ -60,17 +60,14 @@ public class NatsConsumerWithRedeliveryTest extends NatsTestSupport {
onException(Exception.class).maximumRedeliveries(REDELIVERY_COUNT).retryAttemptedLogLevel(LoggingLevel.INFO).retriesExhaustedLogLevel(LoggingLevel.ERROR)
.redeliveryDelay(10).to("mock:exception").handled(true);
- from("direct:send").to("nats://" + getNatsUrl() + "?topic=test&flushConnection=true");
- from("nats://" + getNatsUrl() + "?topic=test&flushConnection=true").choice().when(new Predicate() {
+ from("direct:send").to("nats:test?flushConnection=true");
- @Override
- public boolean matches(Exchange exchange) {
- String s = exchange.getMessage().getBody(String.class);
- if (s.contains("test")) {
- return true;
- }
- return false;
+ from("nats:test?flushConnection=true").choice().when(exchange -> {
+ String s = exchange.getMessage().getBody(String.class);
+ if (s.contains("test")) {
+ return true;
}
+ return false;
}).throwException(RuntimeCamelException.class, "Test for this").end().to(mockResultEndpoint);
}
};
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsProducerTLSTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsProducerTLSTest.java
index f15ab7b..d2ebb50 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsProducerTLSTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsProducerTLSTest.java
@@ -37,7 +37,6 @@ public class NatsProducerTLSTest extends CamelTestSupport {
@Test
public void sendTest() throws Exception {
-
template.sendBody("direct:send", "pippo");
}
@@ -60,7 +59,7 @@ public class NatsProducerTLSTest extends CamelTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:send").to("nats://localhost:4222?topic=test&sslContextParameters=#ssl&secure=true");
+ from("direct:send").to("nats:topic=test?servers=localhost:4222&sslContextParameters=#ssl&secure=true");
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsProducerTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsProducerTest.java
index 0e12d36..6f7d3f0 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsProducerTest.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsProducerTest.java
@@ -23,7 +23,6 @@ public class NatsProducerTest extends NatsTestSupport {
@Test
public void sendTest() throws Exception {
-
template.sendBody("direct:send", "pippo");
}
@@ -32,7 +31,7 @@ public class NatsProducerTest extends NatsTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:send").to("nats://" + getNatsUrl() + "?topic=test");
+ from("direct:send").to("nats:test");
}
};
}
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java
index deaa187..e13affd 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.nats;
+import org.apache.camel.CamelContext;
import org.apache.camel.test.testcontainers.ContainerAwareTestSupport;
import org.apache.camel.test.testcontainers.Wait;
import org.testcontainers.containers.GenericContainer;
@@ -36,11 +37,19 @@ public class NatsTestSupport extends ContainerAwareTestSupport {
.waitingFor(Wait.forLogMessageContaining("Listening for route connections", 1));
}
- public String getNatsUrl() {
+ public String getNatsBrokerUrl() {
return String.format(
"%s:%d",
getContainerHost(CONTAINER_NAME),
getContainerPort(CONTAINER_NAME, 4222)
);
}
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ NatsComponent nats = context.getComponent("nats", NatsComponent.class);
+ nats.setServers(getNatsBrokerUrl());
+ return context;
+ }
}
diff --git a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/NatsEndpointBuilderFactory.java b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/NatsEndpointBuilderFactory.java
index b47029a..f57a85e 100644
--- a/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/NatsEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/main/java/org/apache/camel/builder/endpoint/dsl/NatsEndpointBuilderFactory.java
@@ -42,29 +42,6 @@ public interface NatsEndpointBuilderFactory {
return (AdvancedNatsEndpointConsumerBuilder) this;
}
/**
- * Reference an already instantiated connection to Nats server.
- *
- * The option is a: <code>io.nats.client.Connection</code> type.
- *
- * Group: common
- */
- default NatsEndpointConsumerBuilder connection(Object connection) {
- doSetProperty("connection", connection);
- return this;
- }
- /**
- * Reference an already instantiated connection to Nats server.
- *
- * The option will be converted to a
- * <code>io.nats.client.Connection</code> type.
- *
- * Group: common
- */
- default NatsEndpointConsumerBuilder connection(String connection) {
- doSetProperty("connection", connection);
- return this;
- }
- /**
* Timeout for connection attempts. (in milliseconds).
*
* The option is a: <code>int</code> type.
@@ -89,7 +66,7 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * Define if we want to flush connection or not.
+ * Define if we want to flush connection when stopping or not.
*
* The option is a: <code>boolean</code> type.
*
@@ -101,7 +78,7 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * Define if we want to flush connection or not.
+ * Define if we want to flush connection when stopping or not.
*
* The option will be converted to a <code>boolean</code> type.
*
@@ -353,15 +330,15 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * The name of topic we want to use.
+ * URLs to one or more NAT servers. Use comma to separate URLs when
+ * specifying multiple servers.
*
* The option is a: <code>java.lang.String</code> type.
*
- * Required: true
* Group: common
*/
- default NatsEndpointConsumerBuilder topic(String topic) {
- doSetProperty("topic", topic);
+ default NatsEndpointConsumerBuilder servers(String servers) {
+ doSetProperty("servers", servers);
return this;
}
/**
@@ -435,7 +412,7 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * Consumer pool size.
+ * Consumer thread pool size (default is 10).
*
* The option is a: <code>int</code> type.
*
@@ -446,7 +423,7 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * Consumer pool size.
+ * Consumer thread pool size (default is 10).
*
* The option will be converted to a <code>int</code> type.
*
@@ -634,6 +611,29 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
+ * Reference an already instantiated connection to Nats server.
+ *
+ * The option is a: <code>io.nats.client.Connection</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedNatsEndpointConsumerBuilder connection(Object connection) {
+ doSetProperty("connection", connection);
+ return this;
+ }
+ /**
+ * Reference an already instantiated connection to Nats server.
+ *
+ * The option will be converted to a
+ * <code>io.nats.client.Connection</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedNatsEndpointConsumerBuilder connection(String connection) {
+ doSetProperty("connection", connection);
+ return this;
+ }
+ /**
* Sets whether synchronous processing should be strictly used, or Camel
* is allowed to use asynchronous processing (if supported).
*
@@ -671,29 +671,6 @@ public interface NatsEndpointBuilderFactory {
return (AdvancedNatsEndpointProducerBuilder) this;
}
/**
- * Reference an already instantiated connection to Nats server.
- *
- * The option is a: <code>io.nats.client.Connection</code> type.
- *
- * Group: common
- */
- default NatsEndpointProducerBuilder connection(Object connection) {
- doSetProperty("connection", connection);
- return this;
- }
- /**
- * Reference an already instantiated connection to Nats server.
- *
- * The option will be converted to a
- * <code>io.nats.client.Connection</code> type.
- *
- * Group: common
- */
- default NatsEndpointProducerBuilder connection(String connection) {
- doSetProperty("connection", connection);
- return this;
- }
- /**
* Timeout for connection attempts. (in milliseconds).
*
* The option is a: <code>int</code> type.
@@ -718,7 +695,7 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * Define if we want to flush connection or not.
+ * Define if we want to flush connection when stopping or not.
*
* The option is a: <code>boolean</code> type.
*
@@ -730,7 +707,7 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * Define if we want to flush connection or not.
+ * Define if we want to flush connection when stopping or not.
*
* The option will be converted to a <code>boolean</code> type.
*
@@ -982,15 +959,15 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * The name of topic we want to use.
+ * URLs to one or more NAT servers. Use comma to separate URLs when
+ * specifying multiple servers.
*
* The option is a: <code>java.lang.String</code> type.
*
- * Required: true
* Group: common
*/
- default NatsEndpointProducerBuilder topic(String topic) {
- doSetProperty("topic", topic);
+ default NatsEndpointProducerBuilder servers(String servers) {
+ doSetProperty("servers", servers);
return this;
}
/**
@@ -1152,6 +1129,29 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
+ * Reference an already instantiated connection to Nats server.
+ *
+ * The option is a: <code>io.nats.client.Connection</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedNatsEndpointProducerBuilder connection(Object connection) {
+ doSetProperty("connection", connection);
+ return this;
+ }
+ /**
+ * Reference an already instantiated connection to Nats server.
+ *
+ * The option will be converted to a
+ * <code>io.nats.client.Connection</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedNatsEndpointProducerBuilder connection(String connection) {
+ doSetProperty("connection", connection);
+ return this;
+ }
+ /**
* Sets whether synchronous processing should be strictly used, or Camel
* is allowed to use asynchronous processing (if supported).
*
@@ -1189,29 +1189,6 @@ public interface NatsEndpointBuilderFactory {
return (AdvancedNatsEndpointBuilder) this;
}
/**
- * Reference an already instantiated connection to Nats server.
- *
- * The option is a: <code>io.nats.client.Connection</code> type.
- *
- * Group: common
- */
- default NatsEndpointBuilder connection(Object connection) {
- doSetProperty("connection", connection);
- return this;
- }
- /**
- * Reference an already instantiated connection to Nats server.
- *
- * The option will be converted to a
- * <code>io.nats.client.Connection</code> type.
- *
- * Group: common
- */
- default NatsEndpointBuilder connection(String connection) {
- doSetProperty("connection", connection);
- return this;
- }
- /**
* Timeout for connection attempts. (in milliseconds).
*
* The option is a: <code>int</code> type.
@@ -1234,7 +1211,7 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * Define if we want to flush connection or not.
+ * Define if we want to flush connection when stopping or not.
*
* The option is a: <code>boolean</code> type.
*
@@ -1245,7 +1222,7 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * Define if we want to flush connection or not.
+ * Define if we want to flush connection when stopping or not.
*
* The option will be converted to a <code>boolean</code> type.
*
@@ -1493,15 +1470,15 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
- * The name of topic we want to use.
+ * URLs to one or more NAT servers. Use comma to separate URLs when
+ * specifying multiple servers.
*
* The option is a: <code>java.lang.String</code> type.
*
- * Required: true
* Group: common
*/
- default NatsEndpointBuilder topic(String topic) {
- doSetProperty("topic", topic);
+ default NatsEndpointBuilder servers(String servers) {
+ doSetProperty("servers", servers);
return this;
}
/**
@@ -1612,6 +1589,29 @@ public interface NatsEndpointBuilderFactory {
return this;
}
/**
+ * Reference an already instantiated connection to Nats server.
+ *
+ * The option is a: <code>io.nats.client.Connection</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedNatsEndpointBuilder connection(Object connection) {
+ doSetProperty("connection", connection);
+ return this;
+ }
+ /**
+ * Reference an already instantiated connection to Nats server.
+ *
+ * The option will be converted to a
+ * <code>io.nats.client.Connection</code> type.
+ *
+ * Group: advanced
+ */
+ default AdvancedNatsEndpointBuilder connection(String connection) {
+ doSetProperty("connection", connection);
+ return this;
+ }
+ /**
* Sets whether synchronous processing should be strictly used, or Camel
* is allowed to use asynchronous processing (if supported).
*
@@ -1644,11 +1644,10 @@ public interface NatsEndpointBuilderFactory {
* Since: 2.17
* Maven coordinates: org.apache.camel:camel-nats
*
- * Syntax: <code>nats:servers</code>
+ * Syntax: <code>nats:topic</code>
*
- * Path parameter: servers (required)
- * URLs to one or more NAT servers. Use comma to separate URLs when
- * specifying multiple servers.
+ * Path parameter: topic (required)
+ * The name of topic we want to use
*/
default NatsEndpointBuilder nats(String path) {
class NatsEndpointBuilderImpl extends AbstractEndpointBuilder implements NatsEndpointBuilder, AdvancedNatsEndpointBuilder {
diff --git a/docs/components/modules/ROOT/pages/nats-component.adoc b/docs/components/modules/ROOT/pages/nats-component.adoc
index f0d7924..1533e4c 100644
--- a/docs/components/modules/ROOT/pages/nats-component.adoc
+++ b/docs/components/modules/ROOT/pages/nats-component.adoc
@@ -36,13 +36,14 @@ Where *servers* represents the list of NATS servers.
// component options: START
-The Nats component supports 4 options, which are listed below.
+The Nats component supports 5 options, which are listed below.
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
+| *servers* (common) | URLs to one or more NAT servers. Use comma to separate URLs when specifying multiple servers. | | String
| *useGlobalSslContext Parameters* (security) | Enable usage of global SSL context parameters. | false | boolean
| *basicPropertyBinding* (advanced) | Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
| *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 [...]
@@ -58,7 +59,7 @@ The Nats component supports 4 options, which are listed below.
The Nats endpoint is configured using URI syntax:
----
-nats:servers
+nats:topic
----
with the following path and query parameters:
@@ -69,19 +70,18 @@ with the following path and query parameters:
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
-| *servers* | *Required* URLs to one or more NAT servers. Use comma to separate URLs when specifying multiple servers. | | String
+| *topic* | *Required* The name of topic we want to use | | String
|===
-=== Query Parameters (27 parameters):
+=== Query Parameters (28 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
-| *connection* (common) | Reference an already instantiated connection to Nats server | | Connection
| *connectionTimeout* (common) | Timeout for connection attempts. (in milliseconds) | 2000 | int
-| *flushConnection* (common) | Define if we want to flush connection or not | false | boolean
+| *flushConnection* (common) | Define if we want to flush connection when stopping or not | false | boolean
| *flushTimeout* (common) | Set the flush timeout (in milliseconds) | 1000 | int
| *maxPingsOut* (common) | maximum number of pings have not received a response allowed by the client | 2 | int
| *maxReconnectAttempts* (common) | Max reconnection attempts | 60 | int
@@ -92,17 +92,19 @@ with the following path and query parameters:
| *reconnect* (common) | Whether or not using reconnection feature | true | boolean
| *reconnectTimeWait* (common) | Waiting time before attempts reconnection (in milliseconds) | 2000 | int
| *requestCleanupInterval* (common) | Interval to clean up cancelled/timed out requests. | 5000 | int
-| *topic* (common) | *Required* The name of topic we want to use | | String
+| *servers* (common) | URLs to one or more NAT servers. Use comma to separate URLs when specifying multiple servers. | | String
| *verbose* (common) | Whether or not running in verbose mode | false | boolean
| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
| *maxMessages* (consumer) | Stop receiving messages from a topic we are subscribing to after maxMessages | | String
-| *poolSize* (consumer) | Consumer pool size | 10 | int
+| *poolSize* (consumer) | Consumer thread pool size (default is 10) | 10 | int
| *queueName* (consumer) | The Queue name if we are using nats for a queue configuration | | String
+| *replyToDisabled* (consumer) | Can be used to turn off sending back reply message in the consumer. | false | boolean
| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. | | ExchangePattern
| *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 [...]
| *replySubject* (producer) | the subject to which subscribers should send response | | String
| *basicPropertyBinding* (advanced) | Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
+| *connection* (advanced) | Reference an already instantiated connection to Nats server | | Connection
| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
| *secure* (security) | Set secure option indicating TLS is required | false | boolean
| *sslContextParameters* (security) | To configure security using SSLContextParameters | | SSLContextParameters
@@ -148,10 +150,21 @@ The component supports 5 options, which are listed below.
[width="100%",options="header"]
|=======================================================================
|Name |Type |Description
-
-|CamelNatsMessageTimestamp |long |The timestamp of a consumed message.
+| CamelNatsSID | String | The SID of a consumed message.
+| CamelNatsReplyTo | String | The ReplyTo of a consumed message (may be null).
+| CamelNatsSubject | String | The Subject of a consumed message.
+| CamelNatsQueueName | String | The Queue name of a consumed message (may be null).
+| CamelNatsMessageTimestamp | long | The timestamp of a consumed message.
|=======================================================================
-
+
+== Request/Reply support
+The producer only supports publishing (sending) messages.
+The producer does not support request/reply where it can wait for an expected reply message.
+
+The consumer will when routing the message is complete, send back the message as reply-message if required.
+
+== Examples
+
*Producer example:*
[source,java]
diff --git a/platforms/spring-boot/components-starter/camel-nats-starter/src/main/java/org/apache/camel/component/nats/springboot/NatsComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-nats-starter/src/main/java/org/apache/camel/component/nats/springboot/NatsComponentConfiguration.java
index 645e2af..cdf85f9 100644
--- a/platforms/spring-boot/components-starter/camel-nats-starter/src/main/java/org/apache/camel/component/nats/springboot/NatsComponentConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-nats-starter/src/main/java/org/apache/camel/component/nats/springboot/NatsComponentConfiguration.java
@@ -37,6 +37,11 @@ public class NatsComponentConfiguration
*/
private Boolean enabled;
/**
+ * URLs to one or more NAT servers. Use comma to separate URLs when
+ * specifying multiple servers.
+ */
+ private String servers;
+ /**
* Enable usage of global SSL context parameters.
*/
private Boolean useGlobalSslContextParameters = false;
@@ -66,6 +71,14 @@ public class NatsComponentConfiguration
*/
private Boolean bridgeErrorHandler = false;
+ public String getServers() {
+ return servers;
+ }
+
+ public void setServers(String servers) {
+ this.servers = servers;
+ }
+
public Boolean getUseGlobalSslContextParameters() {
return useGlobalSslContextParameters;
}
[camel] 03/04: CAMEL-14253: camel-nats - Configure brokers on
component level
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 725fed6e733ccf756217a4ddbbf547d9fa7600bd
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Dec 4 10:10:31 2019 +0100
CAMEL-14253: camel-nats - Configure brokers on component level
---
.../camel-nats/src/main/docs/nats-component.adoc | 72 +++++++++++++++++-----
.../camel/component/nats/NatsAuthTestSupport.java | 2 +-
.../component/nats/NatsAuthTokenTestSupport.java | 2 +-
.../camel/component/nats/NatsTestSupport.java | 2 +-
.../modules/ROOT/pages/nats-component.adoc | 72 +++++++++++++++++-----
5 files changed, 117 insertions(+), 33 deletions(-)
diff --git a/components/camel-nats/src/main/docs/nats-component.adoc b/components/camel-nats/src/main/docs/nats-component.adoc
index 8ee8984..bc340ac 100644
--- a/components/camel-nats/src/main/docs/nats-component.adoc
+++ b/components/camel-nats/src/main/docs/nats-component.adoc
@@ -26,10 +26,10 @@ their `pom.xml` for this component.
[source,java]
----------------------
-nats:servers[?options]
+nats:topic[?options]
----------------------
-Where *servers* represents the list of NATS servers.
+Where *topic* is the topic name
== Options
@@ -125,7 +125,7 @@ When using Spring Boot make sure to use the following Maven dependency to have s
----
-The component supports 5 options, which are listed below.
+The component supports 6 options, which are listed below.
@@ -136,13 +136,51 @@ The component supports 5 options, which are listed below.
| *camel.component.nats.bridge-error-handler* | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | Boolean
| *camel.component.nats.enabled* | Whether to enable auto configuration of the nats component. This is enabled by default. | | Boolean
| *camel.component.nats.lazy-start-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 [...]
+| *camel.component.nats.servers* | URLs to one or more NAT servers. Use comma to separate URLs when specifying multiple servers. | | String
| *camel.component.nats.use-global-ssl-context-parameters* | Enable usage of global SSL context parameters. | false | Boolean
|===
// spring-boot-auto-configure options: END
+== Configuring servers
+You configure the NATS servers on either the component or the endpoint.
+For example to configure this once on the component you can do:
+
+[source,java]
+----
+NatsComponent nats = context.getComponent("nats", NatsComponent.class);
+nats.setServers("someserver:4222,someotherserver:42222");
+----
+
+Notice how you can specify multiple servers separated by comma.
+
+Or you can specify the servers in the endpoint URI
+
+[source,java]
+----
+from("direct:send").to("nats:test?servers=localhost:4222");
+----
+
+The endpoint configuration will override any server configuration on the component level.
+
+=== Configuring username and password or token
+
+You can specify username and password for the servers in the server URLs,
+where its `username:password@url`, or `token@url` etc:
+
+[source,java]
+----
+NatsComponent nats = context.getComponent("nats", NatsComponent.class);
+nats.setServers("scott:tiger@someserver:4222,superman:123@someotherserver:42222");
+----
+
+If you are using Camel Main or Spring Boot you can configure the server urls in the `application.properties` file
+[source,properties]
+----
+camel.component.nats.servers=scott:tiger@someserver:4222,superman:123@someotherserver:42222
+----
== Headers
@@ -167,27 +205,31 @@ The consumer will when routing the message is complete, send back the message as
*Producer example:*
[source,java]
------------------------------------------------------------
-from("direct:send").to("nats://localhost:4222?topic=test");
------------------------------------------------------------
+----
+from("direct:send")
+ .to("nats:mytopic");
+----
In case of using Authorization you can directly specify your credentials in the server URL
[source,java]
------------------------------------------------------------
-from("direct:send").to("nats://username:password@localhost:4222?topic=test");
------------------------------------------------------------
+----
+from("direct:send")
+ .to("nats:mytopic?servers=username:password@localhost:4222");
+----
or your token
[source,java]
------------------------------------------------------------
-from("direct:send").to("nats://token@localhost:4222?topic=test");
------------------------------------------------------------
+----
+from("direct:send")
+ .to("nats:mytopic?servers=token@localhost:4222);
+----
*Consumer example:*
[source,java]
-----------------------------------------------------------------------------------------
-from("nats://localhost:4222?topic=test&maxMessages=5&queueName=test").to("mock:result");
-----------------------------------------------------------------------------------------
+----
+from("nats:mytopic?maxMessages=5&queueName=myqueue")
+ .to("mock:result");
+----
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTestSupport.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTestSupport.java
index cfdc929..f7b9549 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTestSupport.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTestSupport.java
@@ -23,7 +23,7 @@ import org.testcontainers.containers.GenericContainer;
public class NatsAuthTestSupport extends ContainerAwareTestSupport {
- public static final String CONTAINER_IMAGE = "nats:2.1.0";
+ public static final String CONTAINER_IMAGE = "nats:2.1.2";
public static final String CONTAINER_NAME = "nats-auth";
public static final String USERNAME = "admin";
public static final String PASSWORD = "password";
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenTestSupport.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenTestSupport.java
index 2a19a28..23c68e8 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenTestSupport.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsAuthTokenTestSupport.java
@@ -23,7 +23,7 @@ import org.testcontainers.containers.GenericContainer;
public class NatsAuthTokenTestSupport extends ContainerAwareTestSupport {
- public static final String CONTAINER_IMAGE = "nats:2.1.0";
+ public static final String CONTAINER_IMAGE = "nats:2.1.2";
public static final String CONTAINER_NAME = "nats-auth-token";
public static final String TOKEN = "!admin23456";
diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java
index e13affd..3352859 100644
--- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java
+++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsTestSupport.java
@@ -23,7 +23,7 @@ import org.testcontainers.containers.GenericContainer;
public class NatsTestSupport extends ContainerAwareTestSupport {
- public static final String CONTAINER_IMAGE = "nats:2.1.0";
+ public static final String CONTAINER_IMAGE = "nats:2.1.2";
public static final String CONTAINER_NAME = "nats";
@Override
diff --git a/docs/components/modules/ROOT/pages/nats-component.adoc b/docs/components/modules/ROOT/pages/nats-component.adoc
index 1533e4c..746faaa 100644
--- a/docs/components/modules/ROOT/pages/nats-component.adoc
+++ b/docs/components/modules/ROOT/pages/nats-component.adoc
@@ -27,10 +27,10 @@ their `pom.xml` for this component.
[source,java]
----------------------
-nats:servers[?options]
+nats:topic[?options]
----------------------
-Where *servers* represents the list of NATS servers.
+Where *topic* is the topic name
== Options
@@ -126,7 +126,7 @@ When using Spring Boot make sure to use the following Maven dependency to have s
----
-The component supports 5 options, which are listed below.
+The component supports 6 options, which are listed below.
@@ -137,13 +137,51 @@ The component supports 5 options, which are listed below.
| *camel.component.nats.bridge-error-handler* | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | Boolean
| *camel.component.nats.enabled* | Whether to enable auto configuration of the nats component. This is enabled by default. | | Boolean
| *camel.component.nats.lazy-start-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 [...]
+| *camel.component.nats.servers* | URLs to one or more NAT servers. Use comma to separate URLs when specifying multiple servers. | | String
| *camel.component.nats.use-global-ssl-context-parameters* | Enable usage of global SSL context parameters. | false | Boolean
|===
// spring-boot-auto-configure options: END
+== Configuring servers
+You configure the NATS servers on either the component or the endpoint.
+For example to configure this once on the component you can do:
+
+[source,java]
+----
+NatsComponent nats = context.getComponent("nats", NatsComponent.class);
+nats.setServers("someserver:4222,someotherserver:42222");
+----
+
+Notice how you can specify multiple servers separated by comma.
+
+Or you can specify the servers in the endpoint URI
+
+[source,java]
+----
+from("direct:send").to("nats:test?servers=localhost:4222");
+----
+
+The endpoint configuration will override any server configuration on the component level.
+
+=== Configuring username and password or token
+
+You can specify username and password for the servers in the server URLs,
+where its `username:password@url`, or `token@url` etc:
+
+[source,java]
+----
+NatsComponent nats = context.getComponent("nats", NatsComponent.class);
+nats.setServers("scott:tiger@someserver:4222,superman:123@someotherserver:42222");
+----
+
+If you are using Camel Main or Spring Boot you can configure the server urls in the `application.properties` file
+[source,properties]
+----
+camel.component.nats.servers=scott:tiger@someserver:4222,superman:123@someotherserver:42222
+----
== Headers
@@ -168,27 +206,31 @@ The consumer will when routing the message is complete, send back the message as
*Producer example:*
[source,java]
------------------------------------------------------------
-from("direct:send").to("nats://localhost:4222?topic=test");
------------------------------------------------------------
+----
+from("direct:send")
+ .to("nats:mytopic");
+----
In case of using Authorization you can directly specify your credentials in the server URL
[source,java]
------------------------------------------------------------
-from("direct:send").to("nats://username:password@localhost:4222?topic=test");
------------------------------------------------------------
+----
+from("direct:send")
+ .to("nats:mytopic?servers=username:password@localhost:4222");
+----
or your token
[source,java]
------------------------------------------------------------
-from("direct:send").to("nats://token@localhost:4222?topic=test");
------------------------------------------------------------
+----
+from("direct:send")
+ .to("nats:mytopic?servers=token@localhost:4222);
+----
*Consumer example:*
[source,java]
-----------------------------------------------------------------------------------------
-from("nats://localhost:4222?topic=test&maxMessages=5&queueName=test").to("mock:result");
-----------------------------------------------------------------------------------------
+----
+from("nats:mytopic?maxMessages=5&queueName=myqueue")
+ .to("mock:result");
+----