2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright © 2017 AT&T Intellectual Property.
6 * Copyright © 2017 Amdocs
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
13 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
22 * ECOMP is a trademark and service mark of AT&T Intellectual Property.
24 package org.onap.crud.service;
26 import java.util.TimerTask;
28 import javax.naming.OperationNotSupportedException;
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;
35 import com.att.ecomp.event.api.EventConsumer;
37 public class CrudAsyncResponseConsumer extends TimerTask {
39 private static Logger logger = LoggerFactory.getInstance().getLogger(CrudAsyncResponseConsumer
42 private static Logger auditLogger = LoggerFactory.getInstance()
43 .getAuditLogger(CrudAsyncResponseConsumer.class.getName());
45 private EventConsumer asyncResponseConsumer;
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());
59 logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "Listening for graph events");
61 if (asyncResponseConsumer == null) {
62 logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR,
63 "Unable to initialize CrudAsyncRequestProcessor");
66 Iterable<String> events = null;
68 events = asyncResponseConsumer.consume();
69 } catch (Exception e) {
70 logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
74 if (events == null || !events.iterator().hasNext()) {
75 logger.info(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_INFO, "No events recieved");
79 for (String event : events) {
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);
96 if (CrudAsyncGraphEventCache.get(graphEvent.getTransactionId()) != null) {
97 CrudAsyncGraphEventCache.get(graphEvent.getTransactionId())
98 .populateGraphEvent(graphEvent);
100 logger.error(CrudServiceMsgs.ASYNC_DATA_SERVICE_ERROR,
101 "Request timed out. Not sending response for transaction-id: "
102 + graphEvent.getTransactionId());
105 } catch (Exception e) {
106 logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
111 asyncResponseConsumer.commitOffsets();
113 catch(OperationNotSupportedException e) {
114 //Dmaap doesnt support commit with offset
115 logger.debug(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());
117 catch (Exception e) {
118 logger.error(CrudServiceMsgs.ASYNC_RESPONSE_CONSUMER_ERROR, e.getMessage());