2 * ===============================LICENSE_START======================================
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
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.dcae.apod.analytics.common.utils;
23 import com.google.common.base.Preconditions;
24 import org.onap.dcae.apod.analytics.common.exception.MessageProcessingException;
25 import org.onap.dcae.apod.analytics.common.service.filter.GenericJsonMessageFilter;
26 import org.onap.dcae.apod.analytics.common.service.filter.JsonMessageFilterProcessorContext;
27 import org.onap.dcae.apod.analytics.common.service.processor.GenericMessageChainProcessor;
28 import org.onap.dcae.apod.analytics.common.service.processor.MessageProcessor;
29 import org.onap.dcae.apod.analytics.common.service.processor.ProcessorContext;
30 import org.slf4j.Logger;
31 import org.slf4j.LoggerFactory;
33 import java.util.Iterator;
34 import java.util.LinkedList;
35 import java.util.List;
39 import javax.annotation.Nonnull;
44 * @author Rajiv Singla . Creation Date: 11/8/2016.
46 public abstract class MessageProcessorUtils {
48 private static final Logger LOG = LoggerFactory.getLogger(MessageProcessorUtils.class);
51 * Provides an abstraction how to apply {@link ProcessorContext} to next {@link MessageProcessor}
52 * in the message processor chain
54 * @param <P> Sub classes of Processor Context
56 public interface MessageProcessorFunction<P extends ProcessorContext> {
59 * Method which provides accumulated {@link ProcessorContext} from previous processors and a reference
60 * to next processor in the chain
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
66 * @return processing context after computing the current Message Processor
68 <M extends MessageProcessor<P>> P apply(P p, M m);
73 * Provides an abstraction to compute a chain of {@link MessageProcessor}
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
80 * @return processing context which results after computing the whole chain
82 public static <P extends ProcessorContext> P computeMessageProcessorChain(
83 final Iterable<? extends MessageProcessor<P>> messageProcessors,
84 final P initialProcessorContext,
85 final MessageProcessorFunction<P> messageProcessorFunction) {
87 // Get message processor iterator
88 final Iterator<? extends MessageProcessor<P>> processorIterator = messageProcessors.iterator();
90 // If no next message processor - return initial processor context
91 if (!processorIterator.hasNext()) {
92 return initialProcessorContext;
95 // An accumulator for processor Context
96 P processorContextAccumulator = initialProcessorContext;
98 while (processorIterator.hasNext()) {
100 final MessageProcessor<P> nextProcessor = processorIterator.next();
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));
111 if (!processorContextAccumulator.canProcessingContinue()) {
112 LOG.debug("Triggering Early Termination, before Message Processor: {}, Incoming Message: {}",
113 nextProcessor.getProcessorInfo().getProcessorName(), processorContextAccumulator.getMessage());
116 processorContextAccumulator = messageProcessorFunction.apply(processorContextAccumulator, nextProcessor);
119 return processorContextAccumulator;
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}
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
131 * @return json message processor context which contains the {@link JsonMessageFilterProcessorContext#isMatched}
132 * status after applying all filter mappings
134 public static JsonMessageFilterProcessorContext processJsonFilterMappings(
135 final String jsonMessage, @Nonnull final Map<String, Set<String>> jsonFilterMappings) {
137 Preconditions.checkState(jsonFilterMappings.size() > 0, "Json Filter Mappings must not be empty");
139 // create initial processor context containing the json message that need to be processed
140 final JsonMessageFilterProcessorContext initialProcessorContext =
141 new JsonMessageFilterProcessorContext(jsonMessage);
143 // Create Json Message Filters
144 final List<GenericJsonMessageFilter> jsonMessageFilters = new LinkedList<>();
147 for (Map.Entry<String, Set<String>> jsonFilterMapping : jsonFilterMappings.entrySet()) {
148 jsonMessageFilters.add(new GenericJsonMessageFilter("Filter-" + i, jsonFilterMapping.getKey(),
149 jsonFilterMapping.getValue()));
153 // Create Generic Message Chain Processor
154 final GenericMessageChainProcessor<JsonMessageFilterProcessorContext> messageChainProcessor =
155 new GenericMessageChainProcessor<>(jsonMessageFilters, initialProcessorContext);
157 // Process chain and return resulting json Message Filter Processor Context
158 return messageChainProcessor.processChain();