src/main/java/org/onap/aai/event/EventBusConsumer.java

   1 /**
   2  * ============LICENSE_START=======================================================
   3  * org.onap.aai
   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
  11  *
  12  *       http://www.apache.org/licenses/LICENSE-2.0
  13  *
  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=========================================================
  20  */
  21 package org.onap.aai.event;
  22
  23 import org.onap.aai.event.api.EventConsumer;
  24
  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;
  33
  34 import java.util.Collections;
  35 import java.util.UUID;
  36 import java.util.concurrent.ScheduledThreadPoolExecutor;
  37
  38 /**
  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.
  42  */
  43 public class EventBusConsumer extends ScheduledPollConsumer {
  44
  45   private Logger logger = LoggerFactory.getInstance().getLogger(EventBusConsumer.class);
  46   private Logger auditLogger = LoggerFactory.getInstance().getAuditLogger(EventBusConsumer.class);
  47   private final AbstractEventBusEndpoint endpoint;
  48
  49   private EventConsumer consumer;
  50
  51   /**
  52    * EventBusConsumer Constructor.
  53    */
  54   public EventBusConsumer(AbstractEventBusEndpoint endpoint, Processor processor, EventConsumer consumer) {
  55     super(endpoint, processor);
  56     super.setDelay(endpoint.getPollingDelay());
  57     this.endpoint = endpoint;
  58
  59     setScheduledExecutorService(new ScheduledThreadPoolExecutor(endpoint.getPoolSize()));
  60
  61     this.consumer = consumer;
  62   }
  63
  64   /**
  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 .
  67    *
  68    * @return the number of messages that were processed off the event queue
  69    */
  70   @Override
  71   protected int poll() throws Exception {
  72
  73     String topic = endpoint.getEventTopic();
  74
  75     logger.debug("Checking for event on topic: " + topic);
  76
  77     int processCount = 0;
  78
  79     try {
  80       Iterable<String> messages = consumer.consumeAndCommit();
  81
  82       for (String message : messages) {
  83         Exchange exchange = endpoint.createExchange();
  84         exchange.getIn().setBody(message);
  85         getScheduledExecutorService().submit(new EventProcessor(exchange, topic));
  86         ++processCount;
  87       }
  88     } catch (Exception e) {
  89       logger.error(RouterCoreMsgs.EVENT_PROCESSING_EXCEPTION, e, e.getLocalizedMessage());
  90     }
  91
  92     logger.debug(RouterCoreMsgs.PROCESS_EVENT, topic, Integer.toString(processCount));
  93
  94     return processCount;
  95   }
  96   @Override
  97   protected void doStop() throws Exception {
  98     super.doStop();
  99     if (endpoint != null) {
 100       endpoint.close();
 101     }
 102   }
 103   @Override
 104   protected void doShutdown() throws Exception {
 105     super.doShutdown();
 106     if (endpoint != null) {
 107       endpoint.close();
 108     }
 109   }
 110
 111   /**
 112    * Class responsible for processing messages pulled off of the event bus.
 113    */
 114   private class EventProcessor implements Runnable {
 115
 116     private Exchange message;
 117
 118     private String topic;
 119
 120     EventProcessor(Exchange message, String topic) {
 121       this.message = message;
 122       this.topic = topic;
 123     }
 124         @Override
 125     public void run() {
 126       try {
 127
 128         MdcContext.initialize(UUID.randomUUID().toString(), "DataRouter", "", "Event-Bus", "");
 129
 130         // Sends the message to the next processor in the defined Camel route
 131         getProcessor().process(message);
 132
 133         Message response = message.getOut();
 134         if (response != null) {
 135           logger.debug("Routing response: " + response.getBody());
 136         }
 137
 138       } catch (Exception e) {
 139         logger.error(RouterCoreMsgs.EVENT_PROCESSING_EXCEPTION,e,e.getLocalizedMessage());
 140       } finally {
 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");
 145         } else {
 146           logger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "SUCCESS");
 147           auditLogger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "SUCCESS");
 148         }
 149       }
 150     }
 151   }
 152 }