You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ni...@apache.org on 2015/01/29 14:50:15 UTC
[1/2] camel git commit: CAMEL-8172 Enable multiple consumers for
kafka endpoint with thanks to Ivan
Repository: camel
Updated Branches:
refs/heads/master 36895c7b8 -> d4780c755
CAMEL-8172 Enable multiple consumers for kafka endpoint with thanks to Ivan
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/b0fb0c35
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/b0fb0c35
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/b0fb0c35
Branch: refs/heads/master
Commit: b0fb0c3545c1cb033b8d0b1499f797cea938b4c4
Parents: 36895c7
Author: Willem Jiang <wi...@gmail.com>
Authored: Thu Jan 29 20:45:36 2015 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Thu Jan 29 20:45:36 2015 +0800
----------------------------------------------------------------------
.../java/org/apache/camel/component/kafka/KafkaEndpoint.java | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/b0fb0c35/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
index e0ee9b0..aa33d4b 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
@@ -23,6 +23,7 @@ import kafka.message.MessageAndMetadata;
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
+import org.apache.camel.MultipleConsumersSupport;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.impl.DefaultEndpoint;
@@ -33,7 +34,7 @@ import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
@UriEndpoint(scheme = "kafka", consumerClass = KafkaConsumer.class, label = "messaging")
-public class KafkaEndpoint extends DefaultEndpoint {
+public class KafkaEndpoint extends DefaultEndpoint implements MultipleConsumersSupport {
@UriPath
private String brokers;
@@ -453,4 +454,8 @@ public class KafkaEndpoint extends DefaultEndpoint {
return configuration.getRequestTimeoutMs();
}
+ @Override
+ public boolean isMultipleConsumersSupported() {
+ return true;
+ }
}
[2/2] camel git commit: CAMEL-3110 Applied the patch of support
GETNEXT and SNMPV3 with thanks to ivxivx
Posted by ni...@apache.org.
CAMEL-3110 Applied the patch of support GETNEXT and SNMPV3 with thanks to ivxivx
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/d4780c75
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/d4780c75
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/d4780c75
Branch: refs/heads/master
Commit: d4780c755aecd514e460694ee7dc7c51663279a9
Parents: b0fb0c3
Author: Willem Jiang <wi...@gmail.com>
Authored: Thu Jan 29 21:47:18 2015 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Thu Jan 29 21:47:18 2015 +0800
----------------------------------------------------------------------
.../camel/component/snmp/SnmpActionType.java | 2 +-
.../component/snmp/SnmpAuthProtocolType.java | 21 +++
.../camel/component/snmp/SnmpEndpoint.java | 81 +++++++++++
.../camel/component/snmp/SnmpOIDPoller.java | 136 ++++++++++++++++---
.../component/snmp/SnmpPrivacyProtocolType.java | 21 +++
5 files changed, 243 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/d4780c75/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpActionType.java
----------------------------------------------------------------------
diff --git a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpActionType.java b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpActionType.java
index ff8e7c1..9ca516c 100644
--- a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpActionType.java
+++ b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpActionType.java
@@ -17,5 +17,5 @@
package org.apache.camel.component.snmp;
public enum SnmpActionType {
- TRAP, POLL
+ TRAP, POLL, GET_NEXT
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d4780c75/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpAuthProtocolType.java
----------------------------------------------------------------------
diff --git a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpAuthProtocolType.java b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpAuthProtocolType.java
new file mode 100644
index 0000000..d2a5dbb
--- /dev/null
+++ b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpAuthProtocolType.java
@@ -0,0 +1,21 @@
+/**
+ * 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.snmp;
+
+public enum SnmpAuthProtocolType {
+ MD5, SHA1
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/d4780c75/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpEndpoint.java b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpEndpoint.java
index 2315156..f1d1812 100644
--- a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpEndpoint.java
+++ b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpEndpoint.java
@@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory;
import org.snmp4j.CommandResponderEvent;
import org.snmp4j.PDU;
import org.snmp4j.mp.SnmpConstants;
+import org.snmp4j.security.SecurityLevel;
@UriEndpoint(scheme = "snmp", label = "monitoring")
public class SnmpEndpoint extends DefaultPollingEndpoint {
@@ -67,6 +68,22 @@ public class SnmpEndpoint extends DefaultPollingEndpoint {
@UriParam(defaultValue = "60")
private int delay = 60;
+ @UriParam(defaultValue = "" + SecurityLevel.AUTH_PRIV)
+ private int securityLevel = SecurityLevel.AUTH_PRIV;
+ @UriParam
+ private String securityName;
+ @UriParam
+ private String authenticationProtocol;
+ @UriParam
+ private String authenticationPassphrase;
+ @UriParam
+ private String privacyProtocol;
+ @UriParam
+ private String privacyPassphrase;
+ @UriParam
+ private String snmpContextName;
+ @UriParam
+ private String snmpContextEngineId;
/**
* creates a snmp endpoint
*
@@ -260,6 +277,70 @@ public class SnmpEndpoint extends DefaultPollingEndpoint {
setAddress(address);
}
+ public int getSecurityLevel() {
+ return securityLevel;
+ }
+
+ public void setSecurityLevel(int securityLevel) {
+ this.securityLevel = securityLevel;
+ }
+
+ public String getSecurityName() {
+ return securityName;
+ }
+
+ public void setSecurityName(String securityName) {
+ this.securityName = securityName;
+ }
+
+ public String getAuthenticationProtocol() {
+ return authenticationProtocol;
+ }
+
+ public void setAuthenticationProtocol(String authenticationProtocol) {
+ this.authenticationProtocol = authenticationProtocol;
+ }
+
+ public String getAuthenticationPassphrase() {
+ return authenticationPassphrase;
+ }
+
+ public void setAuthenticationPassphrase(String authenticationPassphrase) {
+ this.authenticationPassphrase = authenticationPassphrase;
+ }
+
+ public String getPrivacyProtocol() {
+ return privacyProtocol;
+ }
+
+ public void setPrivacyProtocol(String privacyProtocol) {
+ this.privacyProtocol = privacyProtocol;
+ }
+
+ public String getPrivacyPassphrase() {
+ return privacyPassphrase;
+ }
+
+ public void setPrivacyPassphrase(String privacyPassphrase) {
+ this.privacyPassphrase = privacyPassphrase;
+ }
+
+ public String getSnmpContextName() {
+ return snmpContextName;
+ }
+
+ public void setSnmpContextName(String snmpContextName) {
+ this.snmpContextName = snmpContextName;
+ }
+
+ public String getSnmpContextEngineId() {
+ return snmpContextEngineId;
+ }
+
+ public void setSnmpContextEngineId(String snmpContextEngineId) {
+ this.snmpContextEngineId = snmpContextEngineId;
+ }
+
@Override
public String toString() {
// only show address to avoid user and password details to be shown
http://git-wip-us.apache.org/repos/asf/camel/blob/d4780c75/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpOIDPoller.java
----------------------------------------------------------------------
diff --git a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpOIDPoller.java b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpOIDPoller.java
index 9c27742..3ee36a0 100644
--- a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpOIDPoller.java
+++ b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpOIDPoller.java
@@ -23,14 +23,26 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.snmp4j.CommunityTarget;
import org.snmp4j.PDU;
+import org.snmp4j.ScopedPDU;
import org.snmp4j.Snmp;
+import org.snmp4j.Target;
import org.snmp4j.TransportMapping;
+import org.snmp4j.UserTarget;
import org.snmp4j.event.ResponseEvent;
import org.snmp4j.event.ResponseListener;
import org.snmp4j.mp.MPv3;
+import org.snmp4j.mp.SnmpConstants;
+import org.snmp4j.security.AuthMD5;
+import org.snmp4j.security.AuthSHA;
+import org.snmp4j.security.Priv3DES;
+import org.snmp4j.security.PrivAES128;
+import org.snmp4j.security.PrivAES192;
+import org.snmp4j.security.PrivAES256;
+import org.snmp4j.security.PrivDES;
import org.snmp4j.security.SecurityModels;
import org.snmp4j.security.SecurityProtocols;
import org.snmp4j.security.USM;
+import org.snmp4j.security.UsmUser;
import org.snmp4j.smi.Address;
import org.snmp4j.smi.GenericAddress;
import org.snmp4j.smi.OID;
@@ -46,8 +58,8 @@ public class SnmpOIDPoller extends ScheduledPollConsumer implements ResponseList
private Address targetAddress;
private TransportMapping<? extends Address> transport;
private Snmp snmp;
- private USM usm;
- private CommunityTarget target;
+
+ private Target target;
private PDU pdu;
private SnmpEndpoint endpoint;
@@ -74,19 +86,58 @@ public class SnmpOIDPoller extends ScheduledPollConsumer implements ResponseList
}
this.snmp = new Snmp(this.transport);
- this.usm = new USM(SecurityProtocols.getInstance(), new OctetString(MPv3.createLocalEngineID()), 0);
- SecurityModels.getInstance().addSecurityModel(usm);
- // setting up target
- target = new CommunityTarget();
- target.setCommunity(new OctetString(this.endpoint.getSnmpCommunity()));
- target.setAddress(targetAddress);
- target.setRetries(this.endpoint.getRetries());
- target.setTimeout(this.endpoint.getTimeout());
- target.setVersion(this.endpoint.getSnmpVersion());
-
- // creating PDU
- this.pdu = new PDU();
+ if (SnmpConstants.version3 == endpoint.getSnmpVersion()) {
+ UserTarget userTarget = new UserTarget();
+
+ userTarget.setSecurityLevel(endpoint.getSecurityLevel());
+ userTarget.setSecurityName(convertToOctetString(endpoint.getSecurityName()));
+ userTarget.setAddress(targetAddress);
+ userTarget.setRetries(endpoint.getRetries());
+ userTarget.setTimeout(endpoint.getTimeout());
+ userTarget.setVersion(endpoint.getSnmpVersion());
+
+ this.target = userTarget;
+
+ USM usm = new USM(SecurityProtocols.getInstance(), new OctetString(MPv3.createLocalEngineID()), 0);
+ SecurityModels.getInstance().addSecurityModel(usm);
+
+ OID authProtocol = convertAuthenticationProtocol(endpoint.getAuthenticationProtocol());
+
+ OctetString authPwd = convertToOctetString(endpoint.getAuthenticationPassphrase());
+
+ OID privProtocol = convertPrivacyProtocol(endpoint.getPrivacyProtocol());
+
+ OctetString privPwd = convertToOctetString(endpoint.getPrivacyPassphrase());
+
+ UsmUser user = new UsmUser(convertToOctetString(endpoint.getSecurityName()), authProtocol, authPwd, privProtocol, privPwd);
+
+ usm.addUser(convertToOctetString(endpoint.getSecurityName()), user);
+
+ ScopedPDU scopedPDU = new ScopedPDU();
+
+ if (endpoint.getSnmpContextEngineId() != null) {
+ scopedPDU.setContextEngineID(new OctetString(endpoint.getSnmpContextEngineId()));
+ }
+
+ if (endpoint.getSnmpContextName() != null) {
+ scopedPDU.setContextName(new OctetString(endpoint.getSnmpContextName()));
+ }
+
+ this.pdu = scopedPDU;
+ } else {
+ CommunityTarget communityTarget = new CommunityTarget();
+
+ communityTarget.setCommunity(convertToOctetString(endpoint.getSnmpCommunity()));
+ communityTarget.setAddress(targetAddress);
+ communityTarget.setRetries(endpoint.getRetries());
+ communityTarget.setTimeout(endpoint.getTimeout());
+ communityTarget.setVersion(endpoint.getSnmpVersion());
+
+ this.target = communityTarget;
+
+ this.pdu = new PDU();
+ }
// listen to the transport
if (LOG.isDebugEnabled()) {
@@ -113,7 +164,10 @@ public class SnmpOIDPoller extends ScheduledPollConsumer implements ResponseList
@Override
protected int poll() throws Exception {
this.pdu.clear();
- this.pdu.setType(PDU.GET);
+
+ int type = this.getPduType(this.endpoint.getType());
+
+ this.pdu.setType(type);
// prepare the request items
for (OID oid : this.endpoint.getOids()) {
@@ -162,14 +216,62 @@ public class SnmpOIDPoller extends ScheduledPollConsumer implements ResponseList
/** * @return Returns the target.
*/
- public CommunityTarget getTarget() {
+ public Target getTarget() {
return this.target;
}
/**
* @param target The target to set.
*/
- public void setTarget(CommunityTarget target) {
+ public void setTarget(Target target) {
this.target = target;
}
+
+ private OctetString convertToOctetString(String value) {
+ if (value == null) {
+ return null;
+ }
+ return new OctetString(value);
+ }
+
+ private OID convertAuthenticationProtocol(String authenticationProtocol) {
+ if (authenticationProtocol == null) {
+ return null;
+ }
+ if ("MD5".equals(authenticationProtocol)) {
+ return AuthMD5.ID;
+ } else if ("SHA1".equals(authenticationProtocol)) {
+ return AuthSHA.ID;
+ } else {
+ throw new IllegalArgumentException("Unknown authentication protocol: " + authenticationProtocol);
+ }
+ }
+
+ private OID convertPrivacyProtocol(String privacyProtocol) {
+ if (privacyProtocol == null) {
+ return null;
+ }
+ if ("DES".equals(privacyProtocol)) {
+ return PrivDES.ID;
+ } else if ("TRIDES".equals(privacyProtocol)) {
+ return Priv3DES.ID;
+ } else if ("AES128".equals(privacyProtocol)) {
+ return PrivAES128.ID;
+ } else if ("AES192".equals(privacyProtocol)) {
+ return PrivAES192.ID;
+ } else if ("AES256".equals(privacyProtocol)) {
+ return PrivAES256.ID;
+ } else {
+ throw new IllegalArgumentException("Unknown privacy protocol: " + privacyProtocol);
+ }
+ }
+
+ private int getPduType(SnmpActionType type) {
+ if (SnmpActionType.GET_NEXT == type) {
+ return PDU.GETNEXT;
+ } else {
+ return PDU.GET;
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d4780c75/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpPrivacyProtocolType.java
----------------------------------------------------------------------
diff --git a/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpPrivacyProtocolType.java b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpPrivacyProtocolType.java
new file mode 100644
index 0000000..d91ae78
--- /dev/null
+++ b/components/camel-snmp/src/main/java/org/apache/camel/component/snmp/SnmpPrivacyProtocolType.java
@@ -0,0 +1,21 @@
+/**
+ * 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.snmp;
+
+public enum SnmpPrivacyProtocolType {
+ DES, TRIDES, AES128, AES192, AES256;
+}