Copy policy-endpoints from drools-pdp to common

Issue-ID: POLICY-967
Change-Id: I374c155ee102c3e157c60d0a22d7191544abb76a
Signed-off-by: mmis <michael.morris@ericsson.com>

19 files changed, 2535 insertions, 0 deletions

diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumer.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumer.java
new file mode 100644
index 00000000..8cc631c8
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumer.java
@@ -0,0 +1,45 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+import java.io.IOException;
+
+/**
+ * Wrapper around libraries to consume from message bus
+ *
+ */
+public interface BusConsumer {
+
+    /**
+     * fetch messages
+     *
+     * @return list of messages
+     * @throws Exception when error encountered by underlying libraries
+     */
+    public Iterable<String> fetch() throws InterruptedException, IOException;
+
+    /**
+     * close underlying library consumer
+     */
+    public void close();
+}
+
+
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisher.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisher.java
new file mode 100644
index 00000000..f6c65ada
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisher.java
@@ -0,0 +1,39 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+public interface BusPublisher {
+
+    /**
+     * sends a message
+     * 
+     * @param partition id
+     * @param message the message
+     * @return true if success, false otherwise
+     * @throws IllegalArgumentException if no message provided
+     */
+    public boolean send(String partitionId, String message);
+
+    /**
+     * closes the publisher
+     */
+    public void close();
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBase.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBase.java
new file mode 100644
index 00000000..7f4c0ddd
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBase.java
@@ -0,0 +1,127 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+import java.util.List;
+
+import org.onap.policy.common.endpoints.event.comm.bus.ApiKeyEnabled;
+
+/**
+ * Bus Topic Base
+ */
+public abstract class BusTopicBase extends TopicBase implements ApiKeyEnabled {
+
+    /**
+     * API Key
+     */
+    protected String apiKey;
+
+    /**
+     * API Secret
+     */
+    protected String apiSecret;
+
+    /**
+     * Use https
+     */
+    protected boolean useHttps;
+
+    /**
+     * allow self signed certificates
+     */
+    protected boolean allowSelfSignedCerts;
+
+    /**
+     * Instantiates a new Bus Topic Base
+     * 
+     * @param servers list of servers
+     * @param topic topic name
+     * @param apiKey API Key
+     * @param apiSecret API Secret
+     * @param useHttps does connection use HTTPS?
+     * @param allowSelfSignedCerts are self-signed certificates allow
+     * 
+     * @return a Bus Topic Base
+     * @throws IllegalArgumentException if invalid parameters are present
+     */
+    public BusTopicBase(List<String> servers, String topic, String apiKey, String apiSecret, boolean useHttps,
+            boolean allowSelfSignedCerts) {
+
+        super(servers, topic);
+
+        this.apiKey = apiKey;
+        this.apiSecret = apiSecret;
+        this.useHttps = useHttps;
+        this.allowSelfSignedCerts = allowSelfSignedCerts;
+    }
+
+    @Override
+    public String getApiKey() {
+        return apiKey;
+    }
+
+    @Override
+    public String getApiSecret() {
+        return apiSecret;
+    }
+
+    /**
+     * @return if using https
+     */
+    public boolean isUseHttps() {
+        return useHttps;
+    }
+
+    /**
+     * @return if self signed certificates are allowed
+     */
+    public boolean isAllowSelfSignedCerts() {
+        return allowSelfSignedCerts;
+    }
+
+    protected boolean anyNullOrEmpty(String... args) {
+        for (String arg : args) {
+            if (arg == null || arg.isEmpty()) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    protected boolean allNullOrEmpty(String... args) {
+        for (String arg : args) {
+            if (!(arg == null || arg.isEmpty())) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+
+    @Override
+    public String toString() {
+        return "BusTopicBase [apiKey=" + apiKey + ", apiSecret=" + apiSecret + ", useHttps=" + useHttps
+                + ", allowSelfSignedCerts=" + allowSelfSignedCerts + ", toString()=" + super.toString() + "]";
+    }
+
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/FilterableBusConsumer.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/FilterableBusConsumer.java
new file mode 100644
index 00000000..76adf980
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/FilterableBusConsumer.java
@@ -0,0 +1,35 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+/**
+ * BusConsumer that supports server-side filtering.
+ */
+public interface FilterableBusConsumer extends BusConsumer {
+
+    /**
+     * Sets the server-side filter.
+     * 
+     * @param filter new filter value, or {@code null}
+     * @throws IllegalArgumentException if the consumer cannot be built with the new filter
+     */
+    public void setFilter(String filter);
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSink.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSink.java
new file mode 100644
index 00000000..f3c736da
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSink.java
@@ -0,0 +1,215 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.onap.policy.common.endpoints.event.comm.bus.BusTopicSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Transport Agnostic Bus Topic Sink to carry out the core functionality to interact with a sink
+ * regardless if it is UEB or DMaaP.
+ *
+ */
+public abstract class InlineBusTopicSink extends BusTopicBase implements BusTopicSink {
+
+    /**
+     * loggers
+     */
+    private static Logger logger = LoggerFactory.getLogger(InlineBusTopicSink.class);
+    private static final Logger netLogger = LoggerFactory.getLogger(NETWORK_LOGGER);
+
+    /**
+     * The partition key to publish to
+     */
+    protected String partitionId;
+
+    /**
+     * message bus publisher
+     */
+    protected BusPublisher publisher;
+
+    /**
+     * constructor for abstract sink
+     * 
+     * @param servers servers
+     * @param topic topic
+     * @param apiKey api secret
+     * @param apiSecret api secret
+     * @param partitionId partition id
+     * @param useHttps does connection use HTTPS?
+     * @param allowSelfSignedCerts are self-signed certificates allow
+     * @throws IllegalArgumentException in invalid parameters are passed in
+     */
+    public InlineBusTopicSink(List<String> servers, String topic, String apiKey, String apiSecret, String partitionId,
+            boolean useHttps, boolean allowSelfSignedCerts) {
+
+        super(servers, topic, apiKey, apiSecret, useHttps, allowSelfSignedCerts);
+
+        if (partitionId == null || partitionId.isEmpty()) {
+            this.partitionId = UUID.randomUUID().toString();
+        }
+    }
+
+    /**
+     * Initialize the Bus publisher
+     */
+    public abstract void init();
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean start() {
+        logger.info("{}: starting", this);
+
+        synchronized (this) {
+
+            if (this.alive) {
+                return true;
+            }
+
+            if (locked) {
+                throw new IllegalStateException(this + " is locked.");
+            }
+
+            this.alive = true;
+        }
+
+        this.init();
+        return true;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean stop() {
+
+        BusPublisher publisherCopy;
+        synchronized (this) {
+            this.alive = false;
+            publisherCopy = this.publisher;
+            this.publisher = null;
+        }
+
+        if (publisherCopy != null) {
+            try {
+                publisherCopy.close();
+            } catch (Exception e) {
+                logger.warn("{}: cannot stop publisher because of {}", this, e.getMessage(), e);
+            }
+        } else {
+            logger.warn("{}: there is no publisher", this);
+            return false;
+        }
+
+        return true;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean send(String message) {
+
+        if (message == null || message.isEmpty()) {
+            throw new IllegalArgumentException("Message to send is empty");
+        }
+
+        if (!this.alive) {
+            throw new IllegalStateException(this + " is stopped");
+        }
+
+        try {
+            synchronized (this) {
+                this.recentEvents.add(message);
+            }
+
+            netLogger.info("[OUT|{}|{}]{}{}", this.getTopicCommInfrastructure(), this.topic, System.lineSeparator(),
+                    message);
+
+            publisher.send(this.partitionId, message);
+            broadcast(message);
+        } catch (Exception e) {
+            logger.warn("{}: cannot send because of {}", this, e.getMessage(), e);
+            return false;
+        }
+
+        return true;
+    }
+
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void setPartitionKey(String partitionKey) {
+        this.partitionId = partitionKey;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public String getPartitionKey() {
+        return this.partitionId;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void shutdown() {
+        this.stop();
+    }
+
+    @Override
+    protected boolean anyNullOrEmpty(String... args) {
+        for (String arg : args) {
+            if (arg == null || arg.isEmpty()) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    @Override
+    protected boolean allNullOrEmpty(String... args) {
+        for (String arg : args) {
+            if (!(arg == null || arg.isEmpty())) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+
+    @Override
+    public String toString() {
+        return "InlineBusTopicSink [partitionId=" + partitionId + ", alive=" + alive + ", publisher=" + publisher + "]";
+    }
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSink.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSink.java
new file mode 100644
index 00000000..2ad8bb34
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSink.java
@@ -0,0 +1,133 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+import java.util.List;
+import java.util.Map;
+
+import org.onap.policy.common.endpoints.event.comm.Topic;
+import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicSink;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.impl.CambriaPublisherWrapper;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.impl.DmaapDmePublisherWrapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This implementation publishes events for the associated DMAAP topic, inline with the calling
+ * thread.
+ */
+public class InlineDmaapTopicSink extends InlineBusTopicSink implements DmaapTopicSink {
+
+    protected static Logger logger = LoggerFactory.getLogger(InlineDmaapTopicSink.class);
+
+    protected final String userName;
+    protected final String password;
+
+    protected String environment = null;
+    protected String aftEnvironment = null;
+    protected String partner = null;
+    protected String latitude = null;
+    protected String longitude = null;
+
+    protected Map<String, String> additionalProps = null;
+
+    /**
+     * 
+     * @param servers DMaaP servers
+     * @param topic DMaaP Topic to be monitored
+     * @param apiKey DMaaP API Key (optional)
+     * @param apiSecret DMaaP API Secret (optional)
+     * @param consumerGroup DMaaP Reader Consumer Group
+     * @param consumerInstance DMaaP Reader Instance
+     * @param fetchTimeout DMaaP fetch timeout
+     * @param fetchLimit DMaaP fetch limit
+     * @param environment DME2 Environment
+     * @param aftEnvironment DME2 AFT Environment
+     * @param partner DME2 Partner
+     * @param latitude DME2 Latitude
+     * @param longitude DME2 Longitude
+     * @param additionalProps Additional properties to pass to DME2
+     * @param useHttps does connection use HTTPS?
+     * @param allowSelfSignedCerts are self-signed certificates allow
+     * 
+     * @throws IllegalArgumentException An invalid parameter passed in
+     */
+    public InlineDmaapTopicSink(List<String> servers, String topic, String apiKey, String apiSecret, String userName,
+            String password, String partitionKey, String environment, String aftEnvironment, String partner,
+            String latitude, String longitude, Map<String, String> additionalProps, boolean useHttps,
+            boolean allowSelfSignedCerts) {
+
+        super(servers, topic, apiKey, apiSecret, partitionKey, useHttps, allowSelfSignedCerts);
+
+        this.userName = userName;
+        this.password = password;
+
+        this.environment = environment;
+        this.aftEnvironment = aftEnvironment;
+        this.partner = partner;
+
+        this.latitude = latitude;
+        this.longitude = longitude;
+
+        this.additionalProps = additionalProps;
+    }
+
+    public InlineDmaapTopicSink(List<String> servers, String topic, String apiKey, String apiSecret, String userName,
+            String password, String partitionKey, boolean useHttps, boolean allowSelfSignedCerts) {
+
+        super(servers, topic, apiKey, apiSecret, partitionKey, useHttps, allowSelfSignedCerts);
+
+        this.userName = userName;
+        this.password = password;
+    }
+
+
+    @Override
+    public void init() {
+        if (allNullOrEmpty(this.environment, this.aftEnvironment, this.latitude, this.longitude, this.partner)) {
+            this.publisher = new CambriaPublisherWrapper(this.servers, this.topic, this.apiKey, this.apiSecret,
+                    this.userName, this.password, this.useHttps);
+        } else {
+            this.publisher = new DmaapDmePublisherWrapper(this.servers, this.topic, this.userName, this.password,
+                    this.environment, this.aftEnvironment, this.partner, this.latitude, this.longitude,
+                    this.additionalProps, this.useHttps);
+        }
+
+        logger.info("{}: DMAAP SINK created", this);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public CommInfrastructure getTopicCommInfrastructure() {
+        return Topic.CommInfrastructure.DMAAP;
+    }
+
+
+    @Override
+    public String toString() {
+        return "InlineDmaapTopicSink [userName=" + userName + ", password=" + password
+                + ", getTopicCommInfrastructure()=" + getTopicCommInfrastructure() + ", toString()=" + super.toString()
+                + "]";
+    }
+
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSink.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSink.java
new file mode 100644
index 00000000..24cf6073
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSink.java
@@ -0,0 +1,86 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+import java.util.List;
+
+import org.onap.policy.common.endpoints.event.comm.Topic;
+import org.onap.policy.common.endpoints.event.comm.bus.UebTopicSink;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.impl.CambriaPublisherWrapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This implementation publishes events for the associated UEB topic, inline with the calling
+ * thread.
+ */
+public class InlineUebTopicSink extends InlineBusTopicSink implements UebTopicSink {
+
+    /**
+     * logger
+     */
+    private static Logger logger = LoggerFactory.getLogger(InlineUebTopicSink.class);
+
+    /**
+     * Argument-based UEB Topic Writer instantiation
+     * 
+     * @param servers list of UEB servers available for publishing
+     * @param topic the topic to publish to
+     * @param apiKey the api key (optional)
+     * @param apiSecret the api secret (optional)
+     * @param partitionId the partition key (optional, autogenerated if not provided)
+     * @param useHttps does connection use HTTPS?
+     * @param allowSelfSignedCerts are self-signed certificates allow
+     * 
+     * @throws IllegalArgumentException if invalid arguments are detected
+     */
+    public InlineUebTopicSink(List<String> servers, String topic, String apiKey, String apiSecret, String partitionId,
+            boolean useHttps, boolean allowSelfSignedCerts) {
+        super(servers, topic, apiKey, apiSecret, partitionId, useHttps, allowSelfSignedCerts);
+    }
+
+    /**
+     * Instantiation of internal resources
+     */
+    @Override
+    public void init() {
+
+        this.publisher =
+                new CambriaPublisherWrapper(this.servers, this.topic, this.apiKey, this.apiSecret, this.useHttps);
+        logger.info("{}: UEB SINK created", this);
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder builder = new StringBuilder();
+        builder.append("InlineUebTopicSink [getTopicCommInfrastructure()=").append(getTopicCommInfrastructure())
+                .append(", toString()=").append(super.toString()).append("]");
+        return builder.toString();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public CommInfrastructure getTopicCommInfrastructure() {
+        return Topic.CommInfrastructure.UEB;
+    }
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSource.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSource.java
new file mode 100644
index 00000000..d59c63f5
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSource.java
@@ -0,0 +1,324 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+import java.net.MalformedURLException;
+import java.util.List;
+import java.util.UUID;
+
+import org.onap.policy.common.endpoints.event.comm.FilterableTopicSource;
+import org.onap.policy.common.endpoints.event.comm.TopicListener;
+import org.onap.policy.common.endpoints.event.comm.bus.BusTopicSource;
+import org.onap.policy.common.utils.network.NetworkUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This topic source implementation specializes in reading messages over a bus topic source and
+ * notifying its listeners
+ */
+public abstract class SingleThreadedBusTopicSource extends BusTopicBase
+        implements Runnable, BusTopicSource, FilterableTopicSource {
+
+    /**
+     * Not to be converted to PolicyLogger. This will contain all instract /out traffic and only
+     * that in a single file in a concise format.
+     */
+    private static Logger logger = LoggerFactory.getLogger(InlineBusTopicSink.class);
+    private static final Logger netLogger = LoggerFactory.getLogger(NETWORK_LOGGER);
+
+    /**
+     * Bus consumer group
+     */
+    protected final String consumerGroup;
+
+    /**
+     * Bus consumer instance
+     */
+    protected final String consumerInstance;
+
+    /**
+     * Bus fetch timeout
+     */
+    protected final int fetchTimeout;
+
+    /**
+     * Bus fetch limit
+     */
+    protected final int fetchLimit;
+
+    /**
+     * Message Bus Consumer
+     */
+    protected BusConsumer consumer;
+
+    /**
+     * Independent thread reading message over my topic
+     */
+    protected Thread busPollerThread;
+
+
+    /**
+     * 
+     * @param servers Bus servers
+     * @param topic Bus Topic to be monitored
+     * @param apiKey Bus API Key (optional)
+     * @param apiSecret Bus API Secret (optional)
+     * @param consumerGroup Bus Reader Consumer Group
+     * @param consumerInstance Bus Reader Instance
+     * @param fetchTimeout Bus fetch timeout
+     * @param fetchLimit Bus fetch limit
+     * @param useHttps does the bus use https
+     * @param allowSelfSignedCerts are self-signed certificates allowed
+     * @throws IllegalArgumentException An invalid parameter passed in
+     */
+    public SingleThreadedBusTopicSource(List<String> servers, String topic, String apiKey, String apiSecret,
+            String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit, boolean useHttps,
+            boolean allowSelfSignedCerts) {
+
+        super(servers, topic, apiKey, apiSecret, useHttps, allowSelfSignedCerts);
+
+        if (consumerGroup == null || consumerGroup.isEmpty()) {
+            this.consumerGroup = UUID.randomUUID().toString();
+        } else {
+            this.consumerGroup = consumerGroup;
+        }
+
+        if (consumerInstance == null || consumerInstance.isEmpty()) {
+            this.consumerInstance = NetworkUtil.getHostname();
+        } else {
+            this.consumerInstance = consumerInstance;
+        }
+
+        if (fetchTimeout <= 0) {
+            this.fetchTimeout = NO_TIMEOUT_MS_FETCH;
+        } else {
+            this.fetchTimeout = fetchTimeout;
+        }
+
+        if (fetchLimit <= 0) {
+            this.fetchLimit = NO_LIMIT_FETCH;
+        } else {
+            this.fetchLimit = fetchLimit;
+        }
+
+    }
+
+    /**
+     * Initialize the Bus client
+     */
+    public abstract void init() throws MalformedURLException;
+
+    @Override
+    public void register(TopicListener topicListener) {
+
+        super.register(topicListener);
+
+        try {
+            if (!alive && !locked) {
+                this.start();
+            } else {
+                logger.info("{}: register: start not attempted", this);
+            }
+        } catch (Exception e) {
+            logger.warn("{}: cannot start after registration of because of: {}", this, topicListener, e.getMessage(),
+                    e);
+        }
+    }
+
+    @Override
+    public void unregister(TopicListener topicListener) {
+        boolean stop;
+        synchronized (this) {
+            super.unregister(topicListener);
+            stop = this.topicListeners.isEmpty();
+        }
+
+        if (stop) {
+            this.stop();
+        }
+    }
+
+    @Override
+    public boolean start() {
+        logger.info("{}: starting", this);
+
+        synchronized (this) {
+
+            if (alive) {
+                return true;
+            }
+
+            if (locked) {
+                throw new IllegalStateException(this + " is locked.");
+            }
+
+            if (this.busPollerThread == null || !this.busPollerThread.isAlive() || this.consumer == null) {
+
+                try {
+                    this.init();
+                    this.alive = true;
+                    this.busPollerThread = new Thread(this);
+                    this.busPollerThread.setName(this.getTopicCommInfrastructure() + "-source-" + this.getTopic());
+                    busPollerThread.start();
+                } catch (Exception e) {
+                    logger.warn("{}: cannot start because of {}", this, e.getMessage(), e);
+                    throw new IllegalStateException(e);
+                }
+            }
+        }
+
+        return this.alive;
+    }
+
+    @Override
+    public boolean stop() {
+        logger.info("{}: stopping", this);
+
+        synchronized (this) {
+            BusConsumer consumerCopy = this.consumer;
+
+            this.alive = false;
+            this.consumer = null;
+
+            if (consumerCopy != null) {
+                try {
+                    consumerCopy.close();
+                } catch (Exception e) {
+                    logger.warn("{}: stop failed because of {}", this, e.getMessage(), e);
+                }
+            }
+        }
+
+        Thread.yield();
+
+        return true;
+    }
+
+    /**
+     * Run thread method for the Bus Reader
+     */
+    @Override
+    public void run() {
+        while (this.alive) {
+            try {
+                for (String event : this.consumer.fetch()) {
+                    synchronized (this) {
+                        this.recentEvents.add(event);
+                    }
+
+                    netLogger.info("[IN|{}|{}]{}{}", this.getTopicCommInfrastructure(), this.topic,
+                            System.lineSeparator(), event);
+
+                    broadcast(event);
+
+                    if (!this.alive) {
+                        break;
+                    }
+                }
+            } catch (Exception e) {
+                logger.error("{}: cannot fetch because of ", this, e.getMessage(), e);
+            }
+        }
+
+        logger.info("{}: exiting thread", this);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean offer(String event) {
+        if (!this.alive) {
+            throw new IllegalStateException(this + " is not alive.");
+        }
+
+        synchronized (this) {
+            this.recentEvents.add(event);
+        }
+
+        netLogger.info("[IN|{}|{}]{}{}", this.getTopicCommInfrastructure(), this.topic, System.lineSeparator(), event);
+
+
+        return broadcast(event);
+    }
+
+
+    @Override
+    public void setFilter(String filter) {
+        if (consumer instanceof FilterableBusConsumer) {
+            ((FilterableBusConsumer) consumer).setFilter(filter);
+
+        } else {
+            throw new UnsupportedOperationException("no server-side filtering for topic " + topic);
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "SingleThreadedBusTopicSource [consumerGroup=" + consumerGroup + ", consumerInstance=" + consumerInstance
+                + ", fetchTimeout=" + fetchTimeout + ", fetchLimit=" + fetchLimit + ", consumer=" + this.consumer
+                + ", alive=" + alive + ", locked=" + locked + ", uebThread=" + busPollerThread + ", topicListeners="
+                + topicListeners.size() + ", toString()=" + super.toString() + "]";
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public String getConsumerGroup() {
+        return consumerGroup;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public String getConsumerInstance() {
+        return consumerInstance;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void shutdown() {
+        this.stop();
+        this.topicListeners.clear();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public int getFetchTimeout() {
+        return fetchTimeout;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public int getFetchLimit() {
+        return fetchLimit;
+    }
+
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSource.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSource.java
new file mode 100644
index 00000000..de5fed6a
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSource.java
@@ -0,0 +1,179 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+import java.net.MalformedURLException;
+import java.util.List;
+import java.util.Map;
+
+import org.onap.policy.common.endpoints.event.comm.Topic;
+import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicSource;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.impl.CambriaConsumerWrapper;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.impl.DmaapDmeConsumerWrapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This topic reader implementation specializes in reading messages over DMAAP topic and notifying
+ * its listeners
+ */
+public class SingleThreadedDmaapTopicSource extends SingleThreadedBusTopicSource implements DmaapTopicSource, Runnable {
+
+    private static Logger logger = LoggerFactory.getLogger(SingleThreadedDmaapTopicSource.class);
+
+
+    protected final String userName;
+    protected final String password;
+
+    protected String environment = null;
+    protected String aftEnvironment = null;
+    protected String partner = null;
+    protected String latitude = null;
+    protected String longitude = null;
+
+    protected Map<String, String> additionalProps = null;
+
+
+    /**
+     * 
+     * @param servers DMaaP servers
+     * @param topic DMaaP Topic to be monitored
+     * @param apiKey DMaaP API Key (optional)
+     * @param apiSecret DMaaP API Secret (optional)
+     * @param consumerGroup DMaaP Reader Consumer Group
+     * @param consumerInstance DMaaP Reader Instance
+     * @param fetchTimeout DMaaP fetch timeout
+     * @param fetchLimit DMaaP fetch limit
+     * @param environment DME2 Environment
+     * @param aftEnvironment DME2 AFT Environment
+     * @param partner DME2 Partner
+     * @param latitude DME2 Latitude
+     * @param longitude DME2 Longitude
+     * @param additionalProps Additional properties to pass to DME2
+     * @param useHttps does connection use HTTPS?
+     * @param allowSelfSignedCerts are self-signed certificates allow
+     * 
+     * @throws IllegalArgumentException An invalid parameter passed in
+     */
+    public SingleThreadedDmaapTopicSource(List<String> servers, String topic, String apiKey, String apiSecret,
+            String userName, String password, String consumerGroup, String consumerInstance, int fetchTimeout,
+            int fetchLimit, String environment, String aftEnvironment, String partner, String latitude,
+            String longitude, Map<String, String> additionalProps, boolean useHttps, boolean allowSelfSignedCerts) {
+
+        super(servers, topic, apiKey, apiSecret, consumerGroup, consumerInstance, fetchTimeout, fetchLimit, useHttps,
+                allowSelfSignedCerts);
+
+        this.userName = userName;
+        this.password = password;
+
+        this.environment = environment;
+        this.aftEnvironment = aftEnvironment;
+        this.partner = partner;
+
+        this.latitude = latitude;
+        this.longitude = longitude;
+
+        this.additionalProps = additionalProps;
+        try {
+            this.init();
+        } catch (Exception e) {
+            logger.error("ERROR during init of topic {}", this.topic);
+            throw new IllegalArgumentException(e);
+        }
+    }
+
+    /**
+     * 
+     * @param servers DMaaP servers
+     * @param topic DMaaP Topic to be monitored
+     * @param apiKey DMaaP API Key (optional)
+     * @param apiSecret DMaaP API Secret (optional)
+     * @param consumerGroup DMaaP Reader Consumer Group
+     * @param consumerInstance DMaaP Reader Instance
+     * @param fetchTimeout DMaaP fetch timeout
+     * @param fetchLimit DMaaP fetch limit
+     * @param useHttps does connection use HTTPS?
+     * @param allowSelfSignedCerts are self-signed certificates allow
+     * @throws IllegalArgumentException An invalid parameter passed in
+     */
+    public SingleThreadedDmaapTopicSource(List<String> servers, String topic, String apiKey, String apiSecret,
+            String userName, String password, String consumerGroup, String consumerInstance, int fetchTimeout,
+            int fetchLimit, boolean useHttps, boolean allowSelfSignedCerts) {
+
+
+        super(servers, topic, apiKey, apiSecret, consumerGroup, consumerInstance, fetchTimeout, fetchLimit, useHttps,
+                allowSelfSignedCerts);
+
+        this.userName = userName;
+        this.password = password;
+
+        try {
+            this.init();
+        } catch (Exception e) {
+            logger.warn("dmaap-source: cannot create topic {} because of {}", topic, e.getMessage(), e);
+            throw new IllegalArgumentException(e);
+        }
+    }
+
+
+    /**
+     * Initialize the Cambria or MR Client
+     */
+    @Override
+    public void init() throws MalformedURLException {
+        if (anyNullOrEmpty(this.userName, this.password)) {
+            this.consumer = new CambriaConsumerWrapper(this.servers, this.topic, this.apiKey, this.apiSecret,
+                    this.consumerGroup, this.consumerInstance, this.fetchTimeout, this.fetchLimit, this.useHttps,
+                    this.allowSelfSignedCerts);
+        } else if (allNullOrEmpty(this.environment, this.aftEnvironment, this.latitude, this.longitude, this.partner)) {
+            this.consumer = new CambriaConsumerWrapper(this.servers, this.topic, this.apiKey, this.apiSecret,
+                    this.userName, this.password, this.consumerGroup, this.consumerInstance, this.fetchTimeout,
+                    this.fetchLimit, this.useHttps, this.allowSelfSignedCerts);
+        } else {
+            this.consumer = new DmaapDmeConsumerWrapper(this.servers, this.topic, this.apiKey,
+                    this.apiSecret, this.userName, this.password, this.consumerGroup, this.consumerInstance,
+                    this.fetchTimeout, this.fetchLimit, this.environment, this.aftEnvironment, this.partner,
+                    this.latitude, this.longitude, this.additionalProps, this.useHttps);
+        }
+
+        logger.info("{}: INITTED", this);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public CommInfrastructure getTopicCommInfrastructure() {
+        return Topic.CommInfrastructure.DMAAP;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder builder = new StringBuilder();
+        builder.append("SingleThreadedDmaapTopicSource [userName=").append(userName).append(", password=")
+                .append((password == null || password.isEmpty()) ? "-" : password.length())
+                .append(", getTopicCommInfrastructure()=").append(getTopicCommInfrastructure()).append(", toString()=")
+                .append(super.toString()).append("]");
+        return builder.toString();
+    }
+
+
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSource.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSource.java
new file mode 100644
index 00000000..00e45422
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSource.java
@@ -0,0 +1,91 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+import java.util.List;
+
+import org.onap.policy.common.endpoints.event.comm.Topic;
+import org.onap.policy.common.endpoints.event.comm.bus.UebTopicSource;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.impl.CambriaConsumerWrapper;
+
+/**
+ * This topic source implementation specializes in reading messages over an UEB Bus topic source and
+ * notifying its listeners
+ */
+public class SingleThreadedUebTopicSource extends SingleThreadedBusTopicSource implements UebTopicSource {
+
+    /**
+     * 
+     * @param servers UEB servers
+     * @param topic UEB Topic to be monitored
+     * @param apiKey UEB API Key (optional)
+     * @param apiSecret UEB API Secret (optional)
+     * @param consumerGroup UEB Reader Consumer Group
+     * @param consumerInstance UEB Reader Instance
+     * @param fetchTimeout UEB fetch timeout
+     * @param fetchLimit UEB fetch limit
+     * @param useHttps does topicSource use HTTPS?
+     * @param allowSelfSignedCerts does topicSource allow self-signed certs?
+     * 
+     * @throws IllegalArgumentException An invalid parameter passed in
+     */
+
+
+    public SingleThreadedUebTopicSource(List<String> servers, String topic, String apiKey, String apiSecret,
+            String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit, boolean useHttps,
+            boolean allowSelfSignedCerts) {
+
+        super(servers, topic, apiKey, apiSecret, consumerGroup, consumerInstance, fetchTimeout, fetchLimit, useHttps,
+                allowSelfSignedCerts);
+
+        this.allowSelfSignedCerts = allowSelfSignedCerts;
+
+        this.init();
+    }
+
+    /**
+     * Initialize the Cambria client
+     */
+    @Override
+    public void init() {
+        this.consumer = new CambriaConsumerWrapper(this.servers, this.topic, this.apiKey, this.apiSecret,
+                this.consumerGroup, this.consumerInstance, this.fetchTimeout, this.fetchLimit, this.useHttps,
+                this.allowSelfSignedCerts);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public CommInfrastructure getTopicCommInfrastructure() {
+        return Topic.CommInfrastructure.UEB;
+    }
+
+
+    @Override
+    public String toString() {
+        StringBuilder builder = new StringBuilder();
+        builder.append("SingleThreadedUebTopicSource [getTopicCommInfrastructure()=")
+                .append(getTopicCommInfrastructure()).append(", toString()=").append(super.toString()).append("]");
+        return builder.toString();
+    }
+
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBase.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBase.java
new file mode 100644
index 00000000..ed15ddf7
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBase.java
@@ -0,0 +1,229 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.collections4.queue.CircularFifoQueue;
+import org.onap.policy.common.endpoints.event.comm.Topic;
+import org.onap.policy.common.endpoints.event.comm.TopicListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class TopicBase implements Topic {
+
+    /**
+     * logger
+     */
+    private static Logger logger = LoggerFactory.getLogger(TopicBase.class);
+
+    /**
+     * list of servers
+     */
+    protected List<String> servers;
+
+    /**
+     * Topic
+     */
+    protected String topic;
+
+    /**
+     * event cache
+     */
+    protected CircularFifoQueue<String> recentEvents = new CircularFifoQueue<>(10);
+
+    /**
+     * Am I running? reflects invocation of start()/stop() !locked & start() => alive stop() =>
+     * !alive
+     */
+    protected volatile boolean alive = false;
+
+    /**
+     * Am I locked? reflects invocation of lock()/unlock() operations locked => !alive (but not in
+     * the other direction necessarily) locked => !offer, !run, !start, !stop (but this last one is
+     * obvious since locked => !alive)
+     */
+    protected volatile boolean locked = false;
+
+    /**
+     * All my subscribers for new message notifications
+     */
+    protected final ArrayList<TopicListener> topicListeners = new ArrayList<>();
+
+    /**
+     * Instantiates a new Topic Base
+     * 
+     * @param servers list of servers
+     * @param topic topic name
+     * 
+     * @return a Topic Base
+     * @throws IllegalArgumentException if invalid parameters are present
+     */
+    public TopicBase(List<String> servers, String topic) {
+
+        if (servers == null || servers.isEmpty()) {
+            throw new IllegalArgumentException("Server(s) must be provided");
+        }
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException("A Topic must be provided");
+        }
+
+        this.servers = servers;
+        this.topic = topic;
+    }
+
+    @Override
+    public void register(TopicListener topicListener) {
+
+        logger.info("{}: registering {}", this, topicListener);
+
+        synchronized (this) {
+            if (topicListener == null) {
+                throw new IllegalArgumentException("TopicListener must be provided");
+            }
+
+            for (TopicListener listener : this.topicListeners) {
+                if (listener == topicListener) {
+                    return;
+                }
+            }
+
+            this.topicListeners.add(topicListener);
+        }
+    }
+
+    @Override
+    public void unregister(TopicListener topicListener) {
+
+        logger.info("{}: unregistering {}", this, topicListener);
+
+        synchronized (this) {
+            if (topicListener == null) {
+                throw new IllegalArgumentException("TopicListener must be provided");
+            }
+
+            this.topicListeners.remove(topicListener);
+        }
+    }
+
+    /**
+     * broadcast event to all listeners
+     * 
+     * @param message the event
+     * @return true if all notifications are performed with no error, false otherwise
+     */
+    protected boolean broadcast(String message) {
+        List<TopicListener> snapshotListeners = this.snapshotTopicListeners();
+
+        boolean success = true;
+        for (TopicListener topicListener : snapshotListeners) {
+            try {
+                topicListener.onTopicEvent(this.getTopicCommInfrastructure(), this.topic, message);
+            } catch (Exception e) {
+                logger.warn("{}: notification error @ {} because of {}", this, topicListener, e.getMessage(), e);
+                success = false;
+            }
+        }
+        return success;
+    }
+
+    /**
+     * take a snapshot of current topic listeners
+     * 
+     * @return the topic listeners
+     */
+    protected synchronized List<TopicListener> snapshotTopicListeners() {
+        @SuppressWarnings("unchecked")
+        List<TopicListener> listeners = (List<TopicListener>) topicListeners.clone();
+        return listeners;
+    }
+
+    @Override
+    public boolean lock() {
+
+        logger.info("{}: locking", this);
+
+        synchronized (this) {
+            if (this.locked) {
+                return true;
+            }
+
+            this.locked = true;
+        }
+
+        return this.stop();
+    }
+
+    @Override
+    public boolean unlock() {
+        logger.info("{}: unlocking", this);
+
+        synchronized (this) {
+            if (!this.locked) {
+                return true;
+            }
+
+            this.locked = false;
+        }
+
+        try {
+            return this.start();
+        } catch (Exception e) {
+            logger.warn("{}: cannot after unlocking because of {}", this, e.getMessage(), e);
+            return false;
+        }
+    }
+
+    @Override
+    public boolean isLocked() {
+        return this.locked;
+    }
+
+    @Override
+    public String getTopic() {
+        return topic;
+    }
+
+    @Override
+    public boolean isAlive() {
+        return this.alive;
+    }
+
+    @Override
+    public List<String> getServers() {
+        return servers;
+    }
+
+    @Override
+    public synchronized String[] getRecentEvents() {
+        String[] events = new String[recentEvents.size()];
+        return recentEvents.toArray(events);
+    }
+
+
+    @Override
+    public String toString() {
+        return "TopicBase [servers=" + servers + ", topic=" + topic + ", #recentEvents=" + recentEvents.size()
+                + ", locked=" + locked + ", #topicListeners=" + topicListeners.size() + "]";
+    }
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/CambriaConsumerWrapper.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/CambriaConsumerWrapper.java
new file mode 100644
index 00000000..fedde284
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/CambriaConsumerWrapper.java
@@ -0,0 +1,210 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal.impl;
+
+import com.att.nsa.cambria.client.CambriaClientBuilders;
+import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder;
+import com.att.nsa.cambria.client.CambriaConsumer;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.security.GeneralSecurityException;
+import java.util.List;
+
+import org.onap.policy.common.endpoints.event.comm.bus.internal.FilterableBusConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Cambria based consumer
+ */
+public class CambriaConsumerWrapper implements FilterableBusConsumer {
+
+    /**
+     * logger
+     */
+    private static Logger logger = LoggerFactory.getLogger(CambriaConsumerWrapper.class);
+
+    /**
+     * Used to build the consumer.
+     */
+    private final ConsumerBuilder builder;
+
+    /**
+     * Locked while updating {@link #consumer} and {@link #newConsumer}.
+     */
+    private final Object consLocker = new Object();
+
+    /**
+     * Cambria client
+     */
+    private CambriaConsumer consumer;
+
+    /**
+     * Cambria client to use for next fetch
+     */
+    private CambriaConsumer newConsumer = null;
+
+    /**
+     * fetch timeout
+     */
+    protected int fetchTimeout;
+
+    /**
+     * close condition
+     */
+    protected Object closeCondition = new Object();
+
+    /**
+     * Cambria Consumer Wrapper
+     *
+     * @param servers messaging bus hosts
+     * @param topic topic
+     * @param apiKey API Key
+     * @param apiSecret API Secret
+     * @param consumerGroup Consumer Group
+     * @param consumerInstance Consumer Instance
+     * @param fetchTimeout Fetch Timeout
+     * @param fetchLimit Fetch Limit
+     * @throws GeneralSecurityException
+     * @throws MalformedURLException
+     */
+    public CambriaConsumerWrapper(List<String> servers, String topic, String apiKey, String apiSecret,
+            String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit, boolean useHttps,
+            boolean useSelfSignedCerts) {
+        this(servers, topic, apiKey, apiSecret, null, null, consumerGroup, consumerInstance, fetchTimeout, fetchLimit,
+                useHttps, useSelfSignedCerts);
+    }
+
+    public CambriaConsumerWrapper(List<String> servers, String topic, String apiKey, String apiSecret, String username,
+            String password, String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit,
+            boolean useHttps, boolean useSelfSignedCerts) {
+
+        this.fetchTimeout = fetchTimeout;
+
+        this.builder = new CambriaClientBuilders.ConsumerBuilder();
+
+        builder.knownAs(consumerGroup, consumerInstance).usingHosts(servers).onTopic(topic).waitAtServer(fetchTimeout)
+                .receivingAtMost(fetchLimit);
+
+        // Set read timeout to fetch timeout + 30 seconds (TBD: this should be configurable)
+        builder.withSocketTimeout(fetchTimeout + 30000);
+
+        if (useHttps) {
+            builder.usingHttps();
+
+            if (useSelfSignedCerts) {
+                builder.allowSelfSignedCertificates();
+            }
+        }
+
+        if (apiKey != null && !apiKey.isEmpty() && apiSecret != null && !apiSecret.isEmpty()) {
+            builder.authenticatedBy(apiKey, apiSecret);
+        }
+
+        if (username != null && !username.isEmpty() && password != null && !password.isEmpty()) {
+            builder.authenticatedByHttp(username, password);
+        }
+
+        try {
+            this.consumer = builder.build();
+        } catch (MalformedURLException | GeneralSecurityException e) {
+            throw new IllegalArgumentException(e);
+        }
+    }
+
+    @Override
+    public Iterable<String> fetch() throws IOException, InterruptedException {
+        try {
+            return getCurrentConsumer().fetch();
+        } catch (final IOException e) {
+            logger.error("{}: cannot fetch because of {} - backoff for {} ms.", this, e.getMessage(),
+                    this.fetchTimeout);
+            synchronized (this.closeCondition) {
+                this.closeCondition.wait(this.fetchTimeout);
+            }
+
+            throw e;
+        }
+    }
+
+    @Override
+    public void close() {
+        synchronized (closeCondition) {
+            closeCondition.notifyAll();
+        }
+
+        getCurrentConsumer().close();
+    }
+
+    private CambriaConsumer getCurrentConsumer() {
+        CambriaConsumer old = null;
+        CambriaConsumer ret;
+
+        synchronized (consLocker) {
+            if (this.newConsumer != null) {
+                // replace old consumer with new consumer
+                old = this.consumer;
+                this.consumer = this.newConsumer;
+                this.newConsumer = null;
+            }
+
+            ret = this.consumer;
+        }
+
+        if (old != null) {
+            old.close();
+        }
+
+        return ret;
+    }
+
+    @Override
+    public void setFilter(String filter) {
+        logger.info("{}: setting DMAAP server-side filter: {}", this, filter);
+        builder.withServerSideFilter(filter);
+
+        try {
+            CambriaConsumer previous;
+            synchronized (consLocker) {
+                previous = this.newConsumer;
+                this.newConsumer = builder.build();
+            }
+
+            if (previous != null) {
+                // there was already a new consumer - close it
+                previous.close();
+            }
+
+        } catch (MalformedURLException | GeneralSecurityException e) {
+            /*
+             * Since an exception occurred, "consumer" still has its old value, thus it should not
+             * be closed at this point.
+             */
+            throw new IllegalArgumentException(e);
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "CambriaConsumerWrapper [fetchTimeout=" + fetchTimeout + "]";
+    }
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/CambriaPublisherWrapper.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/CambriaPublisherWrapper.java
new file mode 100644
index 00000000..ab5868ab
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/CambriaPublisherWrapper.java
@@ -0,0 +1,122 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal.impl;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import com.att.nsa.cambria.client.CambriaClientBuilders;
+import com.att.nsa.cambria.client.CambriaClientBuilders.PublisherBuilder;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
+import java.net.MalformedURLException;
+import java.security.GeneralSecurityException;
+import java.util.List;
+
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusPublisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Cambria based library publisher
+ */
+public class CambriaPublisherWrapper implements BusPublisher {
+
+    private static Logger logger = LoggerFactory.getLogger(CambriaPublisherWrapper.class);
+
+    /**
+     * The actual Cambria publisher
+     */
+    @JsonIgnore
+    protected volatile CambriaBatchingPublisher publisher;
+
+    public CambriaPublisherWrapper(List<String> servers, String topic, String apiKey, String apiSecret,
+            boolean useHttps) {
+        this(servers, topic, apiKey, apiSecret, null, null, useHttps);
+    }
+
+    public CambriaPublisherWrapper(List<String> servers, String topic, String apiKey, String apiSecret, String username,
+            String password, boolean useHttps) {
+
+        PublisherBuilder builder = new CambriaClientBuilders.PublisherBuilder();
+
+        builder.usingHosts(servers).onTopic(topic);
+
+        // Set read timeout to 30 seconds (TBD: this should be configurable)
+        builder.withSocketTimeout(30000);
+
+        if (useHttps) {
+            builder.usingHttps();
+        }
+
+
+        if (apiKey != null && !apiKey.isEmpty() && apiSecret != null && !apiSecret.isEmpty()) {
+            builder.authenticatedBy(apiKey, apiSecret);
+        }
+
+        if (username != null && !username.isEmpty() && password != null && !password.isEmpty()) {
+            builder.authenticatedByHttp(username, password);
+        }
+
+        try {
+            this.publisher = builder.build();
+        } catch (MalformedURLException | GeneralSecurityException e) {
+            throw new IllegalArgumentException(e);
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean send(String partitionId, String message) {
+        if (message == null) {
+            throw new IllegalArgumentException("No message provided");
+        }
+
+        try {
+            this.publisher.send(partitionId, message);
+        } catch (Exception e) {
+            logger.warn("{}: SEND of {} cannot be performed because of {}", this, message, e.getMessage(), e);
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void close() {
+        logger.info("{}: CLOSE", this);
+
+        try {
+            this.publisher.close();
+        } catch (Exception e) {
+            logger.warn("{}: CLOSE FAILED because of {}", this, e.getMessage(), e);
+        }
+    }
+
+
+    @Override
+    public String toString() {
+        return "CambriaPublisherWrapper []";
+    }
+
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapAafConsumerWrapper.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapAafConsumerWrapper.java
new file mode 100644
index 00000000..29fbf1d5
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapAafConsumerWrapper.java
@@ -0,0 +1,95 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal.impl;
+
+import com.att.nsa.mr.client.impl.MRConsumerImpl;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+import java.net.MalformedURLException;
+import java.util.List;
+import java.util.Properties;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MR based consumer
+ */
+public class DmaapAafConsumerWrapper extends DmaapConsumerWrapper {
+
+    private static Logger logger = LoggerFactory.getLogger(DmaapAafConsumerWrapper.class);
+
+    private final Properties props;
+
+    /**
+     * MR Consumer Wrapper
+     *
+     * @param servers messaging bus hosts
+     * @param topic topic
+     * @param apiKey API Key
+     * @param apiSecret API Secret
+     * @param aafLogin AAF Login
+     * @param aafPassword AAF Password
+     * @param consumerGroup Consumer Group
+     * @param consumerInstance Consumer Instance
+     * @param fetchTimeout Fetch Timeout
+     * @param fetchLimit Fetch Limit
+     * @throws MalformedURLException
+     */
+    public DmaapAafConsumerWrapper(List<String> servers, String topic, String apiKey, String apiSecret, String aafLogin,
+            String aafPassword, String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit,
+            boolean useHttps) throws MalformedURLException {
+
+        super(servers, topic, apiKey, apiSecret, aafLogin, aafPassword, consumerGroup, consumerInstance, fetchTimeout,
+                fetchLimit);
+
+        // super constructor sets servers = {""} if empty to avoid errors when using DME2
+        if ((servers.size() == 1 && ("".equals(servers.get(0)))) || (servers == null) || (servers.isEmpty())) {
+            throw new IllegalArgumentException("Must provide at least one host for HTTP AAF");
+        }
+
+        this.consumer.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+
+        props = new Properties();
+
+        if (useHttps) {
+            props.setProperty(PROTOCOL_PROP, "https");
+            this.consumer.setHost(servers.get(0) + ":3905");
+
+        } else {
+            props.setProperty(PROTOCOL_PROP, "http");
+            this.consumer.setHost(servers.get(0) + ":3904");
+        }
+
+        this.consumer.setProps(props);
+        logger.info("{}: CREATION", this);
+    }
+
+    @Override
+    public String toString() {
+        final MRConsumerImpl consumer = this.consumer;
+
+        return "DmaapConsumerWrapper [" + "consumer.getAuthDate()=" + consumer.getAuthDate()
+                + ", consumer.getAuthKey()=" + consumer.getAuthKey() + ", consumer.getHost()=" + consumer.getHost()
+                + ", consumer.getProtocolFlag()=" + consumer.getProtocolFlag() + ", consumer.getUsername()="
+                + consumer.getUsername() + "]";
+    }
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapAafPublisherWrapper.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapAafPublisherWrapper.java
new file mode 100644
index 00000000..1308bb36
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapAafPublisherWrapper.java
@@ -0,0 +1,39 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal.impl;
+
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+import java.util.List;
+
+/**
+ * DmaapClient library wrapper
+ */
+public class DmaapAafPublisherWrapper extends DmaapPublisherWrapper {
+    /**
+     * MR based Publisher
+     */
+    public DmaapAafPublisherWrapper(List<String> servers, String topic, String aafLogin, String aafPassword,
+            boolean useHttps) {
+
+        super(ProtocolTypeConstants.AAF_AUTH, servers, topic, aafLogin, aafPassword, useHttps);
+    }
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapConsumerWrapper.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapConsumerWrapper.java
new file mode 100644
index 00000000..0806c3d3
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapConsumerWrapper.java
@@ -0,0 +1,146 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal.impl;
+
+import com.att.nsa.mr.client.impl.MRConsumerImpl;
+import com.att.nsa.mr.client.response.MRConsumerResponse;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MR based consumer
+ */
+public abstract class DmaapConsumerWrapper implements BusConsumer {
+
+    /**
+     * logger
+     */
+    private static Logger logger = LoggerFactory.getLogger(DmaapConsumerWrapper.class);
+
+    /**
+     * Name of the "protocol" property.
+     */
+    protected static final String PROTOCOL_PROP = "Protocol";
+
+    /**
+     * fetch timeout
+     */
+    protected int fetchTimeout;
+
+    /**
+     * close condition
+     */
+    protected Object closeCondition = new Object();
+
+    /**
+     * MR Consumer
+     */
+    protected MRConsumerImpl consumer;
+
+    /**
+     * MR Consumer Wrapper
+     *
+     * @param servers messaging bus hosts
+     * @param topic topic
+     * @param apiKey API Key
+     * @param apiSecret API Secret
+     * @param username AAF Login
+     * @param password AAF Password
+     * @param consumerGroup Consumer Group
+     * @param consumerInstance Consumer Instance
+     * @param fetchTimeout Fetch Timeout
+     * @param fetchLimit Fetch Limit
+     * @throws MalformedURLException
+     */
+    public DmaapConsumerWrapper(List<String> servers, String topic, String apiKey, String apiSecret, String username,
+            String password, String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit)
+            throws MalformedURLException {
+
+        this.fetchTimeout = fetchTimeout;
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException("No topic for DMaaP");
+        }
+
+        this.consumer = new MRConsumerImpl(servers, topic, consumerGroup, consumerInstance, fetchTimeout, fetchLimit,
+                null, apiKey, apiSecret);
+
+        this.consumer.setUsername(username);
+        this.consumer.setPassword(password);
+    }
+
+    @Override
+    public Iterable<String> fetch() throws InterruptedException, IOException {
+        final MRConsumerResponse response = this.consumer.fetchWithReturnConsumerResponse();
+        if (response == null) {
+            logger.warn("{}: DMaaP NULL response received", this);
+
+            synchronized (closeCondition) {
+                closeCondition.wait(fetchTimeout);
+            }
+            return new ArrayList<>();
+        } else {
+            logger.debug("DMaaP consumer received {} : {}" + response.getResponseCode(), response.getResponseMessage());
+
+            if (response.getResponseCode() == null || !"200".equals(response.getResponseCode())) {
+
+                logger.error("DMaaP consumer received: {} : {}", response.getResponseCode(),
+                        response.getResponseMessage());
+
+                synchronized (closeCondition) {
+                    closeCondition.wait(fetchTimeout);
+                }
+
+                /* fall through */
+            }
+        }
+
+        if (response.getActualMessages() == null) {
+            return new ArrayList<>();
+        } else {
+            return response.getActualMessages();
+        }
+    }
+
+    @Override
+    public void close() {
+        synchronized (closeCondition) {
+            closeCondition.notifyAll();
+        }
+
+        this.consumer.close();
+    }
+
+    @Override
+    public String toString() {
+        return "DmaapConsumerWrapper [" + "consumer.getAuthDate()=" + consumer.getAuthDate()
+                + ", consumer.getAuthKey()=" + consumer.getAuthKey() + ", consumer.getHost()=" + consumer.getHost()
+                + ", consumer.getProtocolFlag()=" + consumer.getProtocolFlag() + ", consumer.getUsername()="
+                + consumer.getUsername() + "]";
+    }
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapDmeConsumerWrapper.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapDmeConsumerWrapper.java
new file mode 100644
index 00000000..fd998e2b
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapDmeConsumerWrapper.java
@@ -0,0 +1,144 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal.impl;
+
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+import java.net.MalformedURLException;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicSinkFactory;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DmaapDmeConsumerWrapper extends DmaapConsumerWrapper {
+
+    private static Logger logger = LoggerFactory.getLogger(DmaapDmeConsumerWrapper.class);
+
+    private final Properties props;
+
+    public DmaapDmeConsumerWrapper(List<String> servers, String topic, String apiKey, String apiSecret,
+            String dme2Login, String dme2Password, String consumerGroup, String consumerInstance, int fetchTimeout,
+            int fetchLimit, String environment, String aftEnvironment, String dme2Partner, String latitude,
+            String longitude, Map<String, String> additionalProps, boolean useHttps) throws MalformedURLException {
+
+
+
+        super(servers, topic, apiKey, apiSecret, dme2Login, dme2Password, consumerGroup, consumerInstance, fetchTimeout,
+                fetchLimit);
+
+
+        final String dme2RouteOffer = additionalProps.get(DmaapTopicSinkFactory.DME2_ROUTE_OFFER_PROPERTY);
+
+        if (environment == null || environment.isEmpty()) {
+            throw parmException(topic, PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
+        }
+        if (aftEnvironment == null || aftEnvironment.isEmpty()) {
+            throw parmException(topic, PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
+        }
+        if (latitude == null || latitude.isEmpty()) {
+            throw parmException(topic, PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
+        }
+        if (longitude == null || longitude.isEmpty()) {
+            throw parmException(topic, PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
+        }
+
+        if ((dme2Partner == null || dme2Partner.isEmpty()) && (dme2RouteOffer == null || dme2RouteOffer.isEmpty())) {
+            throw new IllegalArgumentException(
+                    "Must provide at least " + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
+                            + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX + " or "
+                            + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
+                            + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX + " for DME2");
+        }
+
+        final String serviceName = servers.get(0);
+
+        this.consumer.setProtocolFlag(ProtocolTypeConstants.DME2.getValue());
+
+        this.consumer.setUsername(dme2Login);
+        this.consumer.setPassword(dme2Password);
+
+        props = new Properties();
+
+        props.setProperty(DmaapTopicSinkFactory.DME2_SERVICE_NAME_PROPERTY, serviceName);
+
+        props.setProperty("username", dme2Login);
+        props.setProperty("password", dme2Password);
+
+        /* These are required, no defaults */
+        props.setProperty("topic", topic);
+
+        props.setProperty("Environment", environment);
+        props.setProperty("AFT_ENVIRONMENT", aftEnvironment);
+
+        if (dme2Partner != null) {
+            props.setProperty("Partner", dme2Partner);
+        }
+        if (dme2RouteOffer != null) {
+            props.setProperty(DmaapTopicSinkFactory.DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
+        }
+
+        props.setProperty("Latitude", latitude);
+        props.setProperty("Longitude", longitude);
+
+        /* These are optional, will default to these values if not set in additionalProps */
+        props.setProperty("AFT_DME2_EP_READ_TIMEOUT_MS", "50000");
+        props.setProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", "240000");
+        props.setProperty("AFT_DME2_EP_CONN_TIMEOUT", "15000");
+        props.setProperty("Version", "1.0");
+        props.setProperty("SubContextPath", "/");
+        props.setProperty("sessionstickinessrequired", "no");
+
+        /* These should not change */
+        props.setProperty("TransportType", "DME2");
+        props.setProperty("MethodType", "GET");
+
+        if (useHttps) {
+            props.setProperty(PROTOCOL_PROP, "https");
+
+        } else {
+            props.setProperty(PROTOCOL_PROP, "http");
+        }
+
+        props.setProperty("contenttype", "application/json");
+
+        if (additionalProps != null) {
+            for (Map.Entry<String, String> entry : additionalProps.entrySet()) {
+                props.put(entry.getKey(), entry.getValue());
+            }
+        }
+
+        MRClientFactory.prop = props;
+        this.consumer.setProps(props);
+
+        logger.info("{}: CREATION", this);
+    }
+
+    private IllegalArgumentException parmException(String topic, String propnm) {
+        return new IllegalArgumentException("Missing " + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+                + topic + propnm + " property for DME2 in DMaaP");
+
+    }
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapDmePublisherWrapper.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapDmePublisherWrapper.java
new file mode 100644
index 00000000..e39954ed
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapDmePublisherWrapper.java
@@ -0,0 +1,112 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal.impl;
+
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+import java.util.List;
+import java.util.Map;
+
+import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicSinkFactory;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
+
+public class DmaapDmePublisherWrapper extends DmaapPublisherWrapper {
+    public DmaapDmePublisherWrapper(List<String> servers, String topic, String username, String password,
+            String environment, String aftEnvironment, String dme2Partner, String latitude, String longitude,
+            Map<String, String> additionalProps, boolean useHttps) {
+
+        super(ProtocolTypeConstants.DME2, servers, topic, username, password, useHttps);
+
+
+
+        String dme2RouteOffer = additionalProps.get(DmaapTopicSinkFactory.DME2_ROUTE_OFFER_PROPERTY);
+
+        if (environment == null || environment.isEmpty()) {
+            throw parmException(topic, PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
+        }
+        if (aftEnvironment == null || aftEnvironment.isEmpty()) {
+            throw parmException(topic, PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
+        }
+        if (latitude == null || latitude.isEmpty()) {
+            throw parmException(topic, PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
+        }
+        if (longitude == null || longitude.isEmpty()) {
+            throw parmException(topic, PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
+        }
+
+        if ((dme2Partner == null || dme2Partner.isEmpty()) && (dme2RouteOffer == null || dme2RouteOffer.isEmpty())) {
+            throw new IllegalArgumentException(
+                    "Must provide at least " + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
+                            + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX + " or "
+                            + PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "." + topic
+                            + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX + " for DME2");
+        }
+
+        String serviceName = servers.get(0);
+
+        /* These are required, no defaults */
+        props.setProperty("Environment", environment);
+        props.setProperty("AFT_ENVIRONMENT", aftEnvironment);
+
+        props.setProperty(DmaapTopicSinkFactory.DME2_SERVICE_NAME_PROPERTY, serviceName);
+
+        if (dme2Partner != null) {
+            props.setProperty("Partner", dme2Partner);
+        }
+        if (dme2RouteOffer != null) {
+            props.setProperty(DmaapTopicSinkFactory.DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
+        }
+
+        props.setProperty("Latitude", latitude);
+        props.setProperty("Longitude", longitude);
+
+        // ServiceName also a default, found in additionalProps
+
+        /* These are optional, will default to these values if not set in optionalProps */
+        props.setProperty("AFT_DME2_EP_READ_TIMEOUT_MS", "50000");
+        props.setProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", "240000");
+        props.setProperty("AFT_DME2_EP_CONN_TIMEOUT", "15000");
+        props.setProperty("Version", "1.0");
+        props.setProperty("SubContextPath", "/");
+        props.setProperty("sessionstickinessrequired", "no");
+
+        /* These should not change */
+        props.setProperty("TransportType", "DME2");
+        props.setProperty("MethodType", "POST");
+
+        for (Map.Entry<String, String> entry : additionalProps.entrySet()) {
+            String key = entry.getKey();
+            String value = entry.getValue();
+
+            if (value != null) {
+                props.setProperty(key, value);
+            }
+        }
+
+        this.publisher.setProps(props);
+    }
+
+    private IllegalArgumentException parmException(String topic, String propnm) {
+        return new IllegalArgumentException("Missing " + PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
+                + topic + propnm + " property for DME2 in DMaaP");
+
+    }
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapPublisherWrapper.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapPublisherWrapper.java
new file mode 100644
index 00000000..396580e9
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/impl/DmaapPublisherWrapper.java
@@ -0,0 +1,164 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.policy.common.endpoints.event.comm.bus.internal.impl;
+
+import com.att.nsa.mr.client.impl.MRSimplerBatchPublisher;
+import com.att.nsa.mr.client.response.MRPublisherResponse;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusPublisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * DmaapClient library wrapper
+ */
+public abstract class DmaapPublisherWrapper implements BusPublisher {
+
+    private static Logger logger = LoggerFactory.getLogger(DmaapPublisherWrapper.class);
+
+    /**
+     * MR based Publisher
+     */
+    protected MRSimplerBatchPublisher publisher;
+    protected Properties props;
+
+    /**
+     * MR Publisher Wrapper
+     *
+     * @param servers messaging bus hosts
+     * @param topic topic
+     * @param username AAF or DME2 Login
+     * @param password AAF or DME2 Password
+     */
+    public DmaapPublisherWrapper(ProtocolTypeConstants protocol, List<String> servers, String topic, String username,
+            String password, boolean useHttps) {
+
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException("No topic for DMaaP");
+        }
+
+
+        if (protocol == ProtocolTypeConstants.AAF_AUTH) {
+            if (servers == null || servers.isEmpty()) {
+                throw new IllegalArgumentException("No DMaaP servers or DME2 partner provided");
+            }
+
+            ArrayList<String> dmaapServers = new ArrayList<>();
+            if (useHttps) {
+                for (String server : servers) {
+                    dmaapServers.add(server + ":3905");
+                }
+
+            } else {
+                for (String server : servers) {
+                    dmaapServers.add(server + ":3904");
+                }
+            }
+
+
+            this.publisher = new MRSimplerBatchPublisher.Builder().againstUrls(dmaapServers).onTopic(topic).build();
+
+            this.publisher.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+        } else if (protocol == ProtocolTypeConstants.DME2) {
+            ArrayList<String> dmaapServers = new ArrayList<>();
+            dmaapServers.add("0.0.0.0:3904");
+
+            this.publisher = new MRSimplerBatchPublisher.Builder().againstUrls(dmaapServers).onTopic(topic).build();
+
+            this.publisher.setProtocolFlag(ProtocolTypeConstants.DME2.getValue());
+        }
+
+        this.publisher.logTo(LoggerFactory.getLogger(MRSimplerBatchPublisher.class.getName()));
+
+        this.publisher.setUsername(username);
+        this.publisher.setPassword(password);
+
+        props = new Properties();
+
+        if (useHttps) {
+            props.setProperty("Protocol", "https");
+        } else {
+            props.setProperty("Protocol", "http");
+        }
+
+        props.setProperty("contenttype", "application/json");
+        props.setProperty("username", username);
+        props.setProperty("password", password);
+
+        props.setProperty("topic", topic);
+
+        this.publisher.setProps(props);
+
+        if (protocol == ProtocolTypeConstants.AAF_AUTH) {
+            this.publisher.setHost(servers.get(0));
+        }
+
+        logger.info("{}: CREATION: using protocol {}", this, protocol.getValue());
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void close() {
+        logger.info("{}: CLOSE", this);
+
+        try {
+            this.publisher.close(1, TimeUnit.SECONDS);
+        } catch (Exception e) {
+            logger.warn("{}: CLOSE FAILED because of {}", this, e.getMessage(), e);
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean send(String partitionId, String message) {
+        if (message == null) {
+            throw new IllegalArgumentException("No message provided");
+        }
+
+        this.publisher.setPubResponse(new MRPublisherResponse());
+        this.publisher.send(partitionId, message);
+        MRPublisherResponse response = this.publisher.sendBatchWithResponse();
+        if (response != null) {
+            logger.debug("DMaaP publisher received {} : {}", response.getResponseCode(), response.getResponseMessage());
+        }
+
+        return true;
+    }
+
+    @Override
+    public String toString() {
+        return "DmaapPublisherWrapper [" + "publisher.getAuthDate()=" + publisher.getAuthDate()
+                + ", publisher.getAuthKey()=" + publisher.getAuthKey() + ", publisher.getHost()=" + publisher.getHost()
+                + ", publisher.getProtocolFlag()=" + publisher.getProtocolFlag() + ", publisher.getUsername()="
+                + publisher.getUsername() + "]";
+    }
+}