-/*\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
+/*
+ * ===============================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.onap.dcae.apod.analytics.common.utils;
+
+import com.google.common.base.Preconditions;
+import org.onap.dcae.apod.analytics.common.exception.MessageProcessingException;
+import org.onap.dcae.apod.analytics.common.service.filter.GenericJsonMessageFilter;
+import org.onap.dcae.apod.analytics.common.service.filter.JsonMessageFilterProcessorContext;
+import org.onap.dcae.apod.analytics.common.service.processor.GenericMessageChainProcessor;
+import org.onap.dcae.apod.analytics.common.service.processor.MessageProcessor;
+import org.onap.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();
+ }
+
+
+}
Code Review / dcaegen2 / analytics / tca.git / blobdiff