src/main/java/org/onap/crud/service/CrudAsyncResponseConsumer.java

   1 /**
   2  * ============LICENSE_START=======================================================
   3  * Gizmo
   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 at
  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 is a trademark and service mark of AT&T Intellectual Property.
  23  */
  24 package org.onap.crud.service;
  25
  26 import java.util.TimerTask;
  27
  28 import javax.naming.OperationNotSupportedException;
  29
  30 import org.onap.aai.cl.api.Logger;
  31 import org.onap.aai.cl.eelf.LoggerFactory;
  32 import org.onap.crud.event.GraphEvent;
  33 import org.onap.crud.logging.CrudServiceMsgs;
  34
  35 import org.onap.aai.event.api.EventConsumer;
  36
  37 public class CrudAsyncResponseConsumer extends TimerTask {
  38
  39   private static Logger logger = LoggerFactory.getInstance().getLogger(CrudAsyncResponseConsumer
  40                                                                        .class.getName());
  41
  42   private static Logger auditLogger = LoggerFactory.getInstance()
  43     .getAuditLogger(CrudAsyncResponseConsumer.class.getName());
  44
  45   private EventConsumer asyncResponseConsumer;
  46
  47
  48   public CrudAsyncResponseConsumer(EventConsumer asyncResponseConsumer) {
  49     this.asyncResponseConsumer = asyncResponseConsumer;
  50     logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
  51                 "CrudAsyncResponseConsumer initialized SUCCESSFULLY! with event consumer "
  52                 + asyncResponseConsumer.getClass().getName());
  53   }
  54
  55
  56   @Override
  57   public void run() {
  58
  59     logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "Listening for graph events");
  60
  61     if (asyncResponseConsumer == null) {
  62       logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR,
  63                    "Unable to initialize CrudAsyncRequestProcessor");
  64     }
  65
  66     Iterable<String> events = null;
  67     try {
  68       events = asyncResponseConsumer.consume();
  69     } catch (Exception e) {
  70       logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
  71       return;
  72     }
  73
  74     if (events == null || !events.iterator().hasNext()) {
  75       logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "No events recieved");
  76
  77     }
  78
  79     for (String event : events) {
  80       try {
  81
  82         GraphEvent graphEvent = GraphEvent.fromJson(event);
  83         auditLogger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
  84                            "Event received of type: " + graphEvent.getObjectType() + " with key: "
  85                            + graphEvent.getObjectKey() + " , transaction-id: "
  86                            + graphEvent.getTransactionId() + " , operation: "
  87                            + graphEvent.getOperation().toString());
  88           logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
  89                       "Event received of type: " + graphEvent.getObjectType() + " with key: "
  90                       + graphEvent.getObjectKey() + " , transaction-id: "
  91                       + graphEvent.getTransactionId() + " , operation: "
  92                       + graphEvent.getOperation().toString());
  93           logger.debug(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO,
  94                        "Event received with payload:" + event);
  95
  96         if (CrudAsyncGraphEventCache.get(graphEvent.getTransactionId()) != null) {
  97           CrudAsyncGraphEventCache.get(graphEvent.getTransactionId())
  98             .populateGraphEvent(graphEvent);
  99         } else {
 100           logger.error(CrudServiceMsgs.ASYNC_DATA_SERVICE_ERROR,
 101                        "Request timed out. Not sending response for transaction-id: "
 102                        + graphEvent.getTransactionId());
 103         }
 104
 105       } catch (Exception e) {
 106         logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
 107       }
 108     }
 109
 110     try {
 111       asyncResponseConsumer.commitOffsets();
 112     }
 113     catch(OperationNotSupportedException e) {
 114         //Dmaap doesnt support commit with offset
 115         logger.debug(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
 116     }
 117     catch (Exception e) {
 118       logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
 119     }
 120
 121   }
 122
 123 }