src/main/java/org/onap/dcae/common/EventSender.java

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * VES Collector
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * Copyright (C) 2018,2020 Nokia. All rights reserved.
   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.dcae.common;
  22
  23 import com.att.nsa.clock.SaClock;
  24 import com.att.nsa.logging.LoggingContext;
  25 import com.att.nsa.logging.log4j.EcompFields;
  26 import io.vavr.collection.Map;
  27 import org.onap.dcae.common.model.VesEvent;
  28 import org.onap.dcae.common.publishing.DMaaPEventPublisher;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31
  32 import java.util.List;
  33
  34 public class EventSender {
  35
  36   private static final Logger metriclog = LoggerFactory.getLogger("com.att.ecomp.metrics");
  37   private Map<String, String[]> streamIdToDmaapIds;
  38   private DMaaPEventPublisher eventPublisher;
  39   private static final Logger log = LoggerFactory.getLogger(EventSender.class);
  40
  41   public EventSender(DMaaPEventPublisher eventPublisher, Map<String, String[]> streamIdToDmaapIds) {
  42     this.eventPublisher = eventPublisher;
  43     this.streamIdToDmaapIds = streamIdToDmaapIds;
  44   }
  45
  46   public void send(List<VesEvent> vesEvents) {
  47     for (VesEvent vesEvent : vesEvents) {
  48       metriclog.info("EVENT_PUBLISH_START");
  49       setLoggingContext(vesEvent);
  50       streamIdToDmaapIds.get(vesEvent.getStreamId())
  51           .onEmpty(() -> log.error("No StreamID defined for publish - Message dropped" + vesEvent.asJsonObject()))
  52           .forEach(streamIds -> sendEventsToStreams(vesEvent, streamIds));
  53       log.debug("Message published" + vesEvent.asJsonObject());
  54     }
  55     log.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");
  56     metriclog.info("EVENT_PUBLISH_END");
  57   }
  58
  59   private void sendEventsToStreams(VesEvent vesEvent, String[] streamIdList) {
  60     for (String streamId : streamIdList) {
  61       log.info("Invoking publisher for streamId/domain:" + streamId);
  62       eventPublisher.sendEvent(vesEvent.asJsonObject(), streamId);
  63     }
  64   }
  65
  66   private void setLoggingContext(VesEvent vesEvent) {
  67     LoggingContext localLC = VESLogger.getLoggingContextForThread(vesEvent.getUniqueId().toString());
  68     localLC.put(EcompFields.kBeginTimestampMs, SaClock.now());
  69     log.debug("event.VESuniqueId" + vesEvent.getUniqueId() + "event.commonEventHeader.domain:" + vesEvent.getDomain());
  70   }
  71 }