2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
6 * Copyright © 2017-2018 Amdocs
7 * ================================================================================
8 * Licensed under the Apache License, Version 2.0 (the "License");
9 * you may not use this file except in compliance with the License.
10 * You may obtain a copy of the License at
12 * http://www.apache.org/licenses/LICENSE-2.0
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the License is distributed on an "AS IS" BASIS,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 * See the License for the specific language governing permissions and
18 * limitations under the License.
19 * ============LICENSE_END=========================================================
21 package org.onap.aai.event;
23 import org.onap.aai.event.api.EventConsumer;
25 import org.apache.camel.Exchange;
26 import org.apache.camel.Message;
27 import org.apache.camel.Processor;
28 import org.apache.camel.impl.ScheduledPollConsumer;
29 import org.onap.aai.logging.RouterCoreMsgs;
30 import org.onap.aai.cl.api.Logger;
31 import org.onap.aai.cl.eelf.LoggerFactory;
32 import org.onap.aai.cl.mdc.MdcContext;
34 import java.util.Collections;
35 import java.util.UUID;
36 import java.util.concurrent.ScheduledThreadPoolExecutor;
39 * The consumer component which is used to pull messages off of the event bus and send them to the
40 * next processor in the route chain. This type of consumer is based off of a scheduled poller so
41 * that events are pulled on a regular basis.
43 public class EventBusConsumer extends ScheduledPollConsumer {
45 private Logger logger = LoggerFactory.getInstance().getLogger(EventBusConsumer.class);
46 private Logger auditLogger = LoggerFactory.getInstance().getAuditLogger(EventBusConsumer.class);
47 private final AbstractEventBusEndpoint endpoint;
49 private EventConsumer consumer;
52 * EventBusConsumer Constructor.
54 public EventBusConsumer(AbstractEventBusEndpoint endpoint, Processor processor, EventConsumer consumer) {
55 super(endpoint, processor);
56 super.setDelay(endpoint.getPollingDelay());
57 this.endpoint = endpoint;
59 setScheduledExecutorService(new ScheduledThreadPoolExecutor(endpoint.getPoolSize()));
61 this.consumer = consumer;
65 * Method which is called by the Camel process on a scheduled basis. This specific implementation
66 * reads messages off of the configured topic and schedules tasks to process them .
68 * @return the number of messages that were processed off the event queue
71 protected int poll() throws Exception {
73 String topic = endpoint.getEventTopic();
75 logger.debug("Checking for event on topic: " + topic);
80 Iterable<String> messages = consumer.consumeAndCommit();
82 for (String message : messages) {
83 Exchange exchange = endpoint.createExchange();
84 exchange.getIn().setBody(message);
85 getScheduledExecutorService().submit(new EventProcessor(exchange, topic));
88 } catch (Exception e) {
89 logger.error(RouterCoreMsgs.EVENT_PROCESSING_EXCEPTION, e, e.getLocalizedMessage());
92 logger.debug(RouterCoreMsgs.PROCESS_EVENT, topic, Integer.toString(processCount));
97 protected void doStop() throws Exception {
99 if (endpoint != null) {
104 protected void doShutdown() throws Exception {
106 if (endpoint != null) {
112 * Class responsible for processing messages pulled off of the event bus.
114 private class EventProcessor implements Runnable {
116 private Exchange message;
118 private String topic;
120 EventProcessor(Exchange message, String topic) {
121 this.message = message;
128 MdcContext.initialize(UUID.randomUUID().toString(), "DataRouter", "", "Event-Bus", "");
130 // Sends the message to the next processor in the defined Camel route
131 getProcessor().process(message);
133 Message response = message.getOut();
134 if (response != null) {
135 logger.debug("Routing response: " + response.getBody());
138 } catch (Exception e) {
139 logger.error(RouterCoreMsgs.EVENT_PROCESSING_EXCEPTION,e,e.getLocalizedMessage());
141 // log exception if an exception occurred and was not handled
142 if (message.getException() != null) {
143 logger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "FAILURE");
144 auditLogger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "FAILURE");
146 logger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "SUCCESS");
147 auditLogger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "SUCCESS");