2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017-2018 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.onap.policy.common.endpoints.event.comm.bus.internal;
23 import java.net.MalformedURLException;
24 import java.util.List;
25 import java.util.UUID;
27 import org.onap.policy.common.endpoints.event.comm.FilterableTopicSource;
28 import org.onap.policy.common.endpoints.event.comm.TopicListener;
29 import org.onap.policy.common.endpoints.event.comm.bus.BusTopicSource;
30 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.FilterableBusConsumer;
31 import org.onap.policy.common.utils.network.NetworkUtil;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
36 * This topic source implementation specializes in reading messages over a bus topic source and
37 * notifying its listeners
39 public abstract class SingleThreadedBusTopicSource extends BusTopicBase
40 implements Runnable, BusTopicSource, FilterableTopicSource {
43 * Not to be converted to PolicyLogger. This will contain all instract /out traffic and only
44 * that in a single file in a concise format.
46 private static Logger logger = LoggerFactory.getLogger(InlineBusTopicSink.class);
47 private static final Logger netLogger = LoggerFactory.getLogger(NETWORK_LOGGER);
52 protected final String consumerGroup;
55 * Bus consumer instance
57 protected final String consumerInstance;
62 protected final int fetchTimeout;
67 protected final int fetchLimit;
70 * Message Bus Consumer
72 protected BusConsumer consumer;
75 * Independent thread reading message over my topic
77 protected Thread busPollerThread;
82 * @param servers Bus servers
83 * @param topic Bus Topic to be monitored
84 * @param apiKey Bus API Key (optional)
85 * @param apiSecret Bus API Secret (optional)
86 * @param consumerGroup Bus Reader Consumer Group
87 * @param consumerInstance Bus Reader Instance
88 * @param fetchTimeout Bus fetch timeout
89 * @param fetchLimit Bus fetch limit
90 * @param useHttps does the bus use https
91 * @param allowSelfSignedCerts are self-signed certificates allowed
92 * @throws IllegalArgumentException An invalid parameter passed in
94 public SingleThreadedBusTopicSource(List<String> servers, String topic, String apiKey, String apiSecret,
95 String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit, boolean useHttps,
96 boolean allowSelfSignedCerts) {
98 super(servers, topic, apiKey, apiSecret, useHttps, allowSelfSignedCerts);
100 if (consumerGroup == null || consumerGroup.isEmpty()) {
101 this.consumerGroup = UUID.randomUUID().toString();
103 this.consumerGroup = consumerGroup;
106 if (consumerInstance == null || consumerInstance.isEmpty()) {
107 this.consumerInstance = NetworkUtil.getHostname();
109 this.consumerInstance = consumerInstance;
112 if (fetchTimeout <= 0) {
113 this.fetchTimeout = NO_TIMEOUT_MS_FETCH;
115 this.fetchTimeout = fetchTimeout;
118 if (fetchLimit <= 0) {
119 this.fetchLimit = NO_LIMIT_FETCH;
121 this.fetchLimit = fetchLimit;
127 * Initialize the Bus client
129 public abstract void init() throws MalformedURLException;
132 public void register(TopicListener topicListener) {
134 super.register(topicListener);
137 if (!alive && !locked) {
140 logger.info("{}: register: start not attempted", this);
142 } catch (Exception e) {
143 logger.warn("{}: cannot start after registration of because of: {}", this, topicListener, e.getMessage(),
149 public void unregister(TopicListener topicListener) {
151 synchronized (this) {
152 super.unregister(topicListener);
153 stop = this.topicListeners.isEmpty();
162 public boolean start() {
163 logger.info("{}: starting", this);
165 synchronized (this) {
172 throw new IllegalStateException(this + " is locked.");
175 if (this.busPollerThread == null || !this.busPollerThread.isAlive() || this.consumer == null) {
180 this.busPollerThread = new Thread(this);
181 this.busPollerThread.setName(this.getTopicCommInfrastructure() + "-source-" + this.getTopic());
182 busPollerThread.start();
183 } catch (Exception e) {
184 logger.warn("{}: cannot start because of {}", this, e.getMessage(), e);
185 throw new IllegalStateException(e);
194 public boolean stop() {
195 logger.info("{}: stopping", this);
197 synchronized (this) {
198 BusConsumer consumerCopy = this.consumer;
201 this.consumer = null;
203 if (consumerCopy != null) {
205 consumerCopy.close();
206 } catch (Exception e) {
207 logger.warn("{}: stop failed because of {}", this, e.getMessage(), e);
218 * Run thread method for the Bus Reader
224 for (String event : this.consumer.fetch()) {
225 synchronized (this) {
226 this.recentEvents.add(event);
229 netLogger.info("[IN|{}|{}]{}{}", this.getTopicCommInfrastructure(), this.topic,
230 System.lineSeparator(), event);
238 } catch (Exception e) {
239 logger.error("{}: cannot fetch because of ", this, e.getMessage(), e);
243 logger.info("{}: exiting thread", this);
250 public boolean offer(String event) {
252 throw new IllegalStateException(this + " is not alive.");
255 synchronized (this) {
256 this.recentEvents.add(event);
259 netLogger.info("[IN|{}|{}]{}{}", this.getTopicCommInfrastructure(), this.topic, System.lineSeparator(), event);
262 return broadcast(event);
267 public void setFilter(String filter) {
268 if (consumer instanceof FilterableBusConsumer) {
269 ((FilterableBusConsumer) consumer).setFilter(filter);
272 throw new UnsupportedOperationException("no server-side filtering for topic " + topic);
277 public String toString() {
278 return "SingleThreadedBusTopicSource [consumerGroup=" + consumerGroup + ", consumerInstance=" + consumerInstance
279 + ", fetchTimeout=" + fetchTimeout + ", fetchLimit=" + fetchLimit + ", consumer=" + this.consumer
280 + ", alive=" + alive + ", locked=" + locked + ", uebThread=" + busPollerThread + ", topicListeners="
281 + topicListeners.size() + ", toString()=" + super.toString() + "]";
288 public String getConsumerGroup() {
289 return consumerGroup;
296 public String getConsumerInstance() {
297 return consumerInstance;
304 public void shutdown() {
306 this.topicListeners.clear();
313 public int getFetchTimeout() {
321 public int getFetchLimit() {