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

   1 /**
   2  * ============LICENSE_START=======================================================
   3  * org.onap.aai
   4  * ================================================================================
   5  * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  * Copyright © 2017 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  * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  22  */
  23 package org.onap.aai.event;
  24
  25 import com.att.nsa.cambria.client.CambriaClientBuilders;
  26 import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder;
  27 import com.att.nsa.cambria.client.CambriaConsumer;
  28
  29 import org.apache.camel.Exchange;
  30 import org.apache.camel.Message;
  31 import org.apache.camel.Processor;
  32 import org.apache.camel.impl.ScheduledPollConsumer;
  33 import org.onap.aai.logging.RouterCoreMsgs;
  34 import org.onap.aai.cl.api.Logger;
  35 import org.onap.aai.cl.eelf.LoggerFactory;
  36 import org.onap.aai.cl.mdc.MdcContext;
  37
  38 import java.net.MalformedURLException;
  39 import java.security.GeneralSecurityException;
  40 import java.util.Arrays;
  41 import java.util.List;
  42 import java.util.UUID;
  43 import java.util.concurrent.ScheduledThreadPoolExecutor;
  44
  45 /**
  46  * The consumer component which is used to pull messages off of the event bus and send them to the
  47  * next processor in the route chain. This type of consumer is based off of a scheduled poller so
  48  * that events are pulled on a regular basis.
  49  */
  50 public class EventBusConsumer extends ScheduledPollConsumer {
  51
  52   private Logger logger = LoggerFactory.getInstance().getLogger(EventBusConsumer.class);
  53   private Logger auditLogger = LoggerFactory.getInstance().getAuditLogger(EventBusConsumer.class);
  54   private final EventBusEndpoint endpoint;
  55
  56   private CambriaConsumer consumer;
  57
  58   /**
  59    * EventBusConsumer Constructor.
  60    */
  61   public EventBusConsumer(EventBusEndpoint endpoint, Processor processor) {
  62     super(endpoint, processor);
  63     super.setDelay(endpoint.getPollingDelay());
  64     this.endpoint = endpoint;
  65
  66     setScheduledExecutorService(new ScheduledThreadPoolExecutor(endpoint.getPoolSize()));
  67
  68     String[] urls = endpoint.getUrl().split(",");
  69
  70     List<String> urlList = null;
  71
  72     if (urls != null) {
  73       urlList = Arrays.asList(urls);
  74     }
  75
  76     try {
  77
  78       ConsumerBuilder consumerBuilder = new CambriaClientBuilders.ConsumerBuilder()
  79           .usingHosts(urlList).onTopic(endpoint.getEventTopic())
  80           .knownAs(endpoint.getGroupName(), endpoint.getGroupId());
  81
  82       String apiKey = endpoint.getApiKey();
  83       String apiSecret = endpoint.getApiSecret();
  84
  85       if (apiKey != null && apiSecret != null) {
  86         consumerBuilder.authenticatedBy(endpoint.getApiKey(), endpoint.getApiSecret());
  87       }
  88
  89       consumer = consumerBuilder.build();
  90
  91     } catch (MalformedURLException | GeneralSecurityException e) {
  92       logger.error(RouterCoreMsgs.EVENT_CONSUMER_CREATION_EXCEPTION, e.getLocalizedMessage());
  93     }
  94   }
  95
  96   /**
  97    * Method which is called by the Camel process on a scheduled basis. This specific implementation
  98    * reads messages off of the configured topic and schedules tasks to process them .
  99    *
 100    * @return the number of messages that were processed off the event queue
 101    */
 102   @Override
 103   protected int poll() throws Exception {
 104
 105     logger.debug("Checking for event on topic: " + endpoint.getEventTopic());
 106
 107     int processCount = 0;
 108
 109     Iterable<String> messages = null;
 110
 111     messages = consumer.fetch();
 112
 113     String topic = endpoint.getEventTopic();
 114
 115     for (String message : messages) {
 116       Exchange exchange = endpoint.createExchange();
 117       exchange.getIn().setBody(message);
 118       getScheduledExecutorService().submit(new EventProcessor(exchange, topic));
 119       ++processCount;
 120     }
 121     return processCount;
 122   }
 123   @Override
 124   protected void doStop() throws Exception {
 125     super.doStop();
 126     if (consumer != null) {
 127       consumer.close();
 128     }
 129   }
 130   @Override
 131   protected void doShutdown() throws Exception {
 132     super.doShutdown();
 133     if (consumer != null) {
 134       consumer.close();
 135     }
 136   }
 137
 138   /**
 139    * Class responsible for processing messages pulled off of the event bus.
 140    */
 141   private class EventProcessor implements Runnable {
 142
 143     private Exchange message;
 144
 145     private String topic;
 146
 147     EventProcessor(Exchange message, String topic) {
 148       this.message = message;
 149       this.topic = topic;
 150     }
 151         @Override
 152     public void run() {
 153       try {
 154
 155         MdcContext.initialize(UUID.randomUUID().toString(), "DataRouter", "", "Event-Bus", "");
 156
 157         // Sends the message to the next processor in the defined Camel route
 158         getProcessor().process(message);
 159
 160         Message response = message.getOut();
 161         if (response != null) {
 162           logger.debug("Routing response: " + response.getBody());
 163         }
 164
 165       } catch (Exception e) {
 166         logger.error(RouterCoreMsgs.EVENT_PROCESSING_EXCEPTION, e.getLocalizedMessage());
 167       } finally {
 168         // log exception if an exception occurred and was not handled
 169         if (message.getException() != null) {
 170           logger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "FAILURE");
 171           auditLogger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "FAILURE");
 172         } else {
 173           logger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "SUCCESS");
 174           auditLogger.info(RouterCoreMsgs.PROCESS_EVENT, topic, "SUCCESS");
 175         }
 176       }
 177     }
 178   }
 179 }