src/main/java/org/openecomp/event/EventBusConsumer.java

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