aboutsummaryrefslogtreecommitdiffstats
path: root/policy-endpoints/src/main/java/org/openecomp/policy/drools/event/comm/bus/internal/SingleThreadedBusTopicSource.java
diff options
context:
space:
mode:
Diffstat (limited to 'policy-endpoints/src/main/java/org/openecomp/policy/drools/event/comm/bus/internal/SingleThreadedBusTopicSource.java')
-rw-r--r--policy-endpoints/src/main/java/org/openecomp/policy/drools/event/comm/bus/internal/SingleThreadedBusTopicSource.java477
1 files changed, 477 insertions, 0 deletions
diff --git a/policy-endpoints/src/main/java/org/openecomp/policy/drools/event/comm/bus/internal/SingleThreadedBusTopicSource.java b/policy-endpoints/src/main/java/org/openecomp/policy/drools/event/comm/bus/internal/SingleThreadedBusTopicSource.java
new file mode 100644
index 00000000..f37c349e
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/openecomp/policy/drools/event/comm/bus/internal/SingleThreadedBusTopicSource.java
@@ -0,0 +1,477 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * Copyright (C) 2017 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.openecomp.policy.drools.event.comm.bus.internal;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.log4j.Logger;
+
+import org.openecomp.policy.drools.event.comm.TopicListener;
+import org.openecomp.policy.drools.event.comm.bus.BusTopicSource;
+import org.openecomp.policy.common.logging.eelf.MessageCodes;
+import org.openecomp.policy.common.logging.eelf.PolicyLogger;
+
+/**
+ * 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 {
+
+ private String className = SingleThreadedBusTopicSource.class.getName();
+ /**
+ * Not to be converted to PolicyLogger.
+ * This will contain all instract /out traffic and only that in a single file in a concise format.
+ */
+ protected static final Logger networkLogger = Logger.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;
+
+ /**
+ * 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;
+
+ /**
+ * Independent thread reading message over my topic
+ */
+ protected Thread busPollerThread;
+
+ /**
+ * All my subscribers for new message notifications
+ */
+ protected final ArrayList<TopicListener> topicListeners = new ArrayList<TopicListener>();
+
+ /**
+ *
+ * @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
+ * @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)
+ throws IllegalArgumentException {
+
+ super(servers, topic, apiKey, apiSecret);
+
+ if (consumerGroup == null || consumerGroup.isEmpty()) {
+ this.consumerGroup = UUID.randomUUID ().toString();
+ } else {
+ this.consumerGroup = consumerGroup;
+ }
+
+ if (consumerInstance == null || consumerInstance.isEmpty()) {
+ this.consumerInstance = DEFAULT_CONSUMER_INSTANCE;
+ } 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 Exception;
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void register(TopicListener topicListener)
+ throws IllegalArgumentException {
+
+ PolicyLogger.info(className,"REGISTER: " + topicListener + " INTO " + this);
+
+ synchronized(this) {
+ if (topicListener == null)
+ throw new IllegalArgumentException("TopicListener must be provided");
+
+ /* check that this listener is not registered already */
+ for (TopicListener listener: this.topicListeners) {
+ if (listener == topicListener) {
+ // already registered
+ return;
+ }
+ }
+
+ this.topicListeners.add(topicListener);
+ }
+
+ try {
+ this.start();
+ } catch (Exception e) {
+ PolicyLogger.info(className, "new registration of " + topicListener +
+ ",but can't start source because of " + e.getMessage());
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void unregister(TopicListener topicListener) {
+
+ PolicyLogger.info(className, "UNREGISTER: " + topicListener + " FROM " + this);
+
+ boolean stop = false;
+ synchronized (this) {
+ if (topicListener == null)
+ throw new IllegalArgumentException("TopicListener must be provided");
+
+ this.topicListeners.remove(topicListener);
+ stop = (this.topicListeners.isEmpty());
+ }
+
+ if (stop) {
+ this.stop();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean lock() {
+ PolicyLogger.info(className, "LOCK: " + this);
+
+ synchronized (this) {
+ if (this.locked)
+ return true;
+
+ this.locked = true;
+ }
+
+ return this.stop();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean unlock() {
+ PolicyLogger.info(className, "UNLOCK: " + this);
+
+ synchronized(this) {
+ if (!this.locked)
+ return true;
+
+ this.locked = false;
+ }
+
+ try {
+ return this.start();
+ } catch (Exception e) {
+ PolicyLogger.warn("can't start after unlocking " + this +
+ " because of " + e.getMessage());
+ return false;
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean start() throws IllegalStateException {
+
+ PolicyLogger.info(className, "START: " + 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) {
+ e.printStackTrace();
+ throw new IllegalStateException(e);
+ }
+ }
+ }
+
+ return this.alive;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean stop() {
+ PolicyLogger.info(className, "STOP: " + this);
+
+ synchronized(this) {
+ BusConsumer consumerCopy = this.consumer;
+
+ this.alive = false;
+ this.consumer = null;
+
+ if (consumerCopy != null) {
+ try {
+ consumerCopy.close();
+ } catch (Exception e) {
+ PolicyLogger.warn(MessageCodes.EXCEPTION_ERROR, e, "CONSUMER.CLOSE", this.toString());
+ }
+ }
+ }
+
+ Thread.yield();
+
+ return true;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isLocked() {
+ return this.locked;
+ }
+
+ /**
+ * 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) {
+
+ /* take a snapshot of listeners */
+ List<TopicListener> snapshotListeners = this.snapshotTopicListeners();
+
+ boolean success = true;
+ for (TopicListener topicListener: snapshotListeners) {
+ try {
+ topicListener.onTopicEvent(this.getTopicCommInfrastructure(), this.topic, message);
+ } catch (Exception e) {
+ PolicyLogger.warn(this.className, "ERROR notifying " + topicListener.toString() +
+ " because of " + e.getMessage() + " @ " + this.toString());
+ 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;
+ }
+
+ /**
+ * 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);
+ }
+
+ if (networkLogger.isInfoEnabled()) {
+ networkLogger.info("IN[" + this.getTopicCommInfrastructure() + "|" +
+ this.topic + "]:" +
+ event);
+ }
+
+ PolicyLogger.info(className, this.topic + " <-- " + event);
+ broadcast(event);
+
+ if (!this.alive)
+ break;
+ }
+ } catch (Exception e) {
+ PolicyLogger.error( MessageCodes.EXCEPTION_ERROR, className, e, "CONSUMER.FETCH", this.toString());
+ }
+ }
+
+ PolicyLogger.warn(this.className, "Exiting: " + this);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean offer(String event) {
+ PolicyLogger.info(className, "OFFER: " + event + " TO " + this);
+
+ if (!this.alive) {
+ throw new IllegalStateException(this + " is not alive.");
+ }
+
+ synchronized (this) {
+ this.recentEvents.add(event);
+ }
+
+ if (networkLogger.isInfoEnabled()) {
+ networkLogger.info("IN[" + this.getTopicCommInfrastructure() + "|" +
+ this.topic + "]:" +
+ event);
+ }
+
+
+ return broadcast(event);
+ }
+
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("SingleThreadedBusTopicSource [consumerGroup=").append(consumerGroup)
+ .append(", consumerInstance=").append(consumerInstance).append(", fetchTimeout=").append(fetchTimeout)
+ .append(", fetchLimit=").append(fetchLimit)
+ .append(", consumer=").append(this.consumer).append(", alive=")
+ .append(alive).append(", locked=").append(locked).append(", uebThread=").append(busPollerThread)
+ .append(", topicListeners=").append(topicListeners.size()).append(", toString()=").append(super.toString())
+ .append("]");
+ return builder.toString();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isAlive() {
+ return alive;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public String getConsumerGroup() {
+ return consumerGroup;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public String getConsumerInstance() {
+ return consumerInstance;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void shutdown() throws IllegalStateException {
+ this.stop();
+ this.topicListeners.clear();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int getFetchTimeout() {
+ return fetchTimeout;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int getFetchLimit() {
+ return fetchLimit;
+ }
+
+}