-/*
- * ===============================LICENSE_START======================================
- * dcae-analytics
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============================LICENSE_END===========================================
- */
-
-package org.openecomp.dcae.apod.analytics.common.utils;
-
-import com.google.common.base.Preconditions;
-import org.openecomp.dcae.apod.analytics.common.exception.MessageProcessingException;
-import org.openecomp.dcae.apod.analytics.common.service.filter.GenericJsonMessageFilter;
-import org.openecomp.dcae.apod.analytics.common.service.filter.JsonMessageFilterProcessorContext;
-import org.openecomp.dcae.apod.analytics.common.service.processor.GenericMessageChainProcessor;
-import org.openecomp.dcae.apod.analytics.common.service.processor.MessageProcessor;
-import org.openecomp.dcae.apod.analytics.common.service.processor.ProcessorContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import javax.annotation.Nonnull;
-
-
-/**
- *
- * @author Rajiv Singla . Creation Date: 11/8/2016.
- */
-public abstract class MessageProcessorUtils {
-
- private static final Logger LOG = LoggerFactory.getLogger(MessageProcessorUtils.class);
-
- /**
- * Provides an abstraction how to apply {@link ProcessorContext} to next {@link MessageProcessor}
- * in the message processor chain
- *
- * @param <P> Sub classes of Processor Context
- */
- public interface MessageProcessorFunction<P extends ProcessorContext> {
-
- /**
- * Method which provides accumulated {@link ProcessorContext} from previous processors and a reference
- * to next processor in the chain
- *
- * @param p accumulated {@link ProcessorContext} from previous processors
- * @param m current {@link MessageProcessor} in the chain
- * @param <M> Message processor sub classes
- *
- * @return processing context after computing the current Message Processor
- */
- <M extends MessageProcessor<P>> P apply(P p, M m);
- }
-
-
- /**
- * Provides an abstraction to compute a chain of {@link MessageProcessor}
- *
- * @param messageProcessors An iterable containing one or more {@link MessageProcessor}s
- * @param initialProcessorContext An initial processing Context
- * @param messageProcessorFunction messageProcessor Function
- * @param <P> Sub classes for Processor Context
- *
- * @return processing context which results after computing the whole chain
- */
- public static <P extends ProcessorContext> P computeMessageProcessorChain(
- final Iterable<? extends MessageProcessor<P>> messageProcessors,
- final P initialProcessorContext,
- final MessageProcessorFunction<P> messageProcessorFunction) {
-
- // Get message processor iterator
- final Iterator<? extends MessageProcessor<P>> processorIterator = messageProcessors.iterator();
-
- // If no next message processor - return initial processor context
- if (!processorIterator.hasNext()) {
- return initialProcessorContext;
- }
-
- // An accumulator for processor Context
- P processorContextAccumulator = initialProcessorContext;
-
- while (processorIterator.hasNext()) {
-
- final MessageProcessor<P> nextProcessor = processorIterator.next();
-
- // If Initial Processor Context is null
- if (processorContextAccumulator == null) {
- final String errorMessage =
- String.format("Processor Context must not be null for Message Process: %s",
- nextProcessor.getProcessorInfo().getProcessorName());
- throw new MessageProcessingException(errorMessage, LOG, new IllegalStateException(errorMessage));
- }
-
-
- if (!processorContextAccumulator.canProcessingContinue()) {
- LOG.debug("Triggering Early Termination, before Message Processor: {}, Incoming Message: {}",
- nextProcessor.getProcessorInfo().getProcessorName(), processorContextAccumulator.getMessage());
- break;
- }
- processorContextAccumulator = messageProcessorFunction.apply(processorContextAccumulator, nextProcessor);
- }
-
- return processorContextAccumulator;
- }
-
-
- /**
- * Utility method to process Json Filter Mappings. Processes incoming json message and applies a list of json
- * filter mappings and returns the resulting {@link JsonMessageFilterProcessorContext}
- *
- * @param jsonMessage json message to which filter mappings will be applies
- * @param jsonFilterMappings Filter mappings contains a Map containing keys as filter json path
- * and values as set of expected value corresponding to filter path
- *
- * @return json message processor context which contains the {@link JsonMessageFilterProcessorContext#isMatched}
- * status after applying all filter mappings
- */
- public static JsonMessageFilterProcessorContext processJsonFilterMappings(
- final String jsonMessage, @Nonnull final Map<String, Set<String>> jsonFilterMappings) {
-
- Preconditions.checkState(jsonFilterMappings.size() > 0, "Json Filter Mappings must not be empty");
-
- // create initial processor context containing the json message that need to be processed
- final JsonMessageFilterProcessorContext initialProcessorContext =
- new JsonMessageFilterProcessorContext(jsonMessage);
-
- // Create Json Message Filters
- final List<GenericJsonMessageFilter> jsonMessageFilters = new LinkedList<>();
-
- int i = 0;
- for (Map.Entry<String, Set<String>> jsonFilterMapping : jsonFilterMappings.entrySet()) {
- jsonMessageFilters.add(new GenericJsonMessageFilter("Filter-" + i, jsonFilterMapping.getKey(),
- jsonFilterMapping.getValue()));
- i++;
- }
-
- // Create Generic Message Chain Processor
- final GenericMessageChainProcessor<JsonMessageFilterProcessorContext> messageChainProcessor =
- new GenericMessageChainProcessor<>(jsonMessageFilters, initialProcessorContext);
-
- // Process chain and return resulting json Message Filter Processor Context
- return messageChainProcessor.processChain();
- }
-
-
-}
+/*\r
+ * ===============================LICENSE_START======================================\r
+ * dcae-analytics\r
+ * ================================================================================\r
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
+ * ================================================================================\r
+ * Licensed under the Apache License, Version 2.0 (the "License");\r
+ * you may not use this file except in compliance with the License.\r
+ * You may obtain a copy of the License at\r
+ *\r
+ * http://www.apache.org/licenses/LICENSE-2.0\r
+ *\r
+ * Unless required by applicable law or agreed to in writing, software\r
+ * distributed under the License is distributed on an "AS IS" BASIS,\r
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+ * See the License for the specific language governing permissions and\r
+ * limitations under the License.\r
+ * ============================LICENSE_END===========================================\r
+ */\r
+\r
+package org.openecomp.dcae.apod.analytics.common.utils;\r
+\r
+import com.google.common.base.Preconditions;\r
+import org.openecomp.dcae.apod.analytics.common.exception.MessageProcessingException;\r
+import org.openecomp.dcae.apod.analytics.common.service.filter.GenericJsonMessageFilter;\r
+import org.openecomp.dcae.apod.analytics.common.service.filter.JsonMessageFilterProcessorContext;\r
+import org.openecomp.dcae.apod.analytics.common.service.processor.GenericMessageChainProcessor;\r
+import org.openecomp.dcae.apod.analytics.common.service.processor.MessageProcessor;\r
+import org.openecomp.dcae.apod.analytics.common.service.processor.ProcessorContext;\r
+import org.slf4j.Logger;\r
+import org.slf4j.LoggerFactory;\r
+\r
+import java.util.Iterator;\r
+import java.util.LinkedList;\r
+import java.util.List;\r
+import java.util.Map;\r
+import java.util.Set;\r
+\r
+import javax.annotation.Nonnull;\r
+\r
+\r
+/**\r
+ *\r
+ * @author Rajiv Singla . Creation Date: 11/8/2016.\r
+ */\r
+public abstract class MessageProcessorUtils {\r
+\r
+ private static final Logger LOG = LoggerFactory.getLogger(MessageProcessorUtils.class);\r
+\r
+ /**\r
+ * Provides an abstraction how to apply {@link ProcessorContext} to next {@link MessageProcessor}\r
+ * in the message processor chain\r
+ *\r
+ * @param <P> Sub classes of Processor Context\r
+ */\r
+ public interface MessageProcessorFunction<P extends ProcessorContext> {\r
+\r
+ /**\r
+ * Method which provides accumulated {@link ProcessorContext} from previous processors and a reference\r
+ * to next processor in the chain\r
+ *\r
+ * @param p accumulated {@link ProcessorContext} from previous processors\r
+ * @param m current {@link MessageProcessor} in the chain\r
+ * @param <M> Message processor sub classes\r
+ *\r
+ * @return processing context after computing the current Message Processor\r
+ */\r
+ <M extends MessageProcessor<P>> P apply(P p, M m);\r
+ }\r
+\r
+\r
+ /**\r
+ * Provides an abstraction to compute a chain of {@link MessageProcessor}\r
+ *\r
+ * @param messageProcessors An iterable containing one or more {@link MessageProcessor}s\r
+ * @param initialProcessorContext An initial processing Context\r
+ * @param messageProcessorFunction messageProcessor Function\r
+ * @param <P> Sub classes for Processor Context\r
+ *\r
+ * @return processing context which results after computing the whole chain\r
+ */\r
+ public static <P extends ProcessorContext> P computeMessageProcessorChain(\r
+ final Iterable<? extends MessageProcessor<P>> messageProcessors,\r
+ final P initialProcessorContext,\r
+ final MessageProcessorFunction<P> messageProcessorFunction) {\r
+\r
+ // Get message processor iterator\r
+ final Iterator<? extends MessageProcessor<P>> processorIterator = messageProcessors.iterator();\r
+\r
+ // If no next message processor - return initial processor context\r
+ if (!processorIterator.hasNext()) {\r
+ return initialProcessorContext;\r
+ }\r
+\r
+ // An accumulator for processor Context\r
+ P processorContextAccumulator = initialProcessorContext;\r
+\r
+ while (processorIterator.hasNext()) {\r
+\r
+ final MessageProcessor<P> nextProcessor = processorIterator.next();\r
+\r
+ // If Initial Processor Context is null\r
+ if (processorContextAccumulator == null) {\r
+ final String errorMessage =\r
+ String.format("Processor Context must not be null for Message Process: %s",\r
+ nextProcessor.getProcessorInfo().getProcessorName());\r
+ throw new MessageProcessingException(errorMessage, LOG, new IllegalStateException(errorMessage));\r
+ }\r
+\r
+\r
+ if (!processorContextAccumulator.canProcessingContinue()) {\r
+ LOG.debug("Triggering Early Termination, before Message Processor: {}, Incoming Message: {}",\r
+ nextProcessor.getProcessorInfo().getProcessorName(), processorContextAccumulator.getMessage());\r
+ break;\r
+ }\r
+ processorContextAccumulator = messageProcessorFunction.apply(processorContextAccumulator, nextProcessor);\r
+ }\r
+\r
+ return processorContextAccumulator;\r
+ }\r
+\r
+\r
+ /**\r
+ * Utility method to process Json Filter Mappings. Processes incoming json message and applies a list of json\r
+ * filter mappings and returns the resulting {@link JsonMessageFilterProcessorContext}\r
+ *\r
+ * @param jsonMessage json message to which filter mappings will be applies\r
+ * @param jsonFilterMappings Filter mappings contains a Map containing keys as filter json path\r
+ * and values as set of expected value corresponding to filter path\r
+ *\r
+ * @return json message processor context which contains the {@link JsonMessageFilterProcessorContext#isMatched}\r
+ * status after applying all filter mappings\r
+ */\r
+ public static JsonMessageFilterProcessorContext processJsonFilterMappings(\r
+ final String jsonMessage, @Nonnull final Map<String, Set<String>> jsonFilterMappings) {\r
+\r
+ Preconditions.checkState(jsonFilterMappings.size() > 0, "Json Filter Mappings must not be empty");\r
+\r
+ // create initial processor context containing the json message that need to be processed\r
+ final JsonMessageFilterProcessorContext initialProcessorContext =\r
+ new JsonMessageFilterProcessorContext(jsonMessage);\r
+\r
+ // Create Json Message Filters\r
+ final List<GenericJsonMessageFilter> jsonMessageFilters = new LinkedList<>();\r
+\r
+ int i = 0;\r
+ for (Map.Entry<String, Set<String>> jsonFilterMapping : jsonFilterMappings.entrySet()) {\r
+ jsonMessageFilters.add(new GenericJsonMessageFilter("Filter-" + i, jsonFilterMapping.getKey(),\r
+ jsonFilterMapping.getValue()));\r
+ i++;\r
+ }\r
+\r
+ // Create Generic Message Chain Processor\r
+ final GenericMessageChainProcessor<JsonMessageFilterProcessorContext> messageChainProcessor =\r
+ new GenericMessageChainProcessor<>(jsonMessageFilters, initialProcessorContext);\r
+\r
+ // Process chain and return resulting json Message Filter Processor Context\r
+ return messageChainProcessor.processChain();\r
+ }\r
+\r
+\r
+}\r
Code Review / dcaegen2 / analytics / tca.git / blobdiff