2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 package org.openecomp.policy.drools.event.comm.bus.internal;
23 import java.util.ArrayList;
24 import java.util.List;
25 import java.util.UUID;
27 import org.slf4j.Logger;
28 import org.slf4j.LoggerFactory;
30 import org.openecomp.policy.drools.event.comm.TopicListener;
31 import org.openecomp.policy.drools.event.comm.bus.BusTopicSource;
34 * This topic source implementation specializes in reading messages
35 * over a bus topic source and notifying its listeners
37 public abstract class SingleThreadedBusTopicSource
39 implements Runnable, BusTopicSource {
42 * Not to be converted to PolicyLogger.
43 * This will contain all instract /out traffic and only that in a single file in a concise format.
45 private static Logger logger = LoggerFactory.getLogger(InlineBusTopicSink.class);
46 private static final Logger netLogger = LoggerFactory.getLogger(NETWORK_LOGGER);
51 protected final String consumerGroup;
54 * Bus consumer instance
56 protected final String consumerInstance;
61 protected final int fetchTimeout;
66 protected final int fetchLimit;
69 * Message Bus Consumer
71 protected BusConsumer consumer;
75 * reflects invocation of start()/stop()
76 * !locked & start() => alive
79 protected volatile boolean alive = false;
83 * reflects invocation of lock()/unlock() operations
84 * locked => !alive (but not in the other direction necessarily)
85 * locked => !offer, !run, !start, !stop (but this last one is obvious
86 * since locked => !alive)
88 protected volatile boolean locked = false;
91 * Independent thread reading message over my topic
93 protected Thread busPollerThread;
96 * All my subscribers for new message notifications
98 protected final ArrayList<TopicListener> topicListeners = new ArrayList<TopicListener>();
103 * @param servers Bus servers
104 * @param topic Bus Topic to be monitored
105 * @param apiKey Bus API Key (optional)
106 * @param apiSecret Bus API Secret (optional)
107 * @param consumerGroup Bus Reader Consumer Group
108 * @param consumerInstance Bus Reader Instance
109 * @param fetchTimeout Bus fetch timeout
110 * @param fetchLimit Bus fetch limit
111 * @param useHttps does the bus use https
112 * @param allowSelfSignedCerts are self-signed certificates allowed
113 * @throws IllegalArgumentException An invalid parameter passed in
115 public SingleThreadedBusTopicSource(List<String> servers,
119 String consumerGroup,
120 String consumerInstance,
124 boolean allowSelfSignedCerts)
125 throws IllegalArgumentException {
127 super(servers, topic, apiKey, apiSecret, useHttps, allowSelfSignedCerts);
129 if (consumerGroup == null || consumerGroup.isEmpty()) {
130 this.consumerGroup = UUID.randomUUID ().toString();
132 this.consumerGroup = consumerGroup;
135 if (consumerInstance == null || consumerInstance.isEmpty()) {
136 this.consumerInstance = DEFAULT_CONSUMER_INSTANCE;
138 this.consumerInstance = consumerInstance;
141 if (fetchTimeout <= 0) {
142 this.fetchTimeout = NO_TIMEOUT_MS_FETCH;
144 this.fetchTimeout = fetchTimeout;
147 if (fetchLimit <= 0) {
148 this.fetchLimit = NO_LIMIT_FETCH;
150 this.fetchLimit = fetchLimit;
156 * Initialize the Bus client
158 public abstract void init() throws Exception;
164 public void register(TopicListener topicListener)
165 throws IllegalArgumentException {
167 logger.info("{}: registering {}", this, topicListener);
170 if (topicListener == null)
171 throw new IllegalArgumentException("TopicListener must be provided");
173 /* check that this listener is not registered already */
174 for (TopicListener listener: this.topicListeners) {
175 if (listener == topicListener) {
176 // already registered
181 this.topicListeners.add(topicListener);
186 } catch (Exception e) {
187 logger.warn("{}: cannot start after registration of because of: {}",
188 this, topicListener, e.getMessage(), e);
196 public void unregister(TopicListener topicListener) {
198 logger.info("{}: unregistering {}", this, topicListener);
200 boolean stop = false;
201 synchronized (this) {
202 if (topicListener == null)
203 throw new IllegalArgumentException("TopicListener must be provided");
205 this.topicListeners.remove(topicListener);
206 stop = (this.topicListeners.isEmpty());
218 public boolean lock() {
220 logger.info("{}: locking", this);
222 synchronized (this) {
236 public boolean unlock() {
237 logger.info("{}: unlocking", this);
248 } catch (Exception e) {
249 logger.warn("{}: cannot after unlocking because of {}", this, e.getMessage(), e);
258 public boolean start() throws IllegalStateException {
259 logger.info("{}: starting", this);
267 throw new IllegalStateException(this + " is locked.");
269 if (this.busPollerThread == null ||
270 !this.busPollerThread.isAlive() ||
271 this.consumer == null) {
276 this.busPollerThread = new Thread(this);
277 this.busPollerThread.setName(this.getTopicCommInfrastructure() + "-source-" + this.getTopic());
278 busPollerThread.start();
279 } catch (Exception e) {
280 logger.warn("{}: cannot start because of {}", this, e.getMessage(), e);
281 throw new IllegalStateException(e);
293 public boolean stop() {
294 logger.info("{}: stopping", this);
297 BusConsumer consumerCopy = this.consumer;
300 this.consumer = null;
302 if (consumerCopy != null) {
304 consumerCopy.close();
305 } catch (Exception e) {
306 logger.warn("{}: stop failed because of {}", this, e.getMessage(), e);
320 public boolean isLocked() {
325 * broadcast event to all listeners
327 * @param message the event
328 * @return true if all notifications are performed with no error, false otherwise
330 protected boolean broadcast(String message) {
332 /* take a snapshot of listeners */
333 List<TopicListener> snapshotListeners = this.snapshotTopicListeners();
335 boolean success = true;
336 for (TopicListener topicListener: snapshotListeners) {
338 topicListener.onTopicEvent(this.getTopicCommInfrastructure(), this.topic, message);
339 } catch (Exception e) {
340 logger.warn("{}: notification error @ {} because of {}",
341 this, topicListener, e.getMessage(), e);
349 * take a snapshot of current topic listeners
351 * @return the topic listeners
353 protected synchronized List<TopicListener> snapshotTopicListeners() {
354 @SuppressWarnings("unchecked")
355 List<TopicListener> listeners = (List<TopicListener>) topicListeners.clone();
360 * Run thread method for the Bus Reader
366 for (String event: this.consumer.fetch()) {
367 synchronized (this) {
368 this.recentEvents.add(event);
371 netLogger.info("[IN|{}|{}]{}{}",
372 this.getTopicCommInfrastructure(), this.topic,
373 System.lineSeparator(), event);
380 } catch (Exception e) {
381 logger.error("{}: cannot fetch because of ", this, e.getMessage(), e);
385 logger.info("{}: exiting thread", this);
392 public boolean offer(String event) {
394 throw new IllegalStateException(this + " is not alive.");
397 synchronized (this) {
398 this.recentEvents.add(event);
401 netLogger.info("[IN|{}|{}]{}{}",this.getTopicCommInfrastructure(),this.topic,
402 System.lineSeparator(), event);
405 return broadcast(event);
410 public String toString() {
411 StringBuilder builder = new StringBuilder();
412 builder.append("SingleThreadedBusTopicSource [consumerGroup=").append(consumerGroup)
413 .append(", consumerInstance=").append(consumerInstance).append(", fetchTimeout=").append(fetchTimeout)
414 .append(", fetchLimit=").append(fetchLimit)
415 .append(", consumer=").append(this.consumer).append(", alive=")
416 .append(alive).append(", locked=").append(locked).append(", uebThread=").append(busPollerThread)
417 .append(", topicListeners=").append(topicListeners.size()).append(", toString()=").append(super.toString())
419 return builder.toString();
426 public boolean isAlive() {
434 public String getConsumerGroup() {
435 return consumerGroup;
442 public String getConsumerInstance() {
443 return consumerInstance;
450 public void shutdown() throws IllegalStateException {
452 this.topicListeners.clear();
459 public int getFetchTimeout() {
467 public int getFetchLimit() {