2  * ============LICENSE_START=======================================================
 
   3  * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
 
   4  * ================================================================================
 
   5  * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
 
   6  * ================================================================================
 
   7  * Licensed under the Apache License, Version 2.0 (the "License");
 
   8  * you may not use this file except in compliance with the License.
 
   9  * You may obtain a copy of the License at
 
  11  *      http://www.apache.org/licenses/LICENSE-2.0
 
  13  * Unless required by applicable law or agreed to in writing, software
 
  14  * distributed under the License is distributed on an "AS IS" BASIS,
 
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 
  16  * See the License for the specific language governing permissions and
 
  17  * limitations under the License.
 
  18  * ============LICENSE_END=========================================================
 
  21 package org.onap.bbs.event.processor.utilities;
 
  23 import static org.onap.bbs.event.processor.utilities.CommonEventFields.COMMON_FORMAT;
 
  24 import static org.onap.bbs.event.processor.utilities.CommonEventFields.CORRELATION_ID;
 
  25 import static org.onap.bbs.event.processor.utilities.ReRegistrationEventFields.ADDITIONAL_FIELDS;
 
  26 import static org.onap.bbs.event.processor.utilities.ReRegistrationEventFields.ATTACHMENT_POINT;
 
  27 import static org.onap.bbs.event.processor.utilities.ReRegistrationEventFields.CVLAN;
 
  28 import static org.onap.bbs.event.processor.utilities.ReRegistrationEventFields.REMOTE_ID;
 
  29 import static org.onap.bbs.event.processor.utilities.ReRegistrationEventFields.SVLAN;
 
  31 import com.google.gson.Gson;
 
  32 import com.google.gson.JsonElement;
 
  33 import com.google.gson.JsonObject;
 
  34 import com.google.gson.JsonParser;
 
  35 import com.google.gson.JsonSyntaxException;
 
  37 import java.util.Optional;
 
  38 import java.util.stream.StreamSupport;
 
  40 import org.onap.bbs.event.processor.exceptions.DmaapException;
 
  41 import org.onap.bbs.event.processor.model.ImmutableReRegistrationConsumerDmaapModel;
 
  42 import org.onap.bbs.event.processor.model.ReRegistrationConsumerDmaapModel;
 
  43 import org.slf4j.Logger;
 
  44 import org.slf4j.LoggerFactory;
 
  45 import org.springframework.util.StringUtils;
 
  47 import reactor.core.publisher.Flux;
 
  48 import reactor.core.publisher.Mono;
 
  50 public class ReRegistrationDmaapConsumerJsonParser {
 
  52     private static final Logger LOGGER = LoggerFactory.getLogger(ReRegistrationDmaapConsumerJsonParser.class);
 
  53     private static final Gson gson = new Gson();
 
  55     private static final String RE_REGISTRATION_DUMPING_TEMPLATE = "%n{"
 
  56             + "\"" + CORRELATION_ID + COMMON_FORMAT + ","
 
  57             + "\"" + ATTACHMENT_POINT + COMMON_FORMAT + ","
 
  58             + "\"" + REMOTE_ID + COMMON_FORMAT + ","
 
  59             + "\"" + CVLAN + COMMON_FORMAT + ","
 
  60             + "\"" + SVLAN + COMMON_FORMAT
 
  63     private String pnfCorrelationId;
 
  65     private String attachmentPoint;
 
  66     private String remoteId;
 
  71      * Translates a response from DMaaP to a reactive {@link ReRegistrationConsumerDmaapModel} model.
 
  72      * @param dmaapResponse Response from DMaaP
 
  73      * @return Re-Registration Consumer DMaaP reactive model
 
  75     public Flux<ReRegistrationConsumerDmaapModel> extractModelFromDmaap(Mono<String> dmaapResponse) {
 
  77                 .flatMapMany(this::parseToMono)
 
  78                 .flatMap(this::createTargetFlux);
 
  81     private Mono<JsonElement> parseToMono(String message) {
 
  82         if (StringUtils.isEmpty(message)) {
 
  83             LOGGER.warn("DMaaP response is empty");
 
  86         return Mono.fromCallable(() -> new JsonParser().parse(message))
 
  87                 .doOnError(e -> e instanceof JsonSyntaxException, e -> LOGGER.error("Invalid JSON. Ignoring"))
 
  88                 .onErrorResume(e -> e instanceof JsonSyntaxException, e -> Mono.empty());
 
  91     private Flux<ReRegistrationConsumerDmaapModel> createTargetFlux(JsonElement jsonElement) {
 
  92         if (jsonElement.isJsonObject()) {
 
  93             return doCreateTargetFlux(Flux.defer(() -> Flux.just(jsonElement.getAsJsonObject())));
 
  95         return doCreateTargetFlux(
 
  96                 Flux.defer(() -> Flux.fromStream(StreamSupport.stream(jsonElement.getAsJsonArray().spliterator(), false)
 
  97                         .map(jsonElementFromArray -> getJsonObjectFromAnArray(jsonElementFromArray)
 
  98                                 .orElseGet(JsonObject::new)))));
 
 101     private Flux<ReRegistrationConsumerDmaapModel> doCreateTargetFlux(Flux<JsonObject> jsonObject) {
 
 103                 .flatMap(this::transform)
 
 104                 .onErrorResume(exception -> exception instanceof DmaapException, e -> Mono.empty());
 
 107     private Mono<ReRegistrationConsumerDmaapModel> transform(JsonObject dmaapResponseJsonObject) {
 
 109         LOGGER.trace("Event from DMaaP to be parsed: \n{}", gson.toJson(dmaapResponseJsonObject));
 
 111         if (!containsProperHeaders(dmaapResponseJsonObject)) {
 
 112             LOGGER.warn("Incorrect JsonObject - missing headers");
 
 116         JsonObject pnfReRegistrationFields =
 
 117                 dmaapResponseJsonObject.getAsJsonObject(ADDITIONAL_FIELDS);
 
 119         pnfCorrelationId = getValueFromJson(dmaapResponseJsonObject, CORRELATION_ID);
 
 121         attachmentPoint = getValueFromJson(pnfReRegistrationFields, ATTACHMENT_POINT);
 
 122         remoteId = getValueFromJson(pnfReRegistrationFields, REMOTE_ID);
 
 123         cvlan = getValueFromJson(pnfReRegistrationFields, CVLAN);
 
 124         svlan = getValueFromJson(pnfReRegistrationFields, SVLAN);
 
 126         if (StringUtils.isEmpty(pnfCorrelationId) || anyImportantPropertyMissing()) {
 
 127             String incorrectEvent = dumpJsonData();
 
 128             LOGGER.warn("Incorrect Re-Registration JSON event: {}", incorrectEvent);
 
 132         return Mono.just(ImmutableReRegistrationConsumerDmaapModel.builder()
 
 133                 .correlationId(pnfCorrelationId)
 
 134                 .attachmentPoint(attachmentPoint)
 
 141     private boolean anyImportantPropertyMissing() {
 
 142         return StringUtils.isEmpty(attachmentPoint)
 
 143                 || StringUtils.isEmpty(remoteId)
 
 144                 || StringUtils.isEmpty(cvlan)
 
 145                 || StringUtils.isEmpty(svlan);
 
 148     private boolean containsProperHeaders(JsonObject jsonObject) {
 
 149         return jsonObject.has(ADDITIONAL_FIELDS);
 
 152     private String dumpJsonData() {
 
 153         return String.format(RE_REGISTRATION_DUMPING_TEMPLATE,
 
 162     Optional<JsonObject> getJsonObjectFromAnArray(JsonElement element) {
 
 163         JsonParser jsonParser = new JsonParser();
 
 164         return element.isJsonPrimitive() ? Optional.of(jsonParser.parse(element.getAsString()).getAsJsonObject())
 
 165                 : Optional.of(jsonParser.parse(element.toString()).getAsJsonObject());
 
 168     private String getValueFromJson(JsonObject jsonObject, String jsonKey) {
 
 169         return jsonObject.has(jsonKey) ? jsonObject.get(jsonKey).getAsString() : "";