src/main/java/org/onap/crud/service/CrudAsyncResponseConsumer.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.crud.service;
  22
  23 import java.util.Objects;
  24 import java.util.TimerTask;
  25 import org.onap.aai.cl.api.Logger;
  26 import org.onap.aai.cl.eelf.LoggerFactory;
  27 import org.onap.aai.event.api.EventConsumer;
  28 import org.onap.crud.logging.CrudServiceMsgs;
  29
  30 public class CrudAsyncResponseConsumer extends TimerTask {
  31
  32     private static Logger logger = LoggerFactory.getInstance().getLogger(CrudAsyncResponseConsumer
  33         .class.getName());
  34
  35     private final EventConsumer asyncResponseConsumer;
  36     private final GraphEventUpdater graphEventUpdater;
  37
  38
  39     public CrudAsyncResponseConsumer(EventConsumer asyncResponseConsumer, GraphEventUpdater graphEventUpdater) {
  40         Objects.requireNonNull(asyncResponseConsumer);
  41         Objects.requireNonNull(graphEventUpdater);
  42         this.asyncResponseConsumer = asyncResponseConsumer;
  43         this.graphEventUpdater = graphEventUpdater;
  44         logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
  45             "CrudAsyncResponseConsumer initialized SUCCESSFULLY! with event consumer "
  46                 + asyncResponseConsumer.getClass().getName());
  47     }
  48
  49
  50     @Override
  51     public void run() {
  52
  53         logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "Listening for graph events");
  54
  55         try {
  56             Iterable<String> events = asyncResponseConsumer.consume();
  57             processEvents(events);
  58             asyncResponseConsumer.commitOffsets();
  59         } catch (Exception e) {
  60             logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e, e.getMessage());
  61         }
  62     }
  63
  64     private void processEvents(Iterable<String> events) {
  65         if (areEventsAvailable(events)) {
  66             for (String event : events) {
  67                 graphEventUpdater.update(event);
  68             }
  69             logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "No events recieved");
  70         } else {
  71             logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "No events recieved");
  72         }
  73     }
  74
  75     private boolean areEventsAvailable(Iterable<String> events) {
  76         return !(events == null || !events.iterator().hasNext());
  77     }
  78
  79 }