dcae-analytics-common/src/main/java/org/openecomp/dcae/apod/analytics/common/utils/MessageProcessorUtils.java

   1 /*
   2  * ===============================LICENSE_START======================================
   3  *  dcae-analytics
   4  * ================================================================================
   5  *    Copyright © 2017 AT&T 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
  10  *
  11  *          http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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===========================================
  19  */
  20
  21 package org.openecomp.dcae.apod.analytics.common.utils;
  22
  23 import com.google.common.base.Preconditions;
  24 import org.openecomp.dcae.apod.analytics.common.exception.MessageProcessingException;
  25 import org.openecomp.dcae.apod.analytics.common.service.filter.GenericJsonMessageFilter;
  26 import org.openecomp.dcae.apod.analytics.common.service.filter.JsonMessageFilterProcessorContext;
  27 import org.openecomp.dcae.apod.analytics.common.service.processor.GenericMessageChainProcessor;
  28 import org.openecomp.dcae.apod.analytics.common.service.processor.MessageProcessor;
  29 import org.openecomp.dcae.apod.analytics.common.service.processor.ProcessorContext;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 import java.util.Iterator;
  34 import java.util.LinkedList;
  35 import java.util.List;
  36 import java.util.Map;
  37 import java.util.Set;
  38
  39 import javax.annotation.Nonnull;
  40
  41
  42 /**
  43  *
  44  * @author Rajiv Singla . Creation Date: 11/8/2016.
  45  */
  46 public abstract class MessageProcessorUtils {
  47
  48     private static final Logger LOG = LoggerFactory.getLogger(MessageProcessorUtils.class);
  49
  50     /**
  51      * Provides an abstraction how to apply {@link ProcessorContext} to next {@link MessageProcessor}
  52      * in the message processor chain
  53      *
  54      * @param <P> Sub classes of Processor Context
  55      */
  56     public interface MessageProcessorFunction<P extends ProcessorContext> {
  57
  58         /**
  59          * Method which provides accumulated {@link ProcessorContext} from previous processors and a reference
  60          * to next processor in the chain
  61          *
  62          * @param p accumulated {@link ProcessorContext} from previous processors
  63          * @param m current {@link MessageProcessor} in the chain
  64          * @param <M> Message processor sub classes
  65          *
  66          * @return processing context after computing the current Message Processor
  67          */
  68         <M extends MessageProcessor<P>> P apply(P p, M m);
  69     }
  70
  71
  72     /**
  73      * Provides an abstraction to compute a chain of {@link MessageProcessor}
  74      *
  75      * @param messageProcessors An iterable containing one or more {@link MessageProcessor}s
  76      * @param initialProcessorContext An initial processing Context
  77      * @param messageProcessorFunction messageProcessor Function
  78      * @param <P> Sub classes for Processor Context
  79      *
  80      * @return processing context which results after computing the whole chain
  81      */
  82     public static <P extends ProcessorContext> P computeMessageProcessorChain(
  83             final Iterable<? extends MessageProcessor<P>> messageProcessors,
  84             final P initialProcessorContext,
  85             final MessageProcessorFunction<P> messageProcessorFunction) {
  86
  87         // Get message processor iterator
  88         final Iterator<? extends MessageProcessor<P>> processorIterator = messageProcessors.iterator();
  89
  90         // If no next message processor - return initial processor context
  91         if (!processorIterator.hasNext()) {
  92             return initialProcessorContext;
  93         }
  94
  95         // An accumulator for processor Context
  96         P processorContextAccumulator = initialProcessorContext;
  97
  98         while (processorIterator.hasNext()) {
  99
 100             final MessageProcessor<P> nextProcessor = processorIterator.next();
 101
 102             // If Initial Processor Context is null
 103             if (processorContextAccumulator == null) {
 104                 final String errorMessage =
 105                         String.format("Processor Context must not be null for Message Process: %s",
 106                                 nextProcessor.getProcessorInfo().getProcessorName());
 107                 throw new MessageProcessingException(errorMessage, LOG, new IllegalStateException(errorMessage));
 108             }
 109
 110
 111             if (!processorContextAccumulator.canProcessingContinue()) {
 112                 LOG.debug("Triggering Early Termination, before Message Processor: {}, Incoming Message: {}",
 113                         nextProcessor.getProcessorInfo().getProcessorName(), processorContextAccumulator.getMessage());
 114                 break;
 115             }
 116             processorContextAccumulator = messageProcessorFunction.apply(processorContextAccumulator, nextProcessor);
 117         }
 118
 119         return processorContextAccumulator;
 120     }
 121
 122
 123     /**
 124      * Utility method to process Json Filter Mappings. Processes incoming json message and applies a list of json
 125      * filter mappings and returns the resulting {@link JsonMessageFilterProcessorContext}
 126      *
 127      * @param jsonMessage json message to which filter mappings will be applies
 128      * @param jsonFilterMappings Filter mappings contains a Map containing keys as filter json path
 129      * and values as set of expected value corresponding to filter path
 130      *
 131      * @return json message processor context which contains the {@link JsonMessageFilterProcessorContext#isMatched}
 132      * status after applying all filter mappings
 133      */
 134     public static JsonMessageFilterProcessorContext processJsonFilterMappings(
 135             final String jsonMessage, @Nonnull final Map<String, Set<String>> jsonFilterMappings) {
 136
 137         Preconditions.checkState(jsonFilterMappings.size() > 0, "Json Filter Mappings must not be empty");
 138
 139         // create initial processor context containing the json message that need to be processed
 140         final JsonMessageFilterProcessorContext initialProcessorContext =
 141                 new JsonMessageFilterProcessorContext(jsonMessage);
 142
 143         // Create Json Message Filters
 144         final List<GenericJsonMessageFilter> jsonMessageFilters = new LinkedList<>();
 145
 146         int i = 0;
 147         for (Map.Entry<String, Set<String>> jsonFilterMapping : jsonFilterMappings.entrySet()) {
 148             jsonMessageFilters.add(new GenericJsonMessageFilter("Filter-" + i, jsonFilterMapping.getKey(),
 149                     jsonFilterMapping.getValue()));
 150             i++;
 151         }
 152
 153         // Create Generic Message Chain Processor
 154         final GenericMessageChainProcessor<JsonMessageFilterProcessorContext> messageChainProcessor =
 155                 new GenericMessageChainProcessor<>(jsonMessageFilters, initialProcessorContext);
 156
 157         // Process chain and return resulting json Message Filter Processor Context
 158         return messageChainProcessor.processChain();
 159     }
 160
 161
 162 }