src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * org.onap.dcaegen2.collectors.ves
   4  * ================================================================================
   5  * Copyright (C) 2017,2020 AT&T Intellectual Property. All rights reserved.
   6  * Copyright (C) 2018-2021 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
  22 package org.onap.dcae.common.publishing;
  23
  24 import io.vavr.collection.Map;
  25 import org.onap.dcae.common.model.VesEvent;
  26 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29 import org.springframework.http.HttpStatus;
  30 import reactor.core.publisher.Flux;
  31
  32 import java.util.List;
  33 import java.util.Objects;
  34
  35 import static org.onap.dcae.common.publishing.MessageRouterHttpStatusMapper.getHttpStatus;
  36 /**
  37  * @author Pawel Szalapski (pawel.szalapski@nokia.com)
  38  */
  39 public class DMaaPEventPublisher {
  40     private static final Logger log = LoggerFactory.getLogger(DMaaPEventPublisher.class);
  41     private Map<String, PublisherConfig> dMaaPConfig;
  42     private final Publisher dmaapPublisher;
  43
  44     public DMaaPEventPublisher(Map<String, PublisherConfig> dMaaPConfig) {
  45         this.dMaaPConfig = dMaaPConfig;
  46         dmaapPublisher = new Publisher();
  47     }
  48
  49     /**
  50      * Reload Dmaap configuration
  51      * @param dmaapConfiguration Dmaap configuration
  52      */
  53     public void reload(Map<String, PublisherConfig> dmaapConfiguration){
  54         dMaaPConfig = dmaapConfiguration;
  55         log.info("reload dmaap configuration");
  56     }
  57
  58     public HttpStatus sendEvent(List<VesEvent> vesEvents, String dmaapId) {
  59         clearVesUniqueIdFromEvent(vesEvents);
  60         io.vavr.collection.List<String> events = mapListOfEventsToVavrList(vesEvents);
  61         Flux<MessageRouterPublishResponse> messageRouterPublishFlux = dmaapPublisher.publishEvents(events, dMaaPConfig.get(dmaapId));
  62         MessageRouterPublishResponse messageRouterPublishResponse = messageRouterPublishFlux.blockFirst();
  63         return getHttpStatus(Objects.requireNonNull(messageRouterPublishResponse));
  64     }
  65
  66     private io.vavr.collection.List<String> mapListOfEventsToVavrList(List<VesEvent> vesEvents) {
  67         return io.vavr.collection.List.ofAll(vesEvents)
  68                 .map(event -> event.asJsonObject().toString());
  69     }
  70
  71     private void clearVesUniqueIdFromEvent(List<VesEvent> events) {
  72         events.stream()
  73                 .filter(event -> event.hasType(VesEvent.VES_UNIQUE_ID))
  74                 .forEach(event -> {
  75                     log.debug("Removing VESuniqueid object from event");
  76                     event.removeElement(VesEvent.VES_UNIQUE_ID);
  77                 });
  78     }
  79 }