You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2018/11/08 08:39:42 UTC
[camel] branch master updated (152a650 -> 9cb832f)
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 152a650 Fixed minor stuff in docs
new e713a08 Up to latest version
new a213f0c Add camel-nsq component.
new 9eea161 Add camel-nsq component.
new ccddc07 camel-nsq component can automatically Finish messages.
new 9ddad3f camel-nsq component supports TLS setup.
new 2e98f12 Removed original Synchronization class and corrected test.
new 858a330 Camel-NSQ: Fixed CS and added the starter
new 44aa18e Camel-NSQ: Added the component to kit
new a6b0f31 Camel-NSQ: Regen
new e46bf2d Camel-NSQ: Cleanup the copied docs
new 9cb832f Camel-NSQ: Changed the placeholder name for the client version to avoid confusion with nats client
The 11 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:
apache-camel/src/main/descriptors/common-bin.xml | 2 +
bom/camel-bom/pom.xml | 10 +
.../component/mybatis/MyBatisTestSupport.java | 2 +-
components/camel-nsq/pom.xml | 127 +++++++++++
.../camel-nsq/src/main/docs/nsq-component.adoc | 110 ++++++++++
.../apache/camel/component/nsq/NsqComponent.java | 59 +++++
.../camel/component/nsq/NsqConfiguration.java | 242 +++++++++++++++++++++
.../apache/camel/component/nsq/NsqConstants.java | 28 +++
.../apache/camel/component/nsq/NsqConsumer.java | 124 +++++++++++
.../apache/camel/component/nsq/NsqEndpoint.java | 92 ++++++++
.../apache/camel/component/nsq/NsqProducer.java | 80 +++++++
.../camel/component/nsq/NsqSynchronization.java | 48 ++++
.../src/main/resources/META-INF/LICENSE.txt | 0
.../src/main/resources/META-INF/NOTICE.txt | 0
.../services/org/apache/camel/component/nsq | 18 ++
.../camel/component/nsq/NsqConsumerTest.java | 103 +++++++++
.../camel/component/nsq/NsqProducerTest.java | 96 ++++++++
.../apache/camel/component/nsq/NsqTestSupport.java | 61 ++++++
.../camel-nsq/src/test/resources/log4j2.properties | 28 +++
components/pom.xml | 1 +
components/readme.adoc | 5 +-
docs/user-manual/en/SUMMARY.md | 1 +
parent/pom.xml | 11 +
.../components-starter/camel-nsq-starter/pom.xml | 61 ++++++
.../springboot/NsqComponentAutoConfiguration.java | 128 +++++++++++
.../nsq/springboot/NsqComponentConfiguration.java | 67 ++++++
.../src/main/resources/META-INF/LICENSE.txt | 0
.../src/main/resources/META-INF/NOTICE.txt | 0
.../src/main/resources/META-INF/spring.factories | 19 ++
.../src/main/resources/META-INF/spring.provides | 17 ++
platforms/spring-boot/components-starter/pom.xml | 1 +
.../camel-spring-boot-dependencies/pom.xml | 10 +
32 files changed, 1549 insertions(+), 2 deletions(-)
create mode 100644 components/camel-nsq/pom.xml
create mode 100644 components/camel-nsq/src/main/docs/nsq-component.adoc
create mode 100644 components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
create mode 100644 components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
create mode 100644 components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConstants.java
create mode 100644 components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
create mode 100644 components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
create mode 100644 components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
create mode 100644 components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java
copy {tooling/maven/guice-maven-plugin => components/camel-nsq}/src/main/resources/META-INF/LICENSE.txt (100%)
copy {tooling/maven/guice-maven-plugin => components/camel-nsq}/src/main/resources/META-INF/NOTICE.txt (100%)
create mode 100644 components/camel-nsq/src/main/resources/META-INF/services/org/apache/camel/component/nsq
create mode 100644 components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java
create mode 100644 components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java
create mode 100644 components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java
create mode 100644 components/camel-nsq/src/test/resources/log4j2.properties
create mode 100644 platforms/spring-boot/components-starter/camel-nsq-starter/pom.xml
create mode 100644 platforms/spring-boot/components-starter/camel-nsq-starter/src/main/java/org/apache/camel/component/nsq/springboot/NsqComponentAutoConfiguration.java
create mode 100644 platforms/spring-boot/components-starter/camel-nsq-starter/src/main/java/org/apache/camel/component/nsq/springboot/NsqComponentConfiguration.java
copy {tooling/maven/guice-maven-plugin => platforms/spring-boot/components-starter/camel-nsq-starter}/src/main/resources/META-INF/LICENSE.txt (100%)
copy {tooling/maven/guice-maven-plugin => platforms/spring-boot/components-starter/camel-nsq-starter}/src/main/resources/META-INF/NOTICE.txt (100%)
create mode 100644 platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/spring.factories
create mode 100644 platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/spring.provides
[camel] 05/11: camel-nsq component supports TLS setup.
Posted by ac...@apache.org.
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 9ddad3fad517dfb9a8832435b0fbee3d84330e6a
Author: mionker <mi...@icloud.com>
AuthorDate: Sun Nov 4 22:14:47 2018 +0100
camel-nsq component supports TLS setup.
---
.../camel-nsq/src/main/docs/nsq-component.adoc | 7 ++-
.../apache/camel/component/nsq/NsqComponent.java | 12 +++--
.../camel/component/nsq/NsqConfiguration.java | 57 +++++++++++++++-------
.../apache/camel/component/nsq/NsqEndpoint.java | 16 ++++--
.../apache/camel/component/nsq/NsqProducer.java | 5 +-
.../camel/component/nsq/NsqSynchronization.java | 5 ++
6 files changed, 75 insertions(+), 27 deletions(-)
diff --git a/components/camel-nsq/src/main/docs/nsq-component.adoc b/components/camel-nsq/src/main/docs/nsq-component.adoc
index 39bed9f..ab48f3a 100644
--- a/components/camel-nsq/src/main/docs/nsq-component.adoc
+++ b/components/camel-nsq/src/main/docs/nsq-component.adoc
@@ -78,8 +78,11 @@ with the following path and query parameters:
| *lookupServerPort* (consumer) | The default port number to use to connect to nsqlookupd server | 4161 | int
| *port* (producer) | The default port number to use to connect to nsqd server | 4150 | int
| *lookupInterval* (consumer) | The retry interval in millicseconds to lookup the topic at the nsqlookupq server. | 5000 | int
-| *requeueInterval* (consumer) | The requeue interval. Default of 0 will
-
+| *autoFinish* (consumer) | Automatically finish the NSQ Message when it is retrievd from the queue and before the Exchange is processed | true | boolean
+| *messageTimeout* (consumer) | The NSQ consumer timeout period for messages retrieved from the queue | 0 | int
+| *requeueInterval* (consumer) | The requeue interval. Default of 0 will use the default configured on the server | 0 | int
+| *secure* (security) | Set secure option indicating TLS is required | false | boolean
+| *sslContextParameters* (security) | To configure security using SSLContextParameters | | SSLContextParameters
|===
// endpoint options: END
// spring-boot-auto-configure options: START
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
index eda6413..e3e9d83 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
@@ -16,11 +16,15 @@ public class NsqComponent extends DefaultComponent implements SSLContextParamete
private boolean useGlobalSslContextParameters;
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
- NsqConfiguration config = new NsqConfiguration();
- setProperties(config, parameters);
- config.setServers(remaining);
+ NsqConfiguration configuration = new NsqConfiguration();
+ setProperties(configuration, parameters);
+ configuration.setServers(remaining);
- NsqEndpoint endpoint = new NsqEndpoint(uri, this, config);
+ if (configuration.getSslContextParameters() == null) {
+ configuration.setSslContextParameters(retrieveGlobalSslContextParameters());
+ }
+
+ NsqEndpoint endpoint = new NsqEndpoint(uri, this, configuration);
return endpoint;
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
index 9cdd5c1..a8eb202 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
@@ -32,14 +32,18 @@ public class NsqConfiguration {
private int port = NSQ_DEFAULT_PORT;
@UriParam(label = "consumer", defaultValue = "5000", description = "The lookup interval")
private long lookupInterval = 5000;
- @UriParam(label = "consumer", defaultValue = "0", description = "The requeue interval")
- private long requeueInterval = 0;
+ @UriParam(label = "consumer", defaultValue = "-1", description = "The requeue interval in milliseconds. A value of -1 is the server default")
+ private long requeueInterval = -1;
+ @UriParam(label = "consumer", defaultValue = "true", description = "Automatically finish the NSQ Message when it is retrieved from the queue and before the Exchange is processed")
+ private Boolean autoFinish = true;
+ @UriParam(label = "consumer", defaultValue = "-1", description = "The NSQ consumer timeout period for messages retrieved from the queue. A value of -1 is the server default")
+ private long messageTimeout = -1;
+ @UriParam(description = "")
+ private String userAgent;
+ @UriParam(label = "security")
+ private boolean secure;
@UriParam(label = "security")
private SSLContextParameters sslContextParameters;
- @UriParam(label = "consumer", defaultValue = "true", description = "Automatically finish the NSQ Message when it is retrievd from the queue and before the Exchange is processed.")
- private Boolean autoFinish = true;
- @UriParam(label = "consumer", description = "The NSQ consumer timeout period for messages retrieved from the queue.")
- private long messageTimeout;
/*
* URL a NSQ lookup server hostname.
@@ -155,17 +159,6 @@ public class NsqConfiguration {
}
/**
- * To configure security using SSLContextParameters
- */
- public SSLContextParameters getSslContextParameters() {
- return sslContextParameters;
- }
-
- public void setSslContextParameters(SSLContextParameters sslContextParameters) {
- this.sslContextParameters = sslContextParameters;
- }
-
- /**
* Automatically finish the NSQ message when it is retrieved from the quese and before the Exchange is processed.
*/
public Boolean getAutoFinish() {
@@ -187,6 +180,36 @@ public class NsqConfiguration {
this.messageTimeout = messageTimeout;
}
+ public String getUserAgent() {
+ return userAgent;
+ }
+
+ public void setUserAgent(String userAgent) {
+ this.userAgent = userAgent;
+ }
+
+ /**
+ * Set secure option indicating TLS is required
+ */
+ public boolean isSecure() {
+ return secure;
+ }
+
+ public void setSecure(boolean secure) {
+ this.secure = secure;
+ }
+
+ /**
+ * To configure security using SSLContextParameters
+ */
+ public SSLContextParameters getSslContextParameters() {
+ return sslContextParameters;
+ }
+
+ public void setSslContextParameters(SSLContextParameters sslContextParameters) {
+ this.sslContextParameters = sslContextParameters;
+ }
+
private String splitServers() {
StringBuilder servers = new StringBuilder();
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
index 53c04ac..7787b00 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
@@ -6,9 +6,11 @@ import io.netty.handler.ssl.SslContext;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
+import org.apache.camel.RuntimeCamelException;
import org.apache.camel.impl.DefaultEndpoint;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
+import org.apache.camel.util.ObjectHelper;
import java.io.IOException;
import java.security.GeneralSecurityException;
@@ -17,7 +19,7 @@ import java.util.concurrent.ExecutorService;
/**
* Represents a nsq endpoint.
*/
-@UriEndpoint(firstVersion = "2.21.0", scheme = "nsq", title = "nsq", syntax="nsq:lookupServer",
+@UriEndpoint(firstVersion = "2.22.0", scheme = "nsq", title = "nsq", syntax="nsq:lookupServer",
consumerClass = NsqConsumer.class, label = "messaging")
public class NsqEndpoint extends DefaultEndpoint {
@@ -33,6 +35,9 @@ public class NsqEndpoint extends DefaultEndpoint {
}
public Consumer createConsumer(Processor processor) throws Exception {
+ if (ObjectHelper.isEmpty(configuration.getTopic())) {
+ throw new RuntimeCamelException("Missing required endpoint configuration: topic must be defined for NSQ consumer");
+ }
return new NsqConsumer(this, processor);
}
@@ -50,12 +55,17 @@ public class NsqEndpoint extends DefaultEndpoint {
public NSQConfig getNsqConfig() throws GeneralSecurityException, IOException {
NSQConfig nsqConfig = new NSQConfig();
- if (getNsqConfiguration().getSslContextParameters() != null) {
+
+ if (getNsqConfiguration().getSslContextParameters() != null && getNsqConfiguration().isSecure()) {
SslContext sslContext = new JdkSslContext(getNsqConfiguration().getSslContextParameters().createSSLContext(getCamelContext()), true, null);
nsqConfig.setSslContext(sslContext);
+ }
+ if (configuration.getUserAgent() != null && !configuration.getUserAgent().isEmpty()) {
+ nsqConfig.setUserAgent(configuration.getUserAgent());
}
- if (configuration.getMessageTimeout() > 0) {
+
+ if (configuration.getMessageTimeout() > -1) {
nsqConfig.setMsgTimeout((int) configuration.getMessageTimeout());
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
index 630a2fa..03a34fc 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
@@ -15,9 +15,11 @@ public class NsqProducer extends DefaultProducer {
private static final Logger LOG = LoggerFactory.getLogger(NsqProducer.class);
private NSQProducer producer;
+ private final NsqConfiguration configuration;
public NsqProducer(NsqEndpoint endpoint) {
super(endpoint);
+ this.configuration = endpoint.getNsqConfiguration();
}
@Override
@@ -29,7 +31,7 @@ public class NsqProducer extends DefaultProducer {
public void process(Exchange exchange) throws Exception {
String topic = exchange.getIn().getHeader(NsqConstants.NSQ_MESSAGE_TOPIC,
- getEndpoint().getNsqConfiguration().getTopic(), String.class);
+ configuration.getTopic(), String.class);
LOG.debug("Publishing to topic: {}", topic);
@@ -48,6 +50,7 @@ public class NsqProducer extends DefaultProducer {
producer.addAddress(server.getHost(),
server.getPort() == 0 ? config.getPort() : server.getPort());
}
+ producer.setConfig(getEndpoint().getNsqConfig());
producer.start();
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java
index 58635ee..010e227 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java
@@ -23,4 +23,9 @@ public class NsqSynchronization extends SynchronizationAdapter {
public void onFailure(Exchange exchange) {
nsqMessage.requeue(requeueInterval);
}
+
+ @Override
+ public boolean allowHandover() {
+ return false;
+ }
}
[camel] 09/11: Camel-NSQ: Regen
Posted by ac...@apache.org.
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 a6b0f31daa1ac9cbb008e585ec5d6220b97bd889
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Nov 8 08:57:17 2018 +0100
Camel-NSQ: Regen
---
components/camel-nsq/src/main/docs/nsq-component.adoc | 12 ++++++++++++
components/readme.adoc | 5 ++++-
docs/user-manual/en/SUMMARY.md | 1 +
.../spring-boot-dm/camel-spring-boot-dependencies/pom.xml | 10 ++++++++++
4 files changed, 27 insertions(+), 1 deletion(-)
diff --git a/components/camel-nsq/src/main/docs/nsq-component.adoc b/components/camel-nsq/src/main/docs/nsq-component.adoc
index 203b520..d4ff4eb 100644
--- a/components/camel-nsq/src/main/docs/nsq-component.adoc
+++ b/components/camel-nsq/src/main/docs/nsq-component.adoc
@@ -94,6 +94,18 @@ with the following path and query parameters:
// spring-boot-auto-configure options: START
=== Spring Boot Auto-Configuration
+
+The component supports 3 options, which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *camel.component.nsq.enabled* | Whether to enable auto configuration of the nsq component. This is enabled by default. | | Boolean
+| *camel.component.nsq.resolve-property-placeholders* | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | Boolean
+| *camel.component.nsq.use-global-ssl-context-parameters* | Enable usage of global SSL context parameters. | false | Boolean
+|===
// spring-boot-auto-configure options: END
diff --git a/components/readme.adoc b/components/readme.adoc
index cbb9429..5122b42 100644
--- a/components/readme.adoc
+++ b/components/readme.adoc
@@ -2,7 +2,7 @@ Components
^^^^^^^^^^
// components: START
-Number of Components: 302 in 204 JAR artifacts (22 deprecated)
+Number of Components: 303 in 205 JAR artifacts (22 deprecated)
[width="100%",cols="4,1,5",options="header"]
|===
@@ -614,6 +614,9 @@ Number of Components: 302 in 204 JAR artifacts (22 deprecated)
| link:camel-netty4-http/src/main/docs/netty4-http-component.adoc[Netty4 HTTP] (camel-netty4-http) +
`netty4-http:protocol:host:port/path` | 2.14 | Netty HTTP server and client using the Netty 4.x library.
+| link:camel-nsq/src/main/docs/nsq-component.adoc[nsq] (camel-nsq) +
+`nsq:lookupServer` | 2.22 | Represents a nsq endpoint.
+
| link:camel-olingo2/camel-olingo2-component/src/main/docs/olingo2-component.adoc[Olingo2] (camel-olingo2) +
`olingo2:apiName/methodName` | 2.14 | Communicates with OData 2.0 services using Apache Olingo.
diff --git a/docs/user-manual/en/SUMMARY.md b/docs/user-manual/en/SUMMARY.md
index e5d0ebc..9abdaca 100644
--- a/docs/user-manual/en/SUMMARY.md
+++ b/docs/user-manual/en/SUMMARY.md
@@ -357,6 +357,7 @@
* [Netty HTTP](netty-http-component.adoc)
* [Netty4](netty4-component.adoc)
* [Netty4 HTTP](netty4-http-component.adoc)
+ * [nsq](nsq-component.adoc)
* [Olingo2](olingo2-component.adoc)
* [Olingo4](olingo4-component.adoc)
* [OPC UA Client](milo-client-component.adoc)
diff --git a/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml b/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml
index c83a37b..0b3b6ce 100644
--- a/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml
+++ b/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml
@@ -2056,6 +2056,16 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-nsq</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-nsq-starter</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-ognl</artifactId>
<version>${project.version}</version>
</dependency>
[camel] 04/11: camel-nsq component can automatically Finish
messages.
Posted by ac...@apache.org.
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 ccddc0762de135c3d638723b2149848c937a13e6
Author: mionker <mi...@icloud.com>
AuthorDate: Sun Nov 4 17:36:10 2018 +0100
camel-nsq component can automatically Finish messages.
---
.../camel/component/nsq/NsqConfiguration.java | 26 ++++++++++++
.../apache/camel/component/nsq/NsqConsumer.java | 49 +++++++++++++++++-----
.../apache/camel/component/nsq/NsqEndpoint.java | 10 ++++-
.../camel/component/nsq/NsqSynchronization.java | 26 ++++++++++++
.../camel/component/nsq/NsqConsumerTest.java | 34 +++++++++++++--
5 files changed, 130 insertions(+), 15 deletions(-)
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
index 2c01f99..9cdd5c1 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
@@ -36,6 +36,10 @@ public class NsqConfiguration {
private long requeueInterval = 0;
@UriParam(label = "security")
private SSLContextParameters sslContextParameters;
+ @UriParam(label = "consumer", defaultValue = "true", description = "Automatically finish the NSQ Message when it is retrievd from the queue and before the Exchange is processed.")
+ private Boolean autoFinish = true;
+ @UriParam(label = "consumer", description = "The NSQ consumer timeout period for messages retrieved from the queue.")
+ private long messageTimeout;
/*
* URL a NSQ lookup server hostname.
@@ -161,6 +165,28 @@ public class NsqConfiguration {
this.sslContextParameters = sslContextParameters;
}
+ /**
+ * Automatically finish the NSQ message when it is retrieved from the quese and before the Exchange is processed.
+ */
+ public Boolean getAutoFinish() {
+ return autoFinish;
+ }
+
+ public void setAutoFinish(Boolean autoFinish) {
+ this.autoFinish = autoFinish;
+ }
+
+ /**
+ * The NSQ message timeout for a consumer.
+ */
+ public long getMessageTimeout() {
+ return messageTimeout;
+ }
+
+ public void setMessageTimeout(long messageTimeout) {
+ this.messageTimeout = messageTimeout;
+ }
+
private String splitServers() {
StringBuilder servers = new StringBuilder();
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
index 6402fe5..68faf37 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
@@ -7,8 +7,10 @@ import com.github.brainlag.nsq.callbacks.NSQMessageCallback;
import com.github.brainlag.nsq.lookup.DefaultNSQLookup;
import com.github.brainlag.nsq.lookup.NSQLookup;
import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.spi.Synchronization;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -25,12 +27,12 @@ public class NsqConsumer extends DefaultConsumer {
private ExecutorService executor;
private boolean active;
NSQConsumer consumer;
- private final NsqConfiguration config;
+ private final NsqConfiguration configuration;
public NsqConsumer(NsqEndpoint endpoint, Processor processor) {
super(endpoint, processor);
this.processor = processor;
- this.config = getEndpoint().getNsqConfiguration();
+ this.configuration = getEndpoint().getNsqConfiguration();
}
@Override
@@ -47,14 +49,14 @@ public class NsqConsumer extends DefaultConsumer {
LOG.debug("Getting NSQ Connection");
NSQLookup lookup = new DefaultNSQLookup();
- for(ServerAddress server : config.getServerAddresses()) {
+ for(ServerAddress server : configuration.getServerAddresses()) {
lookup.addLookupAddress(server.getHost(),
- server.getPort() == 0 ? config.getLookupServerPort() : server.getPort());
+ server.getPort() == 0 ? configuration.getLookupServerPort() : server.getPort());
}
- consumer = new NSQConsumer(lookup, config.getTopic(),
- config.getChannel(), new CamelNsqMessageHandler());
- consumer.setLookupPeriod(config.getLookupInterval());
+ consumer = new NSQConsumer(lookup, configuration.getTopic(),
+ configuration.getChannel(), new CamelNsqMessageHandler(), getEndpoint().getNsqConfig());
+ consumer.setLookupPeriod(configuration.getLookupInterval());
consumer.setExecutor(getEndpoint().createExecutor());
consumer.start();
}
@@ -84,18 +86,45 @@ public class NsqConsumer extends DefaultConsumer {
@Override
public void message(NSQMessage msg) {
LOG.debug("Received Message: {}", msg);
- Exchange exchange = getEndpoint().createExchange();
+ Exchange exchange = getEndpoint().createExchange(ExchangePattern.InOnly);
exchange.getIn().setBody(msg.getMessage());
exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ID, msg.getId());
exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ATTEMPTS, msg.getAttempts());
exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_TIMESTAMP, msg.getTimestamp());
try {
+ if (configuration.getAutoFinish()) {
+ msg.finished();
+ } else {
+ exchange.addOnCompletion(new NsqSynchronization(msg, (int) configuration.getRequeueInterval()));
+ }
processor.process(exchange);
- msg.finished();
} catch (Exception e) {
- msg.requeue((int) config.getRequeueInterval());
+ if (!configuration.getAutoFinish()) {
+ msg.requeue((int) configuration.getRequeueInterval());
+ }
getExceptionHandler().handleException("Error during processing", exchange, e);
}
}
}
+
+ class Sync implements Synchronization {
+
+ @Override
+ public void onComplete(final Exchange exchange) {
+ try {
+ //msg.finished();
+ } catch (Exception e) {
+ LOG.error(String.format("Could not run completion of exchange %s", exchange), e);
+ }
+ }
+
+ @Override
+ public void onFailure(final Exchange exchange) {
+ try {
+ //msg.requeue((int) config.getRequeueInterval());
+ } catch (Exception e) {
+ LOG.error(String.format("Could not run failure of exchange %s", exchange), e);
+ }
+ }
+ }
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
index e792316..53c04ac 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
@@ -50,8 +50,14 @@ public class NsqEndpoint extends DefaultEndpoint {
public NSQConfig getNsqConfig() throws GeneralSecurityException, IOException {
NSQConfig nsqConfig = new NSQConfig();
- SslContext sslContext = new JdkSslContext(getNsqConfiguration().getSslContextParameters().createSSLContext(getCamelContext()), true, null);
- nsqConfig.setSslContext(sslContext);
+ if (getNsqConfiguration().getSslContextParameters() != null) {
+ SslContext sslContext = new JdkSslContext(getNsqConfiguration().getSslContextParameters().createSSLContext(getCamelContext()), true, null);
+ nsqConfig.setSslContext(sslContext);
+
+ }
+ if (configuration.getMessageTimeout() > 0) {
+ nsqConfig.setMsgTimeout((int) configuration.getMessageTimeout());
+ }
return nsqConfig;
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java
new file mode 100644
index 0000000..58635ee
--- /dev/null
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java
@@ -0,0 +1,26 @@
+package org.apache.camel.component.nsq;
+
+import com.github.brainlag.nsq.NSQMessage;
+import org.apache.camel.Exchange;
+import org.apache.camel.support.SynchronizationAdapter;
+
+public class NsqSynchronization extends SynchronizationAdapter {
+
+ private final NSQMessage nsqMessage;
+ private final int requeueInterval;
+
+ public NsqSynchronization(NSQMessage nsqMessage, int requeueInterval) {
+ this.nsqMessage = nsqMessage;
+ this.requeueInterval = requeueInterval;
+ }
+
+ @Override
+ public void onComplete(Exchange exchange) {
+ nsqMessage.finished();
+ }
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ nsqMessage.requeue(requeueInterval);
+ }
+}
diff --git a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java
index fca0ea3..7c15fdd 100644
--- a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java
+++ b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java
@@ -3,6 +3,8 @@ package org.apache.camel.component.nsq;
import com.github.brainlag.nsq.NSQProducer;
import com.github.brainlag.nsq.exceptions.NSQException;
import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.Test;
@@ -12,6 +14,7 @@ import java.util.concurrent.TimeoutException;
public class NsqConsumerTest extends NsqTestSupport {
private static final int NUMBER_OF_MESSAGES = 10000;
+ private static final String TOPIC = "test";
@EndpointInject(uri = "mock:result")
protected MockEndpoint mockResultEndpoint;
@@ -25,7 +28,7 @@ public class NsqConsumerTest extends NsqTestSupport {
producer.addAddress("localhost", 4150);
producer.start();
- producer.produce("test", ("Hello NSQ!").getBytes());
+ producer.produce(TOPIC, ("Hello NSQ!").getBytes());
mockResultEndpoint.assertIsSatisfied();
@@ -42,18 +45,43 @@ public class NsqConsumerTest extends NsqTestSupport {
producer.start();
for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
- producer.produce("test", ("test" + i).getBytes());
+ producer.produce(TOPIC, (String.format("Hello NSQ%d!", i)).getBytes());
}
mockResultEndpoint.assertIsSatisfied();
}
+ @Test
+ public void testRequeue() throws NSQException, TimeoutException, InterruptedException {
+ mockResultEndpoint.setExpectedMessageCount(1);
+ mockResultEndpoint.setAssertPeriod(5000);
+
+ NSQProducer producer = new NSQProducer();
+ producer.addAddress("localhost", 4150);
+ producer.start();
+
+ producer.produce(TOPIC, ("Test Requeue").getBytes());
+
+ mockResultEndpoint.assertIsSatisfied();
+ }
+
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- fromF("nsq://%s?topic=%s&lookupInterval=5s", getNsqConsumerUrl(), "test").to(mockResultEndpoint);
+ fromF("nsq://%s?topic=%s&lookupInterval=2s&autoFinish=false&requeueInterval=1s", getNsqConsumerUrl(), TOPIC)
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ String messageText = exchange.getIn().getBody(String.class);
+ int attempts = exchange.getIn().getHeader(NsqConstants.NSQ_MESSAGE_ATTEMPTS, Integer.class);
+ if (messageText.contains("Requeue") && attempts < 3) {
+ throw new Exception();
+ }
+ }
+ })
+ .to(mockResultEndpoint);
}
};
}
[camel] 11/11: Camel-NSQ: Changed the placeholder name for the
client version to avoid confusion with nats client
Posted by ac...@apache.org.
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 9cb832f5840b15cb1b36128d4ec5264a35a58ac7
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Nov 8 09:39:06 2018 +0100
Camel-NSQ: Changed the placeholder name for the client version to avoid confusion with nats client
---
components/camel-nsq/pom.xml | 4 ++--
parent/pom.xml | 2 +-
2 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/components/camel-nsq/pom.xml b/components/camel-nsq/pom.xml
index 0ac97c5..0953f5df 100644
--- a/components/camel-nsq/pom.xml
+++ b/components/camel-nsq/pom.xml
@@ -27,7 +27,7 @@
<dependency>
<groupId>com.github.brainlag</groupId>
<artifactId>nsq-client</artifactId>
- <version>${nats-client-version}</version>
+ <version>${nsq-client-version}</version>
</dependency>
<!-- testing -->
<dependency>
@@ -124,4 +124,4 @@
</build>
</profile>
</profiles>
-</project>
\ No newline at end of file
+</project>
diff --git a/parent/pom.xml b/parent/pom.xml
index 12bd4ff..68587e0 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -515,7 +515,7 @@
<mvel-version>2.4.2.Final</mvel-version>
<mybatis-version>3.4.6</mybatis-version>
<narayana-version>5.9.0.Final</narayana-version>
- <nats-client-version>1.0.0.RC4</nats-client-version>
+ <nsq-client-version>1.0.0.RC4</nsq-client-version>
<neethi-bundle-version>3.0.1</neethi-bundle-version>
<nekohtml-version>1.9.22</nekohtml-version>
<neoscada-version>0.4.0</neoscada-version>
[camel] 02/11: Add camel-nsq component.
Posted by ac...@apache.org.
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 a213f0cb6b444f851b3f54e9721a2ff62dc48760
Author: mionker <mi...@icloud.com>
AuthorDate: Tue Oct 30 21:30:18 2018 +0100
Add camel-nsq component.
---
bom/camel-bom/pom.xml | 5 +
components/camel-nsq/pom.xml | 127 +++++++++++++
.../apache/camel/component/nsq/NsqComponent.java | 23 +++
.../camel/component/nsq/NsqConfiguration.java | 164 +++++++++++++++++
.../apache/camel/component/nsq/NsqConstants.java | 12 ++
.../apache/camel/component/nsq/NsqConsumer.java | 101 ++++++++++
.../apache/camel/component/nsq/NsqEndpoint.java | 45 +++++
.../apache/camel/component/nsq/NsqProducer.java | 60 ++++++
.../src/main/resources/META-INF/LICENSE.txt | 203 +++++++++++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 ++
.../services/org/apache/camel/component/nsq | 18 ++
.../camel/component/nsq/NsqConsumerTest.java | 60 ++++++
.../camel/component/nsq/NsqProducerTest.java | 95 ++++++++++
.../apache/camel/component/nsq/NsqTestSupport.java | 74 ++++++++
components/pom.xml | 1 +
parent/pom.xml | 1 +
16 files changed, 1000 insertions(+)
diff --git a/bom/camel-bom/pom.xml b/bom/camel-bom/pom.xml
index 6832001..655f9d9 100644
--- a/bom/camel-bom/pom.xml
+++ b/bom/camel-bom/pom.xml
@@ -1855,6 +1855,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-nsq</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-ognl</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/components/camel-nsq/pom.xml b/components/camel-nsq/pom.xml
new file mode 100644
index 0000000..0ac97c5
--- /dev/null
+++ b/components/camel-nsq/pom.xml
@@ -0,0 +1,127 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>components</artifactId>
+ <groupId>org.apache.camel</groupId>
+ <version>2.23.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>camel-nsq</artifactId>
+ <packaging>jar</packaging>
+ <name>Camel :: NSQ</name>
+
+ <properties>
+ <camel.osgi.export.pkg>org.apache.camel.component.nsq.*</camel.osgi.export.pkg>
+ <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=nsq</camel.osgi.export.service>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+ <!-- nsq client -->
+ <dependency>
+ <groupId>com.github.brainlag</groupId>
+ <artifactId>nsq-client</artifactId>
+ <version>${nats-client-version}</version>
+ </dependency>
+ <!-- testing -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <!-- logging -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-testcontainers</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+
+
+ <profiles>
+ <profile>
+ <id>nsq-skip-tests</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>true</skipTests>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+
+ <!-- activate test if the docker socket file is accessible -->
+ <profile>
+ <id>nsq-tests-docker-file</id>
+ <activation>
+ <file>
+ <exists>/var/run/docker.sock</exists>
+ </file>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>${skipTests}</skipTests>
+ <systemPropertyVariables>
+ <visibleassertions.silence>true</visibleassertions.silence>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+
+ <!-- activate test if the DOCKER_HOST env var is set -->
+ <profile>
+ <id>nsq-tests-docker-env</id>
+ <activation>
+ <property>
+ <name>env.DOCKER_HOST</name>
+ </property>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>${skipTests}</skipTests>
+ <systemPropertyVariables>
+ <visibleassertions.silence>true</visibleassertions.silence>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+</project>
\ No newline at end of file
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
new file mode 100644
index 0000000..6c92fd4
--- /dev/null
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
@@ -0,0 +1,23 @@
+package org.apache.camel.component.nsq;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.DefaultComponent;
+
+import java.util.Map;
+
+import static org.apache.camel.util.IntrospectionSupport.setProperties;
+
+/**
+ * Represents the component that manages {@link NsqEndpoint}.
+ */
+public class NsqComponent extends DefaultComponent {
+
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ NsqConfiguration config = new NsqConfiguration();
+ setProperties(config, parameters);
+ config.setServers(remaining);
+
+ NsqEndpoint endpoint = new NsqEndpoint(uri, this, config);
+ return endpoint;
+ }
+}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
new file mode 100644
index 0000000..953c7d7
--- /dev/null
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
@@ -0,0 +1,164 @@
+package org.apache.camel.component.nsq;
+
+import com.github.brainlag.nsq.ServerAddress;
+import com.google.common.collect.Sets;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.apache.camel.spi.UriPath;
+
+import java.util.Set;
+
+import static org.apache.camel.component.nsq.NsqConstants.NSQ_DEFAULT_LOOKUP_PORT;
+import static org.apache.camel.component.nsq.NsqConstants.NSQ_DEFAULT_PORT;
+
+@UriParams
+public class NsqConfiguration {
+
+ @UriPath(description = "The hostnames of one or more nsqlookupd servers (consumer) or nsqd servers (producer).")
+ @Metadata(required = "true")
+ private String servers;
+ @UriParam(description = "The NSQ topic")
+ @Metadata(required = "true")
+ private String topic;
+ @UriParam(label = "consumer", description = "The NSQ channel")
+ private String channel;
+ @UriParam(label = "consumer", defaultValue = "10")
+ private int poolSize = 10;
+ @UriParam(label = "consumer", defaultValue = "4161", description = "The NSQ lookup server port")
+ private int lookupServerPort = NSQ_DEFAULT_LOOKUP_PORT;
+ @UriParam(label = "producer", defaultValue = "4150")
+ private int port = NSQ_DEFAULT_PORT;
+ @UriParam(label = "consumer", defaultValue = "5000", description = "The lookup interval")
+ private long lookupInterval = 5000;
+ @UriParam(label = "consumer", defaultValue = "0", description = "The requeue interval")
+ private long requeueInterval = 0;
+
+ /*
+ * URL a NSQ lookup server hostname.
+ */
+ public String getServers() {
+ return servers;
+ }
+
+ public void setServers(String servers) {
+ this.servers = servers;
+ }
+
+ public Set<ServerAddress> getServerAddresses() {
+
+ Set<ServerAddress> serverAddresses = Sets.newConcurrentHashSet();
+
+ String[] addresses = servers.split(",");
+ for (int i = 0; i < addresses.length; i++) {
+ String[] token = addresses[i].split(":");
+ String host;
+ int port;
+ if (token.length == 2) {
+ host = token[0];
+ port = Integer.parseInt(token[1]);
+
+ } else if (token.length == 1) {
+ host = token[0];
+ port = 0;
+
+ } else {
+ throw new IllegalArgumentException("Invalid address: " + addresses[i]);
+ }
+ serverAddresses.add(new ServerAddress(host, port));
+ }
+ return serverAddresses;
+ }
+
+ /**
+ * The name of topic we want to use
+ */
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ /**
+ * The name of channel we want to use
+ */
+ public String getChannel() {
+ return channel;
+ }
+
+ public void setChannel(String channel) {
+ this.channel = channel;
+ }
+
+ /**
+ * Consumer pool size
+ */
+ public int getPoolSize() {
+ return poolSize;
+ }
+
+ public void setPoolSize(int poolSize) {
+ this.poolSize = poolSize;
+ }
+
+ /**
+ * The port of the nsqdlookupd server
+ */
+ public int getLookupServerPort() {
+ return lookupServerPort;
+ }
+
+ public void setLookupServerPort(int lookupServerPort) {
+ this.lookupServerPort = lookupServerPort;
+ }
+
+ /**
+ * The port of the nsqd server
+ */
+ public int getPort() {
+ return port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ /**
+ * The lookup retry interval
+ */
+ public long getLookupInterval() {
+ return lookupInterval;
+ }
+
+ public void setLookupInterval(long lookupInterval) {
+ this.lookupInterval = lookupInterval;
+ }
+
+ /**
+ * The requeue retry interval
+ */
+ public long getRequeueInterval() {
+ return requeueInterval;
+ }
+
+ public void setRequeueInterval(long requeueInterval) {
+ this.requeueInterval = requeueInterval;
+ }
+
+
+ private String splitServers() {
+ StringBuilder servers = new StringBuilder();
+
+ String[] pieces = getServers().split(",");
+ for (int i = 0; i < pieces.length; i++) {
+ if (i < pieces.length - 1) {
+ servers.append(pieces[i] + ",");
+ } else {
+ servers.append(pieces[i]);
+ }
+ }
+ return servers.toString();
+ }
+}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConstants.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConstants.java
new file mode 100644
index 0000000..b1c1aa2
--- /dev/null
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConstants.java
@@ -0,0 +1,12 @@
+package org.apache.camel.component.nsq;
+
+public class NsqConstants {
+
+ static int NSQ_DEFAULT_PORT = 4150;
+ static int NSQ_DEFAULT_LOOKUP_PORT = 4160;
+ static int NSQ_DEFAULT_LOOKUP_PORT_HTTP = 4161;
+ static String NSQ_MESSAGE_ID = "CamelNsqMessageId";
+ static String NSQ_MESSAGE_ATTEMPTS = "CamelNsqMessageAttempts";
+ static String NSQ_MESSAGE_TIMESTAMP = "CamelNsqMessageTimestamp";
+ static String NSQ_MESSAGE_TOPIC = "CamelNsqMessageTopic";
+}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
new file mode 100644
index 0000000..f41a888
--- /dev/null
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
@@ -0,0 +1,101 @@
+package org.apache.camel.component.nsq;
+
+import com.github.brainlag.nsq.NSQConsumer;
+import com.github.brainlag.nsq.NSQMessage;
+import com.github.brainlag.nsq.ServerAddress;
+import com.github.brainlag.nsq.callbacks.NSQMessageCallback;
+import com.github.brainlag.nsq.lookup.DefaultNSQLookup;
+import com.github.brainlag.nsq.lookup.NSQLookup;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.impl.DefaultConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutorService;
+
+/**
+ * The nsq consumer.
+ */
+public class NsqConsumer extends DefaultConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NsqConsumer.class);
+
+ private final Processor processor;
+ private ExecutorService executor;
+ private boolean active;
+ NSQConsumer consumer;
+ private final NsqConfiguration config;
+
+ public NsqConsumer(NsqEndpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ this.processor = processor;
+ this.config = getEndpoint().getNsqConfiguration();
+ }
+
+ @Override
+ public NsqEndpoint getEndpoint() {
+ return (NsqEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ LOG.debug("Starting NSQ Consumer");
+ executor = getEndpoint().createExecutor();
+
+ LOG.debug("Getting NSQ Connection");
+ NSQLookup lookup = new DefaultNSQLookup();
+
+ for(ServerAddress server : config.getServerAddresses()) {
+ lookup.addLookupAddress(server.getHost(),
+ server.getPort() == 0 ? config.getLookupServerPort() : server.getPort());
+ }
+
+ consumer = new NSQConsumer(lookup, config.getTopic(),
+ config.getChannel(), new CamelNsqMessageHandler());
+ consumer.setLookupPeriod(config.getLookupInterval());
+ consumer.setExecutor(getEndpoint().createExecutor());
+ consumer.start();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+
+ LOG.debug("Stopping NSQ Consumer");
+ if (consumer != null) {
+ consumer.shutdown();
+ }
+ if (executor != null) {
+ if (getEndpoint() != null && getEndpoint().getCamelContext() != null) {
+ getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
+ } else {
+ executor.shutdownNow();
+ }
+ }
+
+ executor = null;
+
+ super.doStop();
+ }
+
+ class CamelNsqMessageHandler implements NSQMessageCallback {
+
+ @Override
+ public void message(NSQMessage msg) {
+ LOG.debug("Received Message: {}", msg);
+ Exchange exchange = getEndpoint().createExchange();
+ exchange.getIn().setBody(new String(msg.getMessage()));
+ exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ID, msg.getId());
+ exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ATTEMPTS, msg.getAttempts());
+ exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_TIMESTAMP, msg.getTimestamp());
+ try {
+ processor.process(exchange);
+ msg.finished();
+ } catch (Exception e) {
+ msg.requeue((int) config.getRequeueInterval());
+ getExceptionHandler().handleException("Error during processing", exchange, e);
+ }
+ }
+ }
+}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
new file mode 100644
index 0000000..1b25880
--- /dev/null
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
@@ -0,0 +1,45 @@
+package org.apache.camel.component.nsq;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+
+import java.util.concurrent.ExecutorService;
+
+/**
+ * Represents a nsq endpoint.
+ */
+@UriEndpoint(firstVersion = "2.21.0", scheme = "nsq", title = "nsq", syntax="nsq:lookupServer",
+ consumerClass = NsqConsumer.class, label = "messaging")
+public class NsqEndpoint extends DefaultEndpoint {
+
+ @UriParam
+ private NsqConfiguration configuration;
+
+ public NsqEndpoint(String uri, NsqComponent component, NsqConfiguration configuration) {
+ super(uri, component);
+ this.configuration = configuration;
+ }
+ public Producer createProducer() throws Exception {
+ return new NsqProducer(this);
+ }
+
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return new NsqConsumer(this, processor);
+ }
+
+ public ExecutorService createExecutor() {
+ return getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, "NsqTopic[" + configuration.getTopic() + "]", configuration.getPoolSize());
+ }
+
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public NsqConfiguration getNsqConfiguration() {
+ return configuration;
+ }
+}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
new file mode 100644
index 0000000..a886eae
--- /dev/null
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
@@ -0,0 +1,60 @@
+package org.apache.camel.component.nsq;
+
+import com.github.brainlag.nsq.NSQProducer;
+import com.github.brainlag.nsq.ServerAddress;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The nsq producer.
+ */
+public class NsqProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NsqProducer.class);
+
+ private NSQProducer producer;
+
+ public NsqProducer(NsqEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public NsqEndpoint getEndpoint() {
+ return (NsqEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+
+ String topic = exchange.getIn().getHeader(NsqConstants.NSQ_MESSAGE_TOPIC,
+ getEndpoint().getNsqConfiguration().getTopic(), String.class);
+
+ LOG.debug("Publishing to topic: {}", topic);
+
+ String body = exchange.getIn().getBody(String.class);
+ producer.produce(topic, body.getBytes());
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ LOG.debug("Starting NSQ Producer");
+
+ NsqConfiguration config = getEndpoint().getNsqConfiguration();
+ producer = new NSQProducer();
+ for(ServerAddress server : config.getServerAddresses()) {
+ producer.addAddress(server.getHost(),
+ server.getPort() == 0 ? config.getPort() : server.getPort());
+ }
+ producer.start();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ LOG.debug("Stopping NSQ Producer");
+ if (producer != null) { producer.shutdown(); }
+ super.doStop();
+ }
+}
diff --git a/components/camel-nsq/src/main/resources/META-INF/LICENSE.txt b/components/camel-nsq/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/components/camel-nsq/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
diff --git a/components/camel-nsq/src/main/resources/META-INF/NOTICE.txt b/components/camel-nsq/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/components/camel-nsq/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
diff --git a/components/camel-nsq/src/main/resources/META-INF/services/org/apache/camel/component/nsq b/components/camel-nsq/src/main/resources/META-INF/services/org/apache/camel/component/nsq
new file mode 100644
index 0000000..df34246
--- /dev/null
+++ b/components/camel-nsq/src/main/resources/META-INF/services/org/apache/camel/component/nsq
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+class=org.apache.camel.component.nsq.NsqComponent
diff --git a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java
new file mode 100644
index 0000000..fca0ea3
--- /dev/null
+++ b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java
@@ -0,0 +1,60 @@
+package org.apache.camel.component.nsq;
+
+import com.github.brainlag.nsq.NSQProducer;
+import com.github.brainlag.nsq.exceptions.NSQException;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Test;
+
+import java.util.concurrent.TimeoutException;
+
+public class NsqConsumerTest extends NsqTestSupport {
+
+ private static final int NUMBER_OF_MESSAGES = 10000;
+
+ @EndpointInject(uri = "mock:result")
+ protected MockEndpoint mockResultEndpoint;
+
+ @Test
+ public void testConsumer() throws NSQException, TimeoutException, InterruptedException {
+ mockResultEndpoint.expectedMessageCount(1);
+ mockResultEndpoint.setAssertPeriod(5000);
+
+ NSQProducer producer = new NSQProducer();
+ producer.addAddress("localhost", 4150);
+ producer.start();
+
+ producer.produce("test", ("Hello NSQ!").getBytes());
+
+ mockResultEndpoint.assertIsSatisfied();
+
+ assertEquals("Hello NSQ!", mockResultEndpoint.getReceivedExchanges().get(0).getIn().getBody(String.class));
+ }
+
+ @Test
+ public void testLoadConsumer() throws NSQException, TimeoutException, InterruptedException {
+ mockResultEndpoint.setExpectedMessageCount(NUMBER_OF_MESSAGES);
+ mockResultEndpoint.setAssertPeriod(5000);
+
+ NSQProducer producer = new NSQProducer();
+ producer.addAddress("localhost", 4150);
+ producer.start();
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ producer.produce("test", ("test" + i).getBytes());
+ }
+
+ mockResultEndpoint.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ fromF("nsq://%s?topic=%s&lookupInterval=5s", getNsqConsumerUrl(), "test").to(mockResultEndpoint);
+ }
+ };
+ }
+}
diff --git a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java
new file mode 100644
index 0000000..7623474
--- /dev/null
+++ b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java
@@ -0,0 +1,95 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.nsq;
+
+import com.github.brainlag.nsq.NSQConsumer;
+import com.github.brainlag.nsq.lookup.DefaultNSQLookup;
+import com.github.brainlag.nsq.lookup.NSQLookup;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class NsqProducerTest extends NsqTestSupport {
+
+ private static final int NUMBER_OF_MESSAGES = 10000;
+ private static final String TEST_MESSAGE = "Hello NSQProducer!";
+
+ @Test
+ public void testProducer() throws Exception {
+
+ CountDownLatch lock = new CountDownLatch(1);
+
+ template.sendBody("direct:send", TEST_MESSAGE);
+
+ AtomicInteger counter = new AtomicInteger(0);
+ NSQLookup lookup = new DefaultNSQLookup();
+ lookup.addLookupAddress("localhost", 4161);
+
+ NSQConsumer consumer = new NSQConsumer(lookup, "test", "testconsumer", (message) -> {
+ counter.incrementAndGet();
+ message.finished();
+ lock.countDown();
+ assertTrue(new String(message.getMessage()) == TEST_MESSAGE);
+ });
+ consumer.start();
+
+ lock.await(30, TimeUnit.SECONDS);
+
+ assertTrue(counter.get() == Long.valueOf(1));
+ consumer.shutdown();
+ }
+
+ @Test
+ public void testLoadProducer() throws Exception {
+
+ CountDownLatch lock = new CountDownLatch(NUMBER_OF_MESSAGES);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+ template.sendBody("direct:send", ("test" + i));
+ }
+
+ AtomicInteger counter = new AtomicInteger(0);
+ NSQLookup lookup = new DefaultNSQLookup();
+ lookup.addLookupAddress("localhost", 4161);
+
+ NSQConsumer consumer = new NSQConsumer(lookup, "test", "testconsumer", (message) -> {
+ counter.incrementAndGet();
+ message.finished();
+ lock.countDown();
+ assertTrue(message.getAttempts() == 1);
+ });
+ consumer.start();
+
+ lock.await(30, TimeUnit.SECONDS);
+
+ assertTrue(counter.get() == Long.valueOf(NUMBER_OF_MESSAGES));
+ consumer.shutdown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:send").to("nsq://" + getNsqProducerUrl() + "?topic=test");
+ }
+ };
+ }
+}
diff --git a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java
new file mode 100644
index 0000000..e210529
--- /dev/null
+++ b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java
@@ -0,0 +1,74 @@
+package org.apache.camel.component.nsq;
+
+import org.apache.camel.test.testcontainers.ContainerAwareTestSupport;
+import org.testcontainers.containers.FixedHostPortGenericContainer;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class NsqTestSupport extends ContainerAwareTestSupport {
+
+ public static final String CONTAINER_NSQLOOKUPD_IMAGE = "nsqio/nsq";
+ public static final String CONTAINER_NSQLOOKUPD_NAME = "nsqlookupd";
+
+ public static final String CONTAINER_NSQD_IMAGE = "nsqio/nsq";
+ public static final String CONTAINER_NSQD_NAME = "nsqd";
+
+ Network network = Network.newNetwork();
+
+ @Override
+ protected List<GenericContainer<?>> createContainers() {
+ return new ArrayList<GenericContainer<?>>(
+ Arrays.asList(
+ new FixedHostPortGenericContainer<>(CONTAINER_NSQLOOKUPD_IMAGE)
+ .withFixedExposedPort(4160, 4160)
+ .withFixedExposedPort(4161, 4161)
+ .withNetworkAliases(CONTAINER_NSQLOOKUPD_NAME)
+ .withCommand("/nsqlookupd").withNetwork(network),
+ new FixedHostPortGenericContainer<>(CONTAINER_NSQD_IMAGE)
+ .withFixedExposedPort(4150, 4150)
+ .withFixedExposedPort(4151, 4151)
+ .withNetworkAliases(CONTAINER_NSQD_NAME)
+ .withCommand(String.format("/nsqd --broadcast-address=%s --lookupd-tcp-address=%s:4160",
+ "localhost", CONTAINER_NSQLOOKUPD_NAME)).withNetwork(network)
+ ));
+ }
+
+ public static GenericContainer<?> nsqlookupdContainer() {
+ return new FixedHostPortGenericContainer(CONTAINER_NSQLOOKUPD_IMAGE)
+ .withFixedExposedPort(4160, 4160)
+ .withFixedExposedPort(4161, 4161)
+ .withNetworkAliases(CONTAINER_NSQLOOKUPD_NAME)
+ .withCommand("/nsqlookupd");
+ }
+
+ public static GenericContainer<?> nsqdContainer() {
+ return new FixedHostPortGenericContainer(CONTAINER_NSQD_IMAGE)
+ .withFixedExposedPort(4150, 4150)
+ .withFixedExposedPort(4151, 4151)
+ .withNetworkAliases(CONTAINER_NSQD_NAME)
+ .withCommand(String.format("/nsqd --broadcast-address=%s --lookupd-tcp-address=%s:4160",
+ "localhost", CONTAINER_NSQLOOKUPD_NAME));
+ }
+
+ public String getNsqConsumerUrl() {
+ String format = String.format(
+ "%s:%d", "localhost", 4161
+ //getContainerHost(CONTAINER_NSQLOOKUPD_NAME),
+ //getContainerPort(CONTAINER_NSQLOOKUPD_NAME, NsqConstants.NSQ_DEFAULT_LOOKUP_PORT_HTTP)
+ );
+ return format;
+ }
+
+ public String getNsqProducerUrl() {
+ String format = String.format(
+ "%s:%d", "localhost", 4150
+ //getContainerHost(CONTAINER_NSQD_NAME),
+ //getContainerPort(CONTAINER_NSQD_NAME, NsqConstants.NSQ_DEFAULT_PORT)
+ );
+ return format;
+ }
+}
diff --git a/components/pom.xml b/components/pom.xml
index 8ceaaee..4356c93 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -229,6 +229,7 @@
<module>camel-netty4</module>
<module>camel-netty-http</module>
<module>camel-netty4-http</module>
+ <module>camel-nsq</module>
<module>camel-ognl</module>
<module>camel-olingo2</module>
<module>camel-olingo4</module>
diff --git a/parent/pom.xml b/parent/pom.xml
index 398800c..334bf70 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -515,6 +515,7 @@
<mvel-version>2.4.2.Final</mvel-version>
<mybatis-version>3.4.6</mybatis-version>
<narayana-version>5.9.0.Final</narayana-version>
+ <nats-client-version>1.0.0.RC4</nats-client-version>
<neethi-bundle-version>3.0.1</neethi-bundle-version>
<nekohtml-version>1.9.22</nekohtml-version>
<neoscada-version>0.4.0</neoscada-version>
[camel] 06/11: Removed original Synchronization class and corrected
test.
Posted by ac...@apache.org.
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 2e98f129a96028e32b38bfb5bf3725a984e033bb
Author: mionker <mi...@icloud.com>
AuthorDate: Wed Nov 7 21:02:34 2018 +0100
Removed original Synchronization class and corrected test.
---
.../apache/camel/component/nsq/NsqConsumer.java | 22 ------------
.../camel/component/nsq/NsqProducerTest.java | 2 +-
.../apache/camel/component/nsq/NsqTestSupport.java | 42 +++++++---------------
3 files changed, 13 insertions(+), 53 deletions(-)
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
index 68faf37..2beaa96 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
@@ -10,7 +10,6 @@ import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;
-import org.apache.camel.spi.Synchronization;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -106,25 +105,4 @@ public class NsqConsumer extends DefaultConsumer {
}
}
}
-
- class Sync implements Synchronization {
-
- @Override
- public void onComplete(final Exchange exchange) {
- try {
- //msg.finished();
- } catch (Exception e) {
- LOG.error(String.format("Could not run completion of exchange %s", exchange), e);
- }
- }
-
- @Override
- public void onFailure(final Exchange exchange) {
- try {
- //msg.requeue((int) config.getRequeueInterval());
- } catch (Exception e) {
- LOG.error(String.format("Could not run failure of exchange %s", exchange), e);
- }
- }
- }
}
diff --git a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java
index 7623474..32175fe 100644
--- a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java
+++ b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java
@@ -46,7 +46,7 @@ public class NsqProducerTest extends NsqTestSupport {
counter.incrementAndGet();
message.finished();
lock.countDown();
- assertTrue(new String(message.getMessage()) == TEST_MESSAGE);
+ assertTrue(new String(message.getMessage()).equals(TEST_MESSAGE));
});
consumer.start();
diff --git a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java
index e210529..2219397 100644
--- a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java
+++ b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java
@@ -17,58 +17,40 @@ public class NsqTestSupport extends ContainerAwareTestSupport {
public static final String CONTAINER_NSQD_IMAGE = "nsqio/nsq";
public static final String CONTAINER_NSQD_NAME = "nsqd";
- Network network = Network.newNetwork();
+ Network network;
@Override
protected List<GenericContainer<?>> createContainers() {
+ network = Network.newNetwork();
return new ArrayList<GenericContainer<?>>(
Arrays.asList(
- new FixedHostPortGenericContainer<>(CONTAINER_NSQLOOKUPD_IMAGE)
- .withFixedExposedPort(4160, 4160)
- .withFixedExposedPort(4161, 4161)
- .withNetworkAliases(CONTAINER_NSQLOOKUPD_NAME)
- .withCommand("/nsqlookupd").withNetwork(network),
- new FixedHostPortGenericContainer<>(CONTAINER_NSQD_IMAGE)
- .withFixedExposedPort(4150, 4150)
- .withFixedExposedPort(4151, 4151)
- .withNetworkAliases(CONTAINER_NSQD_NAME)
- .withCommand(String.format("/nsqd --broadcast-address=%s --lookupd-tcp-address=%s:4160",
- "localhost", CONTAINER_NSQLOOKUPD_NAME)).withNetwork(network)
+ nsqlookupdContainer(network),
+ nsqdContainer(network)
));
}
- public static GenericContainer<?> nsqlookupdContainer() {
- return new FixedHostPortGenericContainer(CONTAINER_NSQLOOKUPD_IMAGE)
+ public static GenericContainer<?> nsqlookupdContainer(Network network) {
+ return new FixedHostPortGenericContainer<>(CONTAINER_NSQLOOKUPD_IMAGE)
.withFixedExposedPort(4160, 4160)
.withFixedExposedPort(4161, 4161)
.withNetworkAliases(CONTAINER_NSQLOOKUPD_NAME)
- .withCommand("/nsqlookupd");
+ .withCommand("/nsqlookupd").withNetwork(network);
}
- public static GenericContainer<?> nsqdContainer() {
- return new FixedHostPortGenericContainer(CONTAINER_NSQD_IMAGE)
+ public static GenericContainer<?> nsqdContainer(Network network) {
+ return new FixedHostPortGenericContainer<>(CONTAINER_NSQD_IMAGE)
.withFixedExposedPort(4150, 4150)
.withFixedExposedPort(4151, 4151)
.withNetworkAliases(CONTAINER_NSQD_NAME)
.withCommand(String.format("/nsqd --broadcast-address=%s --lookupd-tcp-address=%s:4160",
- "localhost", CONTAINER_NSQLOOKUPD_NAME));
+ "localhost", CONTAINER_NSQLOOKUPD_NAME)).withNetwork(network);
}
public String getNsqConsumerUrl() {
- String format = String.format(
- "%s:%d", "localhost", 4161
- //getContainerHost(CONTAINER_NSQLOOKUPD_NAME),
- //getContainerPort(CONTAINER_NSQLOOKUPD_NAME, NsqConstants.NSQ_DEFAULT_LOOKUP_PORT_HTTP)
- );
- return format;
+ return String.format("%s:%d", "localhost", 4161);
}
public String getNsqProducerUrl() {
- String format = String.format(
- "%s:%d", "localhost", 4150
- //getContainerHost(CONTAINER_NSQD_NAME),
- //getContainerPort(CONTAINER_NSQD_NAME, NsqConstants.NSQ_DEFAULT_PORT)
- );
- return format;
+ return String.format("%s:%d", "localhost", 4150);
}
}
[camel] 10/11: Camel-NSQ: Cleanup the copied docs
Posted by ac...@apache.org.
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 e46bf2dd8308ddf1be7064e50c87041d14283be3
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Nov 8 09:34:05 2018 +0100
Camel-NSQ: Cleanup the copied docs
---
.../camel-nsq/src/main/docs/nsq-component.adoc | 40 ----------------------
1 file changed, 40 deletions(-)
diff --git a/components/camel-nsq/src/main/docs/nsq-component.adoc b/components/camel-nsq/src/main/docs/nsq-component.adoc
index d4ff4eb..85fab09 100644
--- a/components/camel-nsq/src/main/docs/nsq-component.adoc
+++ b/components/camel-nsq/src/main/docs/nsq-component.adoc
@@ -108,43 +108,3 @@ The component supports 3 options, which are listed below.
|===
// spring-boot-auto-configure options: END
-
-
-
-
-### Headers
-
-[width="100%",cols="10%,10%,80%",options="header",]
-|=======================================================================
-|Name |Type |Description
-
-|CamelNatsMessageTimestamp |long |The timestamp of a consumed message.
-|=======================================================================
-Â
-*Producer example:*
-
-[source,java]
------------------------------------------------------------
-from("direct:send").to("nats://localhost:4222?topic=test");
------------------------------------------------------------
-
-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");
------------------------------------------------------------
-
-or your token
-
-[source,java]
------------------------------------------------------------
-from("direct:send").to("nats://token@localhost:4222?topic=test");
------------------------------------------------------------
-
-*Consumer example:*
-
-[source,java]
-----------------------------------------------------------------------------------------
-from("nats://localhost:4222?topic=test&maxMessages=5&queueName=test").to("mock:result");
-----------------------------------------------------------------------------------------
[camel] 03/11: Add camel-nsq component.
Posted by ac...@apache.org.
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 9eea1616d04806fbbb48d905dadc12de33f3f7b0
Author: mionker <mi...@icloud.com>
AuthorDate: Fri Nov 2 21:58:37 2018 +0100
Add camel-nsq component.
---
.../camel-nsq/src/main/docs/nsq-component.adoc | 129 +++++++++++++++++++++
.../apache/camel/component/nsq/NsqComponent.java | 24 +++-
.../camel/component/nsq/NsqConfiguration.java | 15 ++-
.../apache/camel/component/nsq/NsqConsumer.java | 2 +-
.../apache/camel/component/nsq/NsqEndpoint.java | 13 +++
.../apache/camel/component/nsq/NsqProducer.java | 4 +-
6 files changed, 179 insertions(+), 8 deletions(-)
diff --git a/components/camel-nsq/src/main/docs/nsq-component.adoc b/components/camel-nsq/src/main/docs/nsq-component.adoc
new file mode 100644
index 0000000..39bed9f
--- /dev/null
+++ b/components/camel-nsq/src/main/docs/nsq-component.adoc
@@ -0,0 +1,129 @@
+[[nsq-component]]
+== nsq Component
+
+
+http://nsq.io/[NSQ]Â is a realtime distributed messaging platform.
+
+Maven users will need to add the following dependency to
+their `pom.xml` for this component.
+
+[source,xml]
+------------------------------------------------------------
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-nsq</artifactId>
+ <!-- use the same version as your Camel core version -->
+ <version>x.y.z</version>
+</dependency>
+------------------------------------------------------------
+
+### URI format
+
+[source,java]
+----------------------
+nsq:servers[?options]
+----------------------
+
+Where *servers* represents the list of NSQ servers - nsqlookupd servers in the case of a consumer and nsqd servers in the case of a producer.
+
+### Options
+
+
+// component options: START
+The Nsq component supports 2 options, which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *useGlobalSslContext Parameters* (security) | Enable usage of global SSL context parameters. | false | boolean
+| *resolveProperty Placeholders* (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
+|===
+// component options: END
+
+
+
+
+
+// endpoint options: START
+The NSQ endpoint is configured using URI syntax:
+
+----
+nsq:servers
+----
+
+with the following path and query parameters:
+
+==== Path Parameters (1 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *servers* | *Required* URLs to one or more nsqlookupq (consumer) or nsqd (producer) servers. Use comma to separate URLs when specifying multiple servers. The port can also be specified using <hostname>:<port>. | | String
+|===
+
+
+==== Query Parameters (25 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *topic* (common) | *Required* The name of topic we want to use | | String
+| *channel* (consumer) | *Required* The name of channel we want to use | | String
+| *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
+| *poolSize* (consumer) | Consumer pool size | 10 | int
+| *lookupServerPort* (consumer) | The default port number to use to connect to nsqlookupd server | 4161 | int
+| *port* (producer) | The default port number to use to connect to nsqd server | 4150 | int
+| *lookupInterval* (consumer) | The retry interval in millicseconds to lookup the topic at the nsqlookupq server. | 5000 | int
+| *requeueInterval* (consumer) | The requeue interval. Default of 0 will
+
+|===
+// endpoint options: END
+// spring-boot-auto-configure options: START
+=== Spring Boot Auto-Configuration
+
+// spring-boot-auto-configure options: END
+
+
+
+
+
+### Headers
+
+[width="100%",cols="10%,10%,80%",options="header",]
+|=======================================================================
+|Name |Type |Description
+
+|CamelNatsMessageTimestamp |long |The timestamp of a consumed message.
+|=======================================================================
+Â
+*Producer example:*
+
+[source,java]
+-----------------------------------------------------------
+from("direct:send").to("nats://localhost:4222?topic=test");
+-----------------------------------------------------------
+
+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");
+-----------------------------------------------------------
+
+or your token
+
+[source,java]
+-----------------------------------------------------------
+from("direct:send").to("nats://token@localhost:4222?topic=test");
+-----------------------------------------------------------
+
+*Consumer example:*
+
+[source,java]
+----------------------------------------------------------------------------------------
+from("nats://localhost:4222?topic=test&maxMessages=5&queueName=test").to("mock:result");
+----------------------------------------------------------------------------------------
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
index 6c92fd4..eda6413 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
@@ -1,17 +1,20 @@
package org.apache.camel.component.nsq;
import org.apache.camel.Endpoint;
+import org.apache.camel.SSLContextParametersAware;
import org.apache.camel.impl.DefaultComponent;
+import org.apache.camel.spi.Metadata;
import java.util.Map;
-import static org.apache.camel.util.IntrospectionSupport.setProperties;
-
/**
* Represents the component that manages {@link NsqEndpoint}.
*/
-public class NsqComponent extends DefaultComponent {
-
+public class NsqComponent extends DefaultComponent implements SSLContextParametersAware {
+
+ @Metadata(label = "security", defaultValue = "false")
+ private boolean useGlobalSslContextParameters;
+
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
NsqConfiguration config = new NsqConfiguration();
setProperties(config, parameters);
@@ -20,4 +23,17 @@ public class NsqComponent extends DefaultComponent {
NsqEndpoint endpoint = new NsqEndpoint(uri, this, config);
return endpoint;
}
+
+ @Override
+ public boolean isUseGlobalSslContextParameters() {
+ return this.useGlobalSslContextParameters;
+ }
+
+ /**
+ * Enable usage of global SSL context parameters.
+ */
+ @Override
+ public void setUseGlobalSslContextParameters(boolean useGlobalSslContextParameters) {
+ this.useGlobalSslContextParameters = useGlobalSslContextParameters;
+ }
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
index 953c7d7..2c01f99 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
@@ -6,6 +6,7 @@ import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
import org.apache.camel.spi.UriPath;
+import org.apache.camel.util.jsse.SSLContextParameters;
import java.util.Set;
@@ -33,6 +34,8 @@ public class NsqConfiguration {
private long lookupInterval = 5000;
@UriParam(label = "consumer", defaultValue = "0", description = "The requeue interval")
private long requeueInterval = 0;
+ @UriParam(label = "security")
+ private SSLContextParameters sslContextParameters;
/*
* URL a NSQ lookup server hostname.
@@ -137,7 +140,7 @@ public class NsqConfiguration {
}
/**
- * The requeue retry interval
+ * The requeue interval
*/
public long getRequeueInterval() {
return requeueInterval;
@@ -147,6 +150,16 @@ public class NsqConfiguration {
this.requeueInterval = requeueInterval;
}
+ /**
+ * To configure security using SSLContextParameters
+ */
+ public SSLContextParameters getSslContextParameters() {
+ return sslContextParameters;
+ }
+
+ public void setSslContextParameters(SSLContextParameters sslContextParameters) {
+ this.sslContextParameters = sslContextParameters;
+ }
private String splitServers() {
StringBuilder servers = new StringBuilder();
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
index f41a888..6402fe5 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
@@ -85,7 +85,7 @@ public class NsqConsumer extends DefaultConsumer {
public void message(NSQMessage msg) {
LOG.debug("Received Message: {}", msg);
Exchange exchange = getEndpoint().createExchange();
- exchange.getIn().setBody(new String(msg.getMessage()));
+ exchange.getIn().setBody(msg.getMessage());
exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ID, msg.getId());
exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ATTEMPTS, msg.getAttempts());
exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_TIMESTAMP, msg.getTimestamp());
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
index 1b25880..e792316 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
@@ -1,5 +1,8 @@
package org.apache.camel.component.nsq;
+import com.github.brainlag.nsq.NSQConfig;
+import io.netty.handler.ssl.JdkSslContext;
+import io.netty.handler.ssl.SslContext;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
@@ -7,6 +10,8 @@ import org.apache.camel.impl.DefaultEndpoint;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
import java.util.concurrent.ExecutorService;
/**
@@ -42,4 +47,12 @@ public class NsqEndpoint extends DefaultEndpoint {
public NsqConfiguration getNsqConfiguration() {
return configuration;
}
+
+ public NSQConfig getNsqConfig() throws GeneralSecurityException, IOException {
+ NSQConfig nsqConfig = new NSQConfig();
+ SslContext sslContext = new JdkSslContext(getNsqConfiguration().getSslContextParameters().createSSLContext(getCamelContext()), true, null);
+ nsqConfig.setSslContext(sslContext);
+
+ return nsqConfig;
+ }
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
index a886eae..630a2fa 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
@@ -33,8 +33,8 @@ public class NsqProducer extends DefaultProducer {
LOG.debug("Publishing to topic: {}", topic);
- String body = exchange.getIn().getBody(String.class);
- producer.produce(topic, body.getBytes());
+ byte[] body = exchange.getIn().getBody(byte[].class);
+ producer.produce(topic, body);
}
@Override
[camel] 07/11: Camel-NSQ: Fixed CS and added the starter
Posted by ac...@apache.org.
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 858a3302a13239273730440f4a17bb07a16e5bc1
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Nov 8 08:18:30 2018 +0100
Camel-NSQ: Fixed CS and added the starter
---
.../camel-nsq/src/main/docs/nsq-component.adoc | 30 +--
.../apache/camel/component/nsq/NsqComponent.java | 20 +-
.../camel/component/nsq/NsqConfiguration.java | 22 ++-
.../apache/camel/component/nsq/NsqConstants.java | 32 +++-
.../apache/camel/component/nsq/NsqConsumer.java | 78 ++++----
.../apache/camel/component/nsq/NsqEndpoint.java | 32 +++-
.../apache/camel/component/nsq/NsqProducer.java | 31 +++-
.../camel/component/nsq/NsqSynchronization.java | 17 ++
.../camel/component/nsq/NsqConsumerTest.java | 47 +++--
.../camel/component/nsq/NsqProducerTest.java | 11 +-
.../apache/camel/component/nsq/NsqTestSupport.java | 45 +++--
.../camel-nsq/src/test/resources/log4j2.properties | 28 +++
.../components-starter/camel-nsq-starter/pom.xml | 61 +++++++
.../springboot/NsqComponentAutoConfiguration.java | 128 +++++++++++++
.../nsq/springboot/NsqComponentConfiguration.java | 67 +++++++
.../src/main/resources/META-INF/LICENSE.txt | 203 +++++++++++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 ++
.../src/main/resources/META-INF/spring.factories | 19 ++
.../src/main/resources/META-INF/spring.provides | 17 ++
platforms/spring-boot/components-starter/pom.xml | 1 +
20 files changed, 789 insertions(+), 111 deletions(-)
diff --git a/components/camel-nsq/src/main/docs/nsq-component.adoc b/components/camel-nsq/src/main/docs/nsq-component.adoc
index ab48f3a..203b520 100644
--- a/components/camel-nsq/src/main/docs/nsq-component.adoc
+++ b/components/camel-nsq/src/main/docs/nsq-component.adoc
@@ -1,6 +1,8 @@
[[nsq-component]]
== nsq Component
+*Available as of Camel version 2.22*
+
http://nsq.io/[NSQ]Â is a realtime distributed messaging platform.
@@ -30,7 +32,7 @@ Where *servers* represents the list of NSQ servers - nsqlookupd servers in the
// component options: START
-The Nsq component supports 2 options, which are listed below.
+The nsq component supports 2 options, which are listed below.
@@ -47,10 +49,10 @@ The Nsq component supports 2 options, which are listed below.
// endpoint options: START
-The NSQ endpoint is configured using URI syntax:
+The nsq endpoint is configured using URI syntax:
----
-nsq:servers
+nsq:lookupServer
----
with the following path and query parameters:
@@ -61,26 +63,30 @@ 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 nsqlookupq (consumer) or nsqd (producer) servers. Use comma to separate URLs when specifying multiple servers. The port can also be specified using <hostname>:<port>. | | String
+| *servers* | *Required* The hostnames of one or more nsqlookupd servers (consumer) or nsqd servers (producer). | | String
|===
-==== Query Parameters (25 parameters):
+==== Query Parameters (16 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
| *topic* (common) | *Required* The name of topic we want to use | | String
-| *channel* (consumer) | *Required* The name of channel we want to use | | String
+| *userAgent* (common) | A String to identify the userAgent | | String
+| *autoFinish* (consumer) | Automatically finish the NSQ message when it is retrieved from the quese and before the Exchange is processed. | true | 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
+| *channel* (consumer) | The name of channel we want to use | | String
+| *lookupInterval* (consumer) | The lookup retry interval | 5000 | long
+| *lookupServerPort* (consumer) | The port of the nsqdlookupd server | 4161 | int
+| *messageTimeout* (consumer) | The NSQ message timeout for a consumer. | -1 | long
| *poolSize* (consumer) | Consumer pool size | 10 | int
-| *lookupServerPort* (consumer) | The default port number to use to connect to nsqlookupd server | 4161 | int
-| *port* (producer) | The default port number to use to connect to nsqd server | 4150 | int
-| *lookupInterval* (consumer) | The retry interval in millicseconds to lookup the topic at the nsqlookupq server. | 5000 | int
-| *autoFinish* (consumer) | Automatically finish the NSQ Message when it is retrievd from the queue and before the Exchange is processed | true | boolean
-| *messageTimeout* (consumer) | The NSQ consumer timeout period for messages retrieved from the queue | 0 | int
-| *requeueInterval* (consumer) | The requeue interval. Default of 0 will use the default configured on the server | 0 | int
+| *requeueInterval* (consumer) | The requeue interval | -1 | long
+| *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
+| *port* (producer) | The port of the nsqd server | 4150 | int
+| *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-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
index e3e9d83..37ede30 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqComponent.java
@@ -1,12 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.camel.component.nsq;
+import java.util.Map;
+
import org.apache.camel.Endpoint;
import org.apache.camel.SSLContextParametersAware;
import org.apache.camel.impl.DefaultComponent;
import org.apache.camel.spi.Metadata;
-import java.util.Map;
-
/**
* Represents the component that manages {@link NsqEndpoint}.
*/
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
index a8eb202..0204540 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConfiguration.java
@@ -1,5 +1,23 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.camel.component.nsq;
+import java.util.Set;
+
import com.github.brainlag.nsq.ServerAddress;
import com.google.common.collect.Sets;
import org.apache.camel.spi.Metadata;
@@ -8,8 +26,6 @@ import org.apache.camel.spi.UriParams;
import org.apache.camel.spi.UriPath;
import org.apache.camel.util.jsse.SSLContextParameters;
-import java.util.Set;
-
import static org.apache.camel.component.nsq.NsqConstants.NSQ_DEFAULT_LOOKUP_PORT;
import static org.apache.camel.component.nsq.NsqConstants.NSQ_DEFAULT_PORT;
@@ -38,7 +54,7 @@ public class NsqConfiguration {
private Boolean autoFinish = true;
@UriParam(label = "consumer", defaultValue = "-1", description = "The NSQ consumer timeout period for messages retrieved from the queue. A value of -1 is the server default")
private long messageTimeout = -1;
- @UriParam(description = "")
+ @UriParam(description = "A String to identify the userAgent")
private String userAgent;
@UriParam(label = "security")
private boolean secure;
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConstants.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConstants.java
index b1c1aa2..29787ff 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConstants.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConstants.java
@@ -1,12 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.camel.component.nsq;
-public class NsqConstants {
+public interface NsqConstants {
- static int NSQ_DEFAULT_PORT = 4150;
- static int NSQ_DEFAULT_LOOKUP_PORT = 4160;
- static int NSQ_DEFAULT_LOOKUP_PORT_HTTP = 4161;
- static String NSQ_MESSAGE_ID = "CamelNsqMessageId";
- static String NSQ_MESSAGE_ATTEMPTS = "CamelNsqMessageAttempts";
- static String NSQ_MESSAGE_TIMESTAMP = "CamelNsqMessageTimestamp";
- static String NSQ_MESSAGE_TOPIC = "CamelNsqMessageTopic";
+ int NSQ_DEFAULT_PORT = 4150;
+ int NSQ_DEFAULT_LOOKUP_PORT = 4160;
+ int NSQ_DEFAULT_LOOKUP_PORT_HTTP = 4161;
+ String NSQ_MESSAGE_ID = "CamelNsqMessageId";
+ String NSQ_MESSAGE_ATTEMPTS = "CamelNsqMessageAttempts";
+ String NSQ_MESSAGE_TIMESTAMP = "CamelNsqMessageTimestamp";
+ String NSQ_MESSAGE_TOPIC = "CamelNsqMessageTopic";
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
index 2beaa96..eb877ff 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqConsumer.java
@@ -1,11 +1,30 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.camel.component.nsq;
+import java.util.concurrent.ExecutorService;
+
import com.github.brainlag.nsq.NSQConsumer;
import com.github.brainlag.nsq.NSQMessage;
import com.github.brainlag.nsq.ServerAddress;
import com.github.brainlag.nsq.callbacks.NSQMessageCallback;
import com.github.brainlag.nsq.lookup.DefaultNSQLookup;
import com.github.brainlag.nsq.lookup.NSQLookup;
+
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Processor;
@@ -13,19 +32,18 @@ import org.apache.camel.impl.DefaultConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.concurrent.ExecutorService;
-
/**
* The nsq consumer.
*/
public class NsqConsumer extends DefaultConsumer {
private static final Logger LOG = LoggerFactory.getLogger(NsqConsumer.class);
+
+ NSQConsumer consumer;
private final Processor processor;
private ExecutorService executor;
private boolean active;
- NSQConsumer consumer;
private final NsqConfiguration configuration;
public NsqConsumer(NsqEndpoint endpoint, Processor processor) {
@@ -36,7 +54,7 @@ public class NsqConsumer extends DefaultConsumer {
@Override
public NsqEndpoint getEndpoint() {
- return (NsqEndpoint) super.getEndpoint();
+ return (NsqEndpoint)super.getEndpoint();
}
@Override
@@ -48,13 +66,11 @@ public class NsqConsumer extends DefaultConsumer {
LOG.debug("Getting NSQ Connection");
NSQLookup lookup = new DefaultNSQLookup();
- for(ServerAddress server : configuration.getServerAddresses()) {
- lookup.addLookupAddress(server.getHost(),
- server.getPort() == 0 ? configuration.getLookupServerPort() : server.getPort());
+ for (ServerAddress server : configuration.getServerAddresses()) {
+ lookup.addLookupAddress(server.getHost(), server.getPort() == 0 ? configuration.getLookupServerPort() : server.getPort());
}
- consumer = new NSQConsumer(lookup, configuration.getTopic(),
- configuration.getChannel(), new CamelNsqMessageHandler(), getEndpoint().getNsqConfig());
+ consumer = new NSQConsumer(lookup, configuration.getTopic(), configuration.getChannel(), new CamelNsqMessageHandler(), getEndpoint().getNsqConfig());
consumer.setLookupPeriod(configuration.getLookupInterval());
consumer.setExecutor(getEndpoint().createExecutor());
consumer.start();
@@ -80,29 +96,29 @@ public class NsqConsumer extends DefaultConsumer {
super.doStop();
}
- class CamelNsqMessageHandler implements NSQMessageCallback {
-
- @Override
- public void message(NSQMessage msg) {
- LOG.debug("Received Message: {}", msg);
- Exchange exchange = getEndpoint().createExchange(ExchangePattern.InOnly);
- exchange.getIn().setBody(msg.getMessage());
- exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ID, msg.getId());
- exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ATTEMPTS, msg.getAttempts());
- exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_TIMESTAMP, msg.getTimestamp());
- try {
- if (configuration.getAutoFinish()) {
- msg.finished();
- } else {
- exchange.addOnCompletion(new NsqSynchronization(msg, (int) configuration.getRequeueInterval()));
- }
- processor.process(exchange);
- } catch (Exception e) {
- if (!configuration.getAutoFinish()) {
- msg.requeue((int) configuration.getRequeueInterval());
- }
- getExceptionHandler().handleException("Error during processing", exchange, e);
+ class CamelNsqMessageHandler implements NSQMessageCallback {
+
+ @Override
+ public void message(NSQMessage msg) {
+ LOG.debug("Received Message: {}", msg);
+ Exchange exchange = getEndpoint().createExchange(ExchangePattern.InOnly);
+ exchange.getIn().setBody(msg.getMessage());
+ exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ID, msg.getId());
+ exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_ATTEMPTS, msg.getAttempts());
+ exchange.getIn().setHeader(NsqConstants.NSQ_MESSAGE_TIMESTAMP, msg.getTimestamp());
+ try {
+ if (configuration.getAutoFinish()) {
+ msg.finished();
+ } else {
+ exchange.addOnCompletion(new NsqSynchronization(msg, (int)configuration.getRequeueInterval()));
}
+ processor.process(exchange);
+ } catch (Exception e) {
+ if (!configuration.getAutoFinish()) {
+ msg.requeue((int)configuration.getRequeueInterval());
+ }
+ getExceptionHandler().handleException("Error during processing", exchange, e);
}
}
+ }
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
index 7787b00..63cf654 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqEndpoint.java
@@ -1,8 +1,30 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.camel.component.nsq;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.concurrent.ExecutorService;
+
import com.github.brainlag.nsq.NSQConfig;
+
import io.netty.handler.ssl.JdkSslContext;
import io.netty.handler.ssl.SslContext;
+
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
@@ -12,15 +34,10 @@ import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.util.ObjectHelper;
-import java.io.IOException;
-import java.security.GeneralSecurityException;
-import java.util.concurrent.ExecutorService;
-
/**
* Represents a nsq endpoint.
*/
-@UriEndpoint(firstVersion = "2.22.0", scheme = "nsq", title = "nsq", syntax="nsq:lookupServer",
- consumerClass = NsqConsumer.class, label = "messaging")
+@UriEndpoint(firstVersion = "2.22.0", scheme = "nsq", title = "nsq", syntax = "nsq:lookupServer", consumerClass = NsqConsumer.class, label = "messaging")
public class NsqEndpoint extends DefaultEndpoint {
@UriParam
@@ -30,6 +47,7 @@ public class NsqEndpoint extends DefaultEndpoint {
super(uri, component);
this.configuration = configuration;
}
+
public Producer createProducer() throws Exception {
return new NsqProducer(this);
}
@@ -66,7 +84,7 @@ public class NsqEndpoint extends DefaultEndpoint {
}
if (configuration.getMessageTimeout() > -1) {
- nsqConfig.setMsgTimeout((int) configuration.getMessageTimeout());
+ nsqConfig.setMsgTimeout((int)configuration.getMessageTimeout());
}
return nsqConfig;
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
index 03a34fc..ec0f760 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqProducer.java
@@ -1,7 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.camel.component.nsq;
import com.github.brainlag.nsq.NSQProducer;
import com.github.brainlag.nsq.ServerAddress;
+
import org.apache.camel.Exchange;
import org.apache.camel.impl.DefaultProducer;
import org.slf4j.Logger;
@@ -24,14 +41,13 @@ public class NsqProducer extends DefaultProducer {
@Override
public NsqEndpoint getEndpoint() {
- return (NsqEndpoint) super.getEndpoint();
+ return (NsqEndpoint)super.getEndpoint();
}
@Override
public void process(Exchange exchange) throws Exception {
- String topic = exchange.getIn().getHeader(NsqConstants.NSQ_MESSAGE_TOPIC,
- configuration.getTopic(), String.class);
+ String topic = exchange.getIn().getHeader(NsqConstants.NSQ_MESSAGE_TOPIC, configuration.getTopic(), String.class);
LOG.debug("Publishing to topic: {}", topic);
@@ -46,9 +62,8 @@ public class NsqProducer extends DefaultProducer {
NsqConfiguration config = getEndpoint().getNsqConfiguration();
producer = new NSQProducer();
- for(ServerAddress server : config.getServerAddresses()) {
- producer.addAddress(server.getHost(),
- server.getPort() == 0 ? config.getPort() : server.getPort());
+ for (ServerAddress server : config.getServerAddresses()) {
+ producer.addAddress(server.getHost(), server.getPort() == 0 ? config.getPort() : server.getPort());
}
producer.setConfig(getEndpoint().getNsqConfig());
producer.start();
@@ -57,7 +72,9 @@ public class NsqProducer extends DefaultProducer {
@Override
protected void doStop() throws Exception {
LOG.debug("Stopping NSQ Producer");
- if (producer != null) { producer.shutdown(); }
+ if (producer != null) {
+ producer.shutdown();
+ }
super.doStop();
}
}
diff --git a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java
index 010e227..30d7aae 100644
--- a/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java
+++ b/components/camel-nsq/src/main/java/org/apache/camel/component/nsq/NsqSynchronization.java
@@ -1,6 +1,23 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.camel.component.nsq;
import com.github.brainlag.nsq.NSQMessage;
+
import org.apache.camel.Exchange;
import org.apache.camel.support.SynchronizationAdapter;
diff --git a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java
index 7c15fdd..23b284f 100644
--- a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java
+++ b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqConsumerTest.java
@@ -1,7 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.camel.component.nsq;
+import java.util.concurrent.TimeoutException;
+
import com.github.brainlag.nsq.NSQProducer;
import com.github.brainlag.nsq.exceptions.NSQException;
+
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -9,8 +28,6 @@ import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.Test;
-import java.util.concurrent.TimeoutException;
-
public class NsqConsumerTest extends NsqTestSupport {
private static final int NUMBER_OF_MESSAGES = 10000;
@@ -28,7 +45,7 @@ public class NsqConsumerTest extends NsqTestSupport {
producer.addAddress("localhost", 4150);
producer.start();
- producer.produce(TOPIC, ("Hello NSQ!").getBytes());
+ producer.produce(TOPIC, "Hello NSQ!".getBytes());
mockResultEndpoint.assertIsSatisfied();
@@ -60,7 +77,7 @@ public class NsqConsumerTest extends NsqTestSupport {
producer.addAddress("localhost", 4150);
producer.start();
- producer.produce(TOPIC, ("Test Requeue").getBytes());
+ producer.produce(TOPIC, "Test Requeue".getBytes());
mockResultEndpoint.assertIsSatisfied();
}
@@ -70,18 +87,16 @@ public class NsqConsumerTest extends NsqTestSupport {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- fromF("nsq://%s?topic=%s&lookupInterval=2s&autoFinish=false&requeueInterval=1s", getNsqConsumerUrl(), TOPIC)
- .process(new Processor() {
- @Override
- public void process(Exchange exchange) throws Exception {
- String messageText = exchange.getIn().getBody(String.class);
- int attempts = exchange.getIn().getHeader(NsqConstants.NSQ_MESSAGE_ATTEMPTS, Integer.class);
- if (messageText.contains("Requeue") && attempts < 3) {
- throw new Exception();
- }
- }
- })
- .to(mockResultEndpoint);
+ fromF("nsq://%s?topic=%s&lookupInterval=2s&autoFinish=false&requeueInterval=1s", getNsqConsumerUrl(), TOPIC).process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ String messageText = exchange.getIn().getBody(String.class);
+ int attempts = exchange.getIn().getHeader(NsqConstants.NSQ_MESSAGE_ATTEMPTS, Integer.class);
+ if (messageText.contains("Requeue") && attempts < 3) {
+ throw new Exception();
+ }
+ }
+ }).to(mockResultEndpoint);
}
};
}
diff --git a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java
index 32175fe..bdd7fb2 100644
--- a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java
+++ b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqProducerTest.java
@@ -16,16 +16,17 @@
*/
package org.apache.camel.component.nsq;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
import com.github.brainlag.nsq.NSQConsumer;
import com.github.brainlag.nsq.lookup.DefaultNSQLookup;
import com.github.brainlag.nsq.lookup.NSQLookup;
+
import org.apache.camel.builder.RouteBuilder;
import org.junit.Test;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
public class NsqProducerTest extends NsqTestSupport {
private static final int NUMBER_OF_MESSAGES = 10000;
@@ -62,7 +63,7 @@ public class NsqProducerTest extends NsqTestSupport {
CountDownLatch lock = new CountDownLatch(NUMBER_OF_MESSAGES);
for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
- template.sendBody("direct:send", ("test" + i));
+ template.sendBody("direct:send", "test" + i);
}
AtomicInteger counter = new AtomicInteger(0);
diff --git a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java
index 2219397..b13875a 100644
--- a/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java
+++ b/components/camel-nsq/src/test/java/org/apache/camel/component/nsq/NsqTestSupport.java
@@ -1,14 +1,30 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
package org.apache.camel.component.nsq;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
import org.apache.camel.test.testcontainers.ContainerAwareTestSupport;
import org.testcontainers.containers.FixedHostPortGenericContainer;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
public class NsqTestSupport extends ContainerAwareTestSupport {
public static final String CONTAINER_NSQLOOKUPD_IMAGE = "nsqio/nsq";
@@ -22,28 +38,17 @@ public class NsqTestSupport extends ContainerAwareTestSupport {
@Override
protected List<GenericContainer<?>> createContainers() {
network = Network.newNetwork();
- return new ArrayList<GenericContainer<?>>(
- Arrays.asList(
- nsqlookupdContainer(network),
- nsqdContainer(network)
- ));
+ return new ArrayList<GenericContainer<?>>(Arrays.asList(nsqlookupdContainer(network), nsqdContainer(network)));
}
public static GenericContainer<?> nsqlookupdContainer(Network network) {
- return new FixedHostPortGenericContainer<>(CONTAINER_NSQLOOKUPD_IMAGE)
- .withFixedExposedPort(4160, 4160)
- .withFixedExposedPort(4161, 4161)
- .withNetworkAliases(CONTAINER_NSQLOOKUPD_NAME)
- .withCommand("/nsqlookupd").withNetwork(network);
+ return new FixedHostPortGenericContainer<>(CONTAINER_NSQLOOKUPD_IMAGE).withFixedExposedPort(4160, 4160).withFixedExposedPort(4161, 4161)
+ .withNetworkAliases(CONTAINER_NSQLOOKUPD_NAME).withCommand("/nsqlookupd").withNetwork(network);
}
public static GenericContainer<?> nsqdContainer(Network network) {
- return new FixedHostPortGenericContainer<>(CONTAINER_NSQD_IMAGE)
- .withFixedExposedPort(4150, 4150)
- .withFixedExposedPort(4151, 4151)
- .withNetworkAliases(CONTAINER_NSQD_NAME)
- .withCommand(String.format("/nsqd --broadcast-address=%s --lookupd-tcp-address=%s:4160",
- "localhost", CONTAINER_NSQLOOKUPD_NAME)).withNetwork(network);
+ return new FixedHostPortGenericContainer<>(CONTAINER_NSQD_IMAGE).withFixedExposedPort(4150, 4150).withFixedExposedPort(4151, 4151).withNetworkAliases(CONTAINER_NSQD_NAME)
+ .withCommand(String.format("/nsqd --broadcast-address=%s --lookupd-tcp-address=%s:4160", "localhost", CONTAINER_NSQLOOKUPD_NAME)).withNetwork(network);
}
public String getNsqConsumerUrl() {
diff --git a/components/camel-nsq/src/test/resources/log4j2.properties b/components/camel-nsq/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..e4def0c
--- /dev/null
+++ b/components/camel-nsq/src/test/resources/log4j2.properties
@@ -0,0 +1,28 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-nsq-test.log
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+appender.out.type = Console
+appender.out.name = out
+appender.out.layout.type = PatternLayout
+appender.out.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+rootLogger.level = INFO
+rootLogger.appenderRef.file.ref = file
diff --git a/platforms/spring-boot/components-starter/camel-nsq-starter/pom.xml b/platforms/spring-boot/components-starter/camel-nsq-starter/pom.xml
new file mode 100644
index 0000000..cfabb89
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-nsq-starter/pom.xml
@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components-starter</artifactId>
+ <version>2.23.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>camel-nsq-starter</artifactId>
+ <packaging>jar</packaging>
+ <name>Spring-Boot Starter :: Camel :: NSQ</name>
+ <description>Spring-Boot Starter for Camel Components</description>
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter</artifactId>
+ <version>${spring-boot-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-nsq</artifactId>
+ <version>${project.version}</version>
+ <!--START OF GENERATED CODE-->
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ </exclusion>
+ </exclusions>
+ <!--END OF GENERATED CODE-->
+ </dependency>
+ <!--START OF GENERATED CODE-->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core-starter</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-spring-boot-starter</artifactId>
+ </dependency>
+ <!--END OF GENERATED CODE-->
+ </dependencies>
+</project>
diff --git a/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/java/org/apache/camel/component/nsq/springboot/NsqComponentAutoConfiguration.java b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/java/org/apache/camel/component/nsq/springboot/NsqComponentAutoConfiguration.java
new file mode 100644
index 0000000..6b02ac0
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/java/org/apache/camel/component/nsq/springboot/NsqComponentAutoConfiguration.java
@@ -0,0 +1,128 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.nsq.springboot;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Generated;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.nsq.NsqComponent;
+import org.apache.camel.spi.ComponentCustomizer;
+import org.apache.camel.spi.HasId;
+import org.apache.camel.spring.boot.CamelAutoConfiguration;
+import org.apache.camel.spring.boot.ComponentConfigurationProperties;
+import org.apache.camel.spring.boot.util.CamelPropertiesHelper;
+import org.apache.camel.spring.boot.util.ConditionalOnCamelContextAndAutoConfigurationBeans;
+import org.apache.camel.spring.boot.util.GroupCondition;
+import org.apache.camel.spring.boot.util.HierarchicalPropertiesEvaluator;
+import org.apache.camel.util.IntrospectionSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+
+/**
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.SpringBootAutoConfigurationMojo")
+@Configuration
+@Conditional({ConditionalOnCamelContextAndAutoConfigurationBeans.class,
+ NsqComponentAutoConfiguration.GroupConditions.class})
+@AutoConfigureAfter(CamelAutoConfiguration.class)
+@EnableConfigurationProperties({ComponentConfigurationProperties.class,
+ NsqComponentConfiguration.class})
+public class NsqComponentAutoConfiguration {
+
+ private static final Logger LOGGER = LoggerFactory
+ .getLogger(NsqComponentAutoConfiguration.class);
+ @Autowired
+ private ApplicationContext applicationContext;
+ @Autowired
+ private CamelContext camelContext;
+ @Autowired
+ private NsqComponentConfiguration configuration;
+ @Autowired(required = false)
+ private List<ComponentCustomizer<NsqComponent>> customizers;
+
+ static class GroupConditions extends GroupCondition {
+ public GroupConditions() {
+ super("camel.component", "camel.component.nsq");
+ }
+ }
+
+ @Lazy
+ @Bean(name = "nsq-component")
+ @ConditionalOnMissingBean(NsqComponent.class)
+ public NsqComponent configureNsqComponent() throws Exception {
+ NsqComponent component = new NsqComponent();
+ component.setCamelContext(camelContext);
+ Map<String, Object> parameters = new HashMap<>();
+ IntrospectionSupport.getProperties(configuration, parameters, null,
+ false);
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ Object value = entry.getValue();
+ Class<?> paramClass = value.getClass();
+ if (paramClass.getName().endsWith("NestedConfiguration")) {
+ Class nestedClass = null;
+ try {
+ nestedClass = (Class) paramClass.getDeclaredField(
+ "CAMEL_NESTED_CLASS").get(null);
+ HashMap<String, Object> nestedParameters = new HashMap<>();
+ IntrospectionSupport.getProperties(value, nestedParameters,
+ null, false);
+ Object nestedProperty = nestedClass.newInstance();
+ CamelPropertiesHelper.setCamelProperties(camelContext,
+ nestedProperty, nestedParameters, false);
+ entry.setValue(nestedProperty);
+ } catch (NoSuchFieldException e) {
+ }
+ }
+ }
+ CamelPropertiesHelper.setCamelProperties(camelContext, component,
+ parameters, false);
+ if (ObjectHelper.isNotEmpty(customizers)) {
+ for (ComponentCustomizer<NsqComponent> customizer : customizers) {
+ boolean useCustomizer = (customizer instanceof HasId)
+ ? HierarchicalPropertiesEvaluator.evaluate(
+ applicationContext.getEnvironment(),
+ "camel.component.customizer",
+ "camel.component.nsq.customizer",
+ ((HasId) customizer).getId())
+ : HierarchicalPropertiesEvaluator.evaluate(
+ applicationContext.getEnvironment(),
+ "camel.component.customizer",
+ "camel.component.nsq.customizer");
+ if (useCustomizer) {
+ LOGGER.debug("Configure component {}, with customizer {}",
+ component, customizer);
+ customizer.customize(component);
+ }
+ }
+ }
+ return component;
+ }
+}
\ No newline at end of file
diff --git a/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/java/org/apache/camel/component/nsq/springboot/NsqComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/java/org/apache/camel/component/nsq/springboot/NsqComponentConfiguration.java
new file mode 100644
index 0000000..3b7a3f6
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/java/org/apache/camel/component/nsq/springboot/NsqComponentConfiguration.java
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.nsq.springboot;
+
+import javax.annotation.Generated;
+import org.apache.camel.spring.boot.ComponentConfigurationPropertiesCommon;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Represents a nsq endpoint.
+ *
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.SpringBootAutoConfigurationMojo")
+@ConfigurationProperties(prefix = "camel.component.nsq")
+public class NsqComponentConfiguration
+ extends
+ ComponentConfigurationPropertiesCommon {
+
+ /**
+ * Whether to enable auto configuration of the nsq component. This is
+ * enabled by default.
+ */
+ private Boolean enabled;
+ /**
+ * Enable usage of global SSL context parameters.
+ */
+ private Boolean useGlobalSslContextParameters = false;
+ /**
+ * Whether the component should resolve property placeholders on itself when
+ * starting. Only properties which are of String type can use property
+ * placeholders.
+ */
+ private Boolean resolvePropertyPlaceholders = true;
+
+ public Boolean getUseGlobalSslContextParameters() {
+ return useGlobalSslContextParameters;
+ }
+
+ public void setUseGlobalSslContextParameters(
+ Boolean useGlobalSslContextParameters) {
+ this.useGlobalSslContextParameters = useGlobalSslContextParameters;
+ }
+
+ public Boolean getResolvePropertyPlaceholders() {
+ return resolvePropertyPlaceholders;
+ }
+
+ public void setResolvePropertyPlaceholders(
+ Boolean resolvePropertyPlaceholders) {
+ this.resolvePropertyPlaceholders = resolvePropertyPlaceholders;
+ }
+}
\ No newline at end of file
diff --git a/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/LICENSE.txt b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
diff --git a/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/NOTICE.txt b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
diff --git a/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/spring.factories b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/spring.factories
new file mode 100644
index 0000000..f95faae
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,19 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.apache.camel.component.nsq.springboot.NsqComponentAutoConfiguration
diff --git a/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/spring.provides b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/spring.provides
new file mode 100644
index 0000000..bc145e9
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-nsq-starter/src/main/resources/META-INF/spring.provides
@@ -0,0 +1,17 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+provides: camel-nsq
diff --git a/platforms/spring-boot/components-starter/pom.xml b/platforms/spring-boot/components-starter/pom.xml
index 07d9880..8208920 100644
--- a/platforms/spring-boot/components-starter/pom.xml
+++ b/platforms/spring-boot/components-starter/pom.xml
@@ -269,6 +269,7 @@
<module>camel-netty-starter</module>
<module>camel-netty4-http-starter</module>
<module>camel-netty4-starter</module>
+ <module>camel-nsq-starter</module>
<module>camel-ognl-starter</module>
<module>camel-olingo2-starter</module>
<module>camel-olingo4-starter</module>
[camel] 08/11: Camel-NSQ: Added the component to kit
Posted by ac...@apache.org.
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 44aa18e3ff40819cb8d874cf0ba0c81bd80fb811
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Nov 8 08:23:08 2018 +0100
Camel-NSQ: Added the component to kit
---
apache-camel/src/main/descriptors/common-bin.xml | 2 ++
bom/camel-bom/pom.xml | 5 +++++
parent/pom.xml | 10 ++++++++++
3 files changed, 17 insertions(+)
diff --git a/apache-camel/src/main/descriptors/common-bin.xml b/apache-camel/src/main/descriptors/common-bin.xml
index b376768..8c63bdb 100644
--- a/apache-camel/src/main/descriptors/common-bin.xml
+++ b/apache-camel/src/main/descriptors/common-bin.xml
@@ -200,6 +200,7 @@
<include>org.apache.camel:camel-netty4</include>
<include>org.apache.camel:camel-netty-http</include>
<include>org.apache.camel:camel-netty4-http</include>
+ <include>org.apache.camel:camel-nsq</include>
<include>org.apache.camel:camel-ognl</include>
<include>org.apache.camel:camel-olingo2-api</include>
<include>org.apache.camel:camel-olingo2</include>
@@ -521,6 +522,7 @@
<include>org.apache.camel:camel-netty4-starter</include>
<include>org.apache.camel:camel-netty-http-starter</include>
<include>org.apache.camel:camel-netty-starter</include>
+ <include>org.apache.camel:camel-nsq-starter</include>
<include>org.apache.camel:camel-ognl-starter</include>
<include>org.apache.camel:camel-olingo2-starter</include>
<include>org.apache.camel:camel-olingo4-starter</include>
diff --git a/bom/camel-bom/pom.xml b/bom/camel-bom/pom.xml
index 655f9d9..69a6f75 100644
--- a/bom/camel-bom/pom.xml
+++ b/bom/camel-bom/pom.xml
@@ -1860,6 +1860,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-nsq-starter</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-ognl</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/parent/pom.xml b/parent/pom.xml
index 334bf70..12bd4ff 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -1728,6 +1728,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-nsq</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-ognl</artifactId>
<version>${project.version}</version>
</dependency>
@@ -3232,6 +3237,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-nsq-starter</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-ognl-starter</artifactId>
<version>${project.version}</version>
</dependency>
[camel] 01/11: Up to latest version
Posted by ac...@apache.org.
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 e713a08f4fc54864e68a54356552047859e7e781
Author: elbaz michael <mi...@outlook.fr>
AuthorDate: Tue Nov 6 23:17:49 2018 +0100
Up to latest version
I don't know if is relevant but i'm using the latest camel version (2.22.1) and the method `component.createSqlSessionFactory()` is now protected so you should use `component.getSqlSessionFactory()` instead (i don't running the test)
---
.../java/org/apache/camel/component/mybatis/MyBatisTestSupport.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisTestSupport.java b/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisTestSupport.java
index b9caf9b..e84a448 100644
--- a/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisTestSupport.java
+++ b/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisTestSupport.java
@@ -79,7 +79,7 @@ public abstract class MyBatisTestSupport extends CamelTestSupport {
protected Connection createConnection() throws Exception {
MyBatisComponent component = context.getComponent("mybatis", MyBatisComponent.class);
- return component.createSqlSessionFactory().getConfiguration().getEnvironment().getDataSource().getConnection();
+ return component.getSqlSessionFactory().getConfiguration().getEnvironment().getDataSource().getConnection();
}
}