aboutsummaryrefslogtreecommitdiffstats
path: root/src/main
diff options
context:
space:
mode:
authorDaniel Silverthorn <daniel.silverthorn@amdocs.com>2018-05-25 16:09:38 -0400
committerDaniel Silverthorn <daniel.silverthorn@amdocs.com>2018-05-25 16:10:36 -0400
commit6891c2008b93e09f5a67e45b859696cd13466f8a (patch)
treee7401e68f98c38432fce0bcfec7e5333be92a896 /src/main
parent98f4f48c7d0352e2416123067e88a24cb3708f8c (diff)
Remove cambria client dependency
Change-Id: I9760839ae44df851640b271d032a39f4bb3691c2 Issue-ID: AAI-1182 Signed-off-by: Daniel Silverthorn <daniel.silverthorn@amdocs.com>
Diffstat (limited to 'src/main')
-rw-r--r--src/main/java/org/onap/aai/event/AbstractEventBusEndpoint.java43
-rw-r--r--src/main/java/org/onap/aai/event/DMaaPEventBusComponent.java (renamed from src/main/java/org/onap/aai/event/EventBusComponent.java)14
-rw-r--r--src/main/java/org/onap/aai/event/DMaaPEventBusEndpoint.java (renamed from src/main/java/org/onap/aai/event/EventBusEndpoint.java)87
-rw-r--r--src/main/java/org/onap/aai/event/EventBusConsumer.java52
-rw-r--r--src/main/java/org/onap/aai/event/EventBusProducer.java4
-rw-r--r--src/main/java/org/onap/aai/event/KafkaEventBusComponent.java49
-rw-r--r--src/main/java/org/onap/aai/event/KafkaEventBusEndpoint.java129
-rw-r--r--src/main/resources/META-INF/services/org/apache/camel/component/dmaap-event-bus1
-rw-r--r--src/main/resources/META-INF/services/org/apache/camel/component/event-bus1
-rw-r--r--src/main/resources/META-INF/services/org/apache/camel/component/kafka-event-bus1
10 files changed, 298 insertions, 83 deletions
diff --git a/src/main/java/org/onap/aai/event/AbstractEventBusEndpoint.java b/src/main/java/org/onap/aai/event/AbstractEventBusEndpoint.java
new file mode 100644
index 0000000..abbfe63
--- /dev/null
+++ b/src/main/java/org/onap/aai/event/AbstractEventBusEndpoint.java
@@ -0,0 +1,43 @@
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2017-2018 Amdocs
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.aai.event;
+
+import org.apache.camel.Component;
+import org.apache.camel.impl.DefaultEndpoint;
+
+public abstract class AbstractEventBusEndpoint extends DefaultEndpoint {
+ public AbstractEventBusEndpoint() {
+ }
+
+ public AbstractEventBusEndpoint(String endpointUri, Component component) {
+ super(endpointUri, component);
+ }
+
+ public AbstractEventBusEndpoint(String endpointUri) {
+ super(endpointUri);
+ }
+
+ abstract void close();
+ abstract int getPollingDelay();
+ abstract int getPoolSize();
+ abstract String getEventTopic();
+
+}
diff --git a/src/main/java/org/onap/aai/event/EventBusComponent.java b/src/main/java/org/onap/aai/event/DMaaPEventBusComponent.java
index b257c7a..696e0f4 100644
--- a/src/main/java/org/onap/aai/event/EventBusComponent.java
+++ b/src/main/java/org/onap/aai/event/DMaaPEventBusComponent.java
@@ -27,22 +27,22 @@ import org.apache.camel.impl.UriEndpointComponent;
import java.util.Map;
/**
- * Represents the component that manages {@link EventBusEndpoint}.
+ * Represents the component that manages {@link DMaaPEventBusEndpoint}.
*/
-public class EventBusComponent extends UriEndpointComponent {
+public class DMaaPEventBusComponent extends UriEndpointComponent {
- public EventBusComponent() {
- super(EventBusEndpoint.class);
+ public DMaaPEventBusComponent() {
+ super(DMaaPEventBusEndpoint.class);
}
- public EventBusComponent(CamelContext context) {
- super(context, EventBusEndpoint.class);
+ public DMaaPEventBusComponent(CamelContext context) {
+ super(context, DMaaPEventBusEndpoint.class);
}
@Override
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters)
throws Exception {
- Endpoint endpoint = new EventBusEndpoint(uri, this);
+ Endpoint endpoint = new DMaaPEventBusEndpoint(uri, this);
setProperties(endpoint, parameters);
return endpoint;
}
diff --git a/src/main/java/org/onap/aai/event/EventBusEndpoint.java b/src/main/java/org/onap/aai/event/DMaaPEventBusEndpoint.java
index ee7cba8..36e930a 100644
--- a/src/main/java/org/onap/aai/event/EventBusEndpoint.java
+++ b/src/main/java/org/onap/aai/event/DMaaPEventBusEndpoint.java
@@ -24,19 +24,19 @@ package org.onap.aai.event;
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.Metadata;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.eclipse.jetty.util.security.Password;
+import org.onap.aai.event.client.DMaaPEventConsumer;
/**
* Represents a EventBus endpoint.
*/
-@UriEndpoint(scheme = "event-bus", syntax = "event-bus:name",
- consumerClass = EventBusConsumer.class, title = "event-bus")
-public class EventBusEndpoint extends DefaultEndpoint {
+@UriEndpoint(scheme = "dmaap-event-bus", syntax = "dmaap-event-bus:name",
+ consumerClass = EventBusConsumer.class, title = "dmaap-event-bus")
+public class DMaaPEventBusEndpoint extends AbstractEventBusEndpoint {
@UriPath
@Metadata(required = "true")
private String name;
@@ -44,16 +44,16 @@ public class EventBusEndpoint extends DefaultEndpoint {
@UriParam(label = "eventTopic")
@Metadata(required = "true")
private String eventTopic;
- @UriParam(label = "groupName")
+ @UriParam(label = "consumerGroup")
@Metadata(required = "true")
- private String groupName;
- @UriParam(label = "groupId")
+ private String consumerGroup;
+ @UriParam(label = "consumerId")
@Metadata(required = "true")
- private String groupId;
- @UriParam(label = "apiKey")
- private String apiKey;
- @UriParam(label = "apiSecret")
- private String apiSecret;
+ private String consumerId;
+ @UriParam(label = "username")
+ private String username;
+ @UriParam(label = "password")
+ private String password;
@UriParam(label = "url")
@Metadata(required = "true")
private String url;
@@ -63,23 +63,36 @@ public class EventBusEndpoint extends DefaultEndpoint {
@UriParam(label = "pollingDelay")
@Metadata(required = "true", defaultValue="30000")
private int pollingDelay = 30000;
+ @UriParam(label = "transportType")
+ @Metadata(required = "true", defaultValue="HTTPAUTH")
+ private String transportType;
+
+ private DMaaPEventConsumer dmaapConsumer;
- public EventBusEndpoint() {}
+ public DMaaPEventBusEndpoint() {}
- public EventBusEndpoint(String uri, EventBusComponent component) {
+ public DMaaPEventBusEndpoint(String uri, DMaaPEventBusComponent component) {
super(uri, component);
}
- public EventBusEndpoint(String endpointUri) {
+ public DMaaPEventBusEndpoint(String endpointUri) {
super(endpointUri);
}
+
+ @Override
+ void close() {
+ // Don't have to do anything for DMaaP
+ }
+
@Override
public Producer createProducer() throws Exception {
return new EventBusProducer(this);
}
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- return new EventBusConsumer(this, processor);
+ // TODO: other overloads based on filled-in properties
+ dmaapConsumer = new DMaaPEventConsumer(url, eventTopic, Password.deobfuscate(username), Password.deobfuscate(password), consumerGroup, consumerId, 15000, 1000, transportType);
+ return new EventBusConsumer(this, processor, dmaapConsumer);
}
@Override
public boolean isSingleton() {
@@ -102,36 +115,38 @@ public class EventBusEndpoint extends DefaultEndpoint {
this.eventTopic = eventTopic;
}
- public String getGroupName() {
- return groupName;
+ public String getConsumerGroup() {
+ return consumerGroup;
}
- public void setGroupName(String groupName) {
- this.groupName = groupName;
+ public void setConsumerGroup(String consumerGroup) {
+ this.consumerGroup = consumerGroup;
}
- public String getGroupId() {
- return groupId;
+ public String getConsumerId() {
+ return consumerId;
}
- public void setGroupId(String groupId) {
- this.groupId = groupId;
+ public void setConsumerId(String consumerId) {
+ this.consumerId = consumerId;
}
- public String getApiKey() {
- return apiKey == null ? null : Password.deobfuscate(apiKey);
+ public String getUsername() {
+ return username == null ? null : Password.deobfuscate(username);
+ //return username;
}
- public void setApiKey(String apiKey) {
- this.apiKey = apiKey;
+ public void setUsername(String username) {
+ this.username = username;
}
- public String getApiSecret() {
- return apiSecret == null ? null : Password.deobfuscate(apiSecret);
+ public String getPassword() {
+ return password == null ? null : Password.deobfuscate(password);
+ //return password;
}
- public void setApiSecret(String apiSecret) {
- this.apiSecret = apiSecret;
+ public void setPassword(String password) {
+ this.password = password;
}
public int getPoolSize() {
@@ -157,5 +172,13 @@ public class EventBusEndpoint extends DefaultEndpoint {
public void setUrl(String url) {
this.url = url;
}
+
+ public String getTransportType() {
+ return transportType;
+ }
+
+ public void setTransportType(String transportType) {
+ this.transportType = transportType;
+ }
}
diff --git a/src/main/java/org/onap/aai/event/EventBusConsumer.java b/src/main/java/org/onap/aai/event/EventBusConsumer.java
index a9d5478..e795e2c 100644
--- a/src/main/java/org/onap/aai/event/EventBusConsumer.java
+++ b/src/main/java/org/onap/aai/event/EventBusConsumer.java
@@ -20,9 +20,7 @@
*/
package org.onap.aai.event;
-import com.att.nsa.cambria.client.CambriaClientBuilders;
-import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder;
-import com.att.nsa.cambria.client.CambriaConsumer;
+import org.onap.aai.event.api.EventConsumer;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
@@ -33,10 +31,6 @@ import org.onap.aai.cl.api.Logger;
import org.onap.aai.cl.eelf.LoggerFactory;
import org.onap.aai.cl.mdc.MdcContext;
-import java.net.MalformedURLException;
-import java.security.GeneralSecurityException;
-import java.util.Arrays;
-import java.util.List;
import java.util.UUID;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -49,46 +43,21 @@ public class EventBusConsumer extends ScheduledPollConsumer {
private Logger logger = LoggerFactory.getInstance().getLogger(EventBusConsumer.class);
private Logger auditLogger = LoggerFactory.getInstance().getAuditLogger(EventBusConsumer.class);
- private final EventBusEndpoint endpoint;
+ private final AbstractEventBusEndpoint endpoint;
- private CambriaConsumer consumer;
+ private EventConsumer consumer;
/**
* EventBusConsumer Constructor.
*/
- public EventBusConsumer(EventBusEndpoint endpoint, Processor processor) {
+ public EventBusConsumer(AbstractEventBusEndpoint endpoint, Processor processor, EventConsumer consumer) {
super(endpoint, processor);
super.setDelay(endpoint.getPollingDelay());
this.endpoint = endpoint;
setScheduledExecutorService(new ScheduledThreadPoolExecutor(endpoint.getPoolSize()));
- String[] urls = endpoint.getUrl().split(",");
-
- List<String> urlList = null;
-
- if (urls != null) {
- urlList = Arrays.asList(urls);
- }
-
- try {
-
- ConsumerBuilder consumerBuilder = new CambriaClientBuilders.ConsumerBuilder()
- .usingHosts(urlList).onTopic(endpoint.getEventTopic())
- .knownAs(endpoint.getGroupName(), endpoint.getGroupId());
-
- String apiKey = endpoint.getApiKey();
- String apiSecret = endpoint.getApiSecret();
-
- if (apiKey != null && apiSecret != null) {
- consumerBuilder.authenticatedBy(endpoint.getApiKey(), endpoint.getApiSecret());
- }
-
- consumer = consumerBuilder.build();
-
- } catch (MalformedURLException | GeneralSecurityException e) {
- logger.error(RouterCoreMsgs.EVENT_CONSUMER_CREATION_EXCEPTION, e, e.getLocalizedMessage());
- }
+ this.consumer = consumer;
}
/**
@@ -104,7 +73,8 @@ public class EventBusConsumer extends ScheduledPollConsumer {
int processCount = 0;
- Iterable<String> messages = consumer.fetch();
+ //Iterable<String> messages = consumer.fetch();
+ Iterable<String> messages = consumer.consumeAndCommit();
String topic = endpoint.getEventTopic();
@@ -119,15 +89,15 @@ public class EventBusConsumer extends ScheduledPollConsumer {
@Override
protected void doStop() throws Exception {
super.doStop();
- if (consumer != null) {
- consumer.close();
+ if (endpoint != null) {
+ endpoint.close();
}
}
@Override
protected void doShutdown() throws Exception {
super.doShutdown();
- if (consumer != null) {
- consumer.close();
+ if (endpoint != null) {
+ endpoint.close();
}
}
diff --git a/src/main/java/org/onap/aai/event/EventBusProducer.java b/src/main/java/org/onap/aai/event/EventBusProducer.java
index ec2a0da..dd8dbed 100644
--- a/src/main/java/org/onap/aai/event/EventBusProducer.java
+++ b/src/main/java/org/onap/aai/event/EventBusProducer.java
@@ -27,9 +27,9 @@ import org.apache.camel.impl.DefaultProducer;
* The EventBus producer.
*/
public class EventBusProducer extends DefaultProducer {
- private EventBusEndpoint endpoint;
+ private AbstractEventBusEndpoint endpoint;
- public EventBusProducer(EventBusEndpoint endpoint) {
+ public EventBusProducer(AbstractEventBusEndpoint endpoint) {
super(endpoint);
this.endpoint = endpoint;
}
diff --git a/src/main/java/org/onap/aai/event/KafkaEventBusComponent.java b/src/main/java/org/onap/aai/event/KafkaEventBusComponent.java
new file mode 100644
index 0000000..07b4c82
--- /dev/null
+++ b/src/main/java/org/onap/aai/event/KafkaEventBusComponent.java
@@ -0,0 +1,49 @@
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2017-2018 Amdocs
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.aai.event;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.UriEndpointComponent;
+
+import java.util.Map;
+
+/**
+ * Represents the component that manages {@link KafkaEventBusEndpoint}.
+ */
+public class KafkaEventBusComponent extends UriEndpointComponent {
+
+ public KafkaEventBusComponent() {
+ super(KafkaEventBusEndpoint.class);
+ }
+
+ public KafkaEventBusComponent(CamelContext context) {
+ super(context, KafkaEventBusEndpoint.class);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters)
+ throws Exception {
+ Endpoint endpoint = new KafkaEventBusEndpoint(uri, this);
+ setProperties(endpoint, parameters);
+ return endpoint;
+ }
+}
diff --git a/src/main/java/org/onap/aai/event/KafkaEventBusEndpoint.java b/src/main/java/org/onap/aai/event/KafkaEventBusEndpoint.java
new file mode 100644
index 0000000..4194c89
--- /dev/null
+++ b/src/main/java/org/onap/aai/event/KafkaEventBusEndpoint.java
@@ -0,0 +1,129 @@
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright © 2017-2018 Amdocs
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.aai.event;
+
+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.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.onap.aai.event.client.KafkaEventConsumer;
+
+/**
+ * Represents a EventBus endpoint.
+ */
+@UriEndpoint(scheme = "kafka-event-bus", syntax = "kafka-event-bus:name",
+ consumerClass = EventBusConsumer.class, title = "kafka-event-bus")
+public class KafkaEventBusEndpoint extends AbstractEventBusEndpoint {
+ @UriParam(label = "url")
+ @Metadata(required = "true")
+ private String url;
+ @UriParam(label = "eventTopic")
+ @Metadata(required = "true")
+ private String eventTopic;
+ @UriParam(label = "consumerGroup")
+ @Metadata(required = "true")
+ private String consumerGroup;
+ @UriParam(label = "poolSize")
+ @Metadata(required = "true", defaultValue="20")
+ private int poolSize = 20;
+ @UriParam(label = "pollingDelay")
+ @Metadata(required = "true", defaultValue="30000")
+ private int pollingDelay = 30000;
+
+ private KafkaEventConsumer consumer;
+
+ public KafkaEventBusEndpoint(String uri, KafkaEventBusComponent component) {
+ super(uri, component);
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new EventBusProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ consumer = new KafkaEventConsumer(url, eventTopic, consumerGroup);
+ return new EventBusConsumer(this, processor, consumer);
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return false;
+ }
+
+ public String getUrl() {
+ return url;
+ }
+
+ public void setUrl(String url) {
+ this.url = url;
+ }
+
+ @Override
+ String getEventTopic() {
+ return eventTopic;
+ }
+
+ public void setEventTopic(String eventTopic) {
+ this.eventTopic = eventTopic;
+ }
+
+ public String getConsumerGroup() {
+ return consumerGroup;
+ }
+
+ public void setConsumerGroup(String consumerGroup) {
+ this.consumerGroup = consumerGroup;
+ }
+
+ @Override
+ int getPoolSize() {
+ return poolSize;
+ }
+
+ public void setPoolSize(int poolSize) {
+ this.poolSize = poolSize;
+ }
+
+ @Override
+ void close() {
+ consumer.close();
+ }
+
+ @Override
+ int getPollingDelay() {
+ return pollingDelay;
+ }
+
+ public void setPollingDelay(int pollingDelay) {
+ this.pollingDelay = pollingDelay;
+ }
+
+
+
+
+
+
+}
diff --git a/src/main/resources/META-INF/services/org/apache/camel/component/dmaap-event-bus b/src/main/resources/META-INF/services/org/apache/camel/component/dmaap-event-bus
new file mode 100644
index 0000000..f1fee02
--- /dev/null
+++ b/src/main/resources/META-INF/services/org/apache/camel/component/dmaap-event-bus
@@ -0,0 +1 @@
+class=org.onap.aai.event.DMaaPEventBusComponent
diff --git a/src/main/resources/META-INF/services/org/apache/camel/component/event-bus b/src/main/resources/META-INF/services/org/apache/camel/component/event-bus
deleted file mode 100644
index f795067..0000000
--- a/src/main/resources/META-INF/services/org/apache/camel/component/event-bus
+++ /dev/null
@@ -1 +0,0 @@
-class=org.onap.aai.event.EventBusComponent
diff --git a/src/main/resources/META-INF/services/org/apache/camel/component/kafka-event-bus b/src/main/resources/META-INF/services/org/apache/camel/component/kafka-event-bus
new file mode 100644
index 0000000..89f243d
--- /dev/null
+++ b/src/main/resources/META-INF/services/org/apache/camel/component/kafka-event-bus
@@ -0,0 +1 @@
+class=org.onap.aai.event.KafkaEventBusComponent