wso2 · SanojPunchihewa · Dec 13, 2024 · Nov 4, 2024 · Nov 11, 2024 · Nov 13, 2024
@@ -626,4 +626,6 @@ public enum ENDPOINT_TIMEOUT_TYPE { ENDPOINT_TIMEOUT, GLOBAL_TIMEOUT, HTTP_CONNE
 
     public static final String ANALYTICS_METADATA = "ANALYTICS_METADATA";
 
+    public static final String SCATTER_MESSAGES = "SCATTER_MESSAGES";
+    public static final String CONTINUE_FLOW_TRIGGERED_FROM_MEDIATOR_WORKER = "CONTINUE_FLOW_TRIGGERED_FROM_MEDIATOR_WORKER";
 }
@@ -148,4 +148,17 @@ public static void tryEndFlow(MessageContext messageContext, String componentNam
 //			closeFlowForcefully(messageContext);
 		}
 	}
+
+	/**
+	 * This method will close the event collector and finish the flow when a Scatter Gather mediator is used.
+	 *
+	 * @param messageContext synapse message context.
+	 */
+	public static void closeEventsAfterScatterGather(MessageContext messageContext) {
+
+		if (isOpenTelemetryEnabled()) {
+			OpenTelemetryManagerHolder.getOpenTelemetryManager().getHandler()
+					.handleScatterGatherFinishEvent(messageContext);
+		}
+	}
 }
@@ -45,4 +45,11 @@ public interface CloseEventHandler {
      * @param synCtx                    Message context.
      */
     void handleTryEndFlow(BasicStatisticDataUnit basicStatisticDataUnit, MessageContext synCtx);
+
+    /**
+     * Handles a close flow event.
+     *
+     * @param synCtx Message context.
+     */
+    void handleScatterGatherFinishEvent(MessageContext synCtx);
 }
@@ -430,6 +430,25 @@ private void handleCallbackFinishEvent(MessageContext messageContext) {
         }
     }
 
+    public void handleScatterGatherFinishEvent(MessageContext messageContext) {
+        TracingScope tracingScope = tracingScopeManager.getTracingScope(messageContext);
+        synchronized (tracingScope.getSpanStore()) {
+            cleanupContinuationStateSequences(tracingScope.getSpanStore(), messageContext);
+            cleanUpActiveSpans(tracingScope.getSpanStore(), messageContext);
+            SpanWrapper outerLevelSpanWrapper = tracingScope.getSpanStore().getOuterLevelSpanWrapper();
+            tracingScope.getSpanStore().finishSpan(outerLevelSpanWrapper, messageContext);
+            tracingScopeManager.cleanupTracingScope(tracingScope.getTracingScopeId());
+        }
+    }
+
+    private void cleanUpActiveSpans(SpanStore spanStore, MessageContext messageContext) {
+        List<SpanWrapper> activeSpanWrappers = spanStore.getActiveSpanWrappers();
+        for (int i = activeSpanWrappers.size() - 1; i > 0; i--) {
+            SpanWrapper spanWrapper = activeSpanWrappers.get(i);
+            spanStore.finishSpan(spanWrapper, messageContext);
+        }
+    }
+
     @Override
     public void handleStateStackInsertion(MessageContext synCtx, String seqName, SequenceType seqType) {
         TracingScope tracingScope = tracingScopeManager.getTracingScope(synCtx);

@@ -103,7 +103,8 @@ public class MediatorFactoryFinder implements XMLToObjectMapper {
             ForEachMediatorFactory.class,
             JSONTransformMediatorFactory.class,
             NTLMMediatorFactory.class,
-            VariableMediatorFactory.class
+            VariableMediatorFactory.class,
+            ScatterGatherMediatorFactory.class
     };
 
     private final static MediatorFactoryFinder instance  = new MediatorFactoryFinder();

@@ -77,7 +77,8 @@ public class MediatorSerializerFinder {
             ForEachMediatorSerializer.class,
             JSONTransformMediatorSerializer.class,
             NTLMMediatorSerializer.class,
-            VariableMediatorSerializer.class
+            VariableMediatorSerializer.class,
+            ScatterGatherMediatorSerializer.class
     };
 
     private final static MediatorSerializerFinder instance = new MediatorSerializerFinder();

@@ -0,0 +1,185 @@
+/*
+ *  Copyright (c) 2024, WSO2 LLC. (http://www.wso2.org) All Rights Reserved.
+ *
+ *  WSO2 LLC. licenses this file to you 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.
+ */
+
+package org.apache.synapse.config.xml;
+
+import org.apache.axiom.om.OMAttribute;
+import org.apache.axiom.om.OMElement;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.synapse.Mediator;
+import org.apache.synapse.SynapseException;
+import org.apache.synapse.mediators.eip.Target;
+import org.apache.synapse.mediators.v2.ScatterGather;
+import org.jaxen.JaxenException;
+
+import java.util.Iterator;
+import java.util.Properties;
+import javax.xml.namespace.QName;
+
+/**
+ * The &lt;scatter-gather&gt; mediator is used to copy messages in Synapse to similar messages but with
+ * different message contexts and aggregate the responses back.
+ *
+ * <pre>
+ * &lt;scatter-gather parallel-execution=(true | false) result-target=(body | variable) content-type=(JSON | XML)&gt;
+ *   &lt;aggregation value="expression" condition="expression" timeout="long"
+ *     min-messages="expression" max-messages="expression"/&gt;
+ *   &lt;sequence&gt;
+ *     (mediator)+
+ *   &lt;/sequence&gt;+
+ * &lt;/scatter-gather&gt;
+ * </pre>
+ */
+public class ScatterGatherMediatorFactory extends AbstractMediatorFactory {
+
+    /**
+     * This will hold the QName of the clone mediator element in the xml configuration
+     */
+    private static final QName SCATTER_GATHER_Q
+            = new QName(XMLConfigConstants.SYNAPSE_NAMESPACE, "scatter-gather");
+    private static final QName ELEMENT_AGGREGATE_Q
+            = new QName(XMLConfigConstants.SYNAPSE_NAMESPACE, "aggregation");
+    private static final QName ATT_AGGREGATE_EXPRESSION = new QName("expression");
+    private static final QName ATT_CONDITION = new QName("condition");
+    private static final QName ATT_TIMEOUT = new QName("timeout");
+    private static final QName ATT_MIN_MESSAGES = new QName("min-messages");
+    private static final QName ATT_MAX_MESSAGES = new QName("max-messages");
+    private static final QName SEQUENCE_Q = new QName(XMLConfigConstants.SYNAPSE_NAMESPACE, "sequence");
+    private static final QName PARALLEL_EXEC_Q = new QName("parallel-execution");
+    private static final QName RESULT_TARGET_Q = new QName("result-target");
+    private static final QName ROOT_ELEMENT_Q = new QName("root-element");
+    private static final QName CONTENT_TYPE_Q = new QName("content-type");
+
+    private static final SequenceMediatorFactory fac = new SequenceMediatorFactory();
+
+    public Mediator createSpecificMediator(OMElement elem, Properties properties) {
+
+        boolean asynchronousExe = true;
+
+        ScatterGather mediator = new ScatterGather();
+        processAuditStatus(mediator, elem);
+
+        OMAttribute parallelExecAttr = elem.getAttribute(PARALLEL_EXEC_Q);
+        if (parallelExecAttr != null && parallelExecAttr.getAttributeValue().equals("false")) {
+            asynchronousExe = false;
+        }
+        mediator.setParallelExecution(asynchronousExe);
+
+        OMAttribute contentTypeAttr = elem.getAttribute(CONTENT_TYPE_Q);
+        if (contentTypeAttr == null || StringUtils.isBlank(contentTypeAttr.getAttributeValue())) {
+            String msg = "The 'content-type' attribute is required for the configuration of a Scatter Gather mediator";
+            throw new SynapseException(msg);
+        } else {
+            if ("JSON".equals(contentTypeAttr.getAttributeValue())) {
+                mediator.setContentType(ScatterGather.JSON_TYPE);
+            } else if ("XML".equals(contentTypeAttr.getAttributeValue())) {
+                OMAttribute rootElementAttr = elem.getAttribute(ROOT_ELEMENT_Q);
+                if (rootElementAttr != null && StringUtils.isNotBlank(rootElementAttr.getAttributeValue())) {
+                    mediator.setRootElementName(rootElementAttr.getAttributeValue());
+                    mediator.setContentType(ScatterGather.XML_TYPE);
+                } else {
+                    String msg = "The 'root-element' attribute is required for the configuration of a " +
+                            "Scatter Gather mediator when the 'content-type' is 'XML'";
+                    throw new SynapseException(msg);
+                }
+            } else {
+                String msg = "The 'content-type' attribute should be either 'JSON' or 'XML'";
+                throw new SynapseException(msg);
+            }
+        }
+
+        OMAttribute resultTargetAttr = elem.getAttribute(RESULT_TARGET_Q);
+        if (resultTargetAttr == null || StringUtils.isBlank(resultTargetAttr.getAttributeValue())) {
+            String msg = "The 'result-target' attribute is required for the configuration of a Scatter Gather mediator";
+            throw new SynapseException(msg);
+        } else {
+            mediator.setResultTarget(resultTargetAttr.getAttributeValue());
+        }
+
+        Iterator sequenceListElements = elem.getChildrenWithName(SEQUENCE_Q);
+        if (!sequenceListElements.hasNext()) {
+            String msg = "A 'sequence' element is required for the configuration of a Scatter Gather mediator";
+            throw new SynapseException(msg);
+        }
+        while (sequenceListElements.hasNext()) {
+            OMElement sequence = (OMElement) sequenceListElements.next();
+            if (sequence != null) {
+                Target target = new Target();
+                target.setSequence(fac.createAnonymousSequence(sequence, properties));
+                target.setAsynchronous(asynchronousExe);
+                mediator.addTarget(target);
+            }
+        }
+
+        OMElement aggregateElement = elem.getFirstChildWithName(ELEMENT_AGGREGATE_Q);
+        if (aggregateElement != null) {
+            OMAttribute aggregateExpr = aggregateElement.getAttribute(ATT_AGGREGATE_EXPRESSION);
+            if (aggregateExpr != null) {
+                try {
+                    mediator.setAggregationExpression(
+                            SynapsePathFactory.getSynapsePath(aggregateElement, ATT_AGGREGATE_EXPRESSION));
+                } catch (JaxenException e) {
+                    handleException("Unable to load the aggregating expression", e);
+                }
+            } else {
+                String msg = "The 'expression' attribute is required for the configuration of a Scatter Gather mediator";
+                throw new SynapseException(msg);
+            }
+
+            OMAttribute conditionExpr = aggregateElement.getAttribute(ATT_CONDITION);
+            if (conditionExpr != null) {
+                try {
+                    mediator.setCorrelateExpression(
+                            SynapsePathFactory.getSynapsePath(aggregateElement, ATT_CONDITION));
+                } catch (JaxenException e) {
+                    handleException("Unable to load the condition expression", e);
+                }
+            }
+
+            OMAttribute completeTimeout = aggregateElement.getAttribute(ATT_TIMEOUT);
+            if (completeTimeout != null) {
+                mediator.setCompletionTimeoutMillis(Long.parseLong(completeTimeout.getAttributeValue()));
+            }
+
+            OMAttribute minMessages = aggregateElement.getAttribute(ATT_MIN_MESSAGES);
+            if (minMessages != null) {
+                mediator.setMinMessagesToComplete(new ValueFactory().createValue("min-messages", aggregateElement));
+            }
+
+            OMAttribute maxMessages = aggregateElement.getAttribute(ATT_MAX_MESSAGES);
+            if (maxMessages != null) {
+                mediator.setMaxMessagesToComplete(new ValueFactory().createValue("max-messages", aggregateElement));
+            }
+        } else {
+            String msg = "The 'aggregation' element is required for the configuration of a Scatter Gather mediator";
+            throw new SynapseException(msg);
+        }
+        addAllCommentChildrenToList(elem, mediator.getCommentsList());
+        return mediator;
+    }
+
+    /**
+     * This method will implement the getTagQName method of the MediatorFactory interface
+     *
+     * @return QName of the clone element in xml configuration
+     */
+    public QName getTagQName() {
+
+        return SCATTER_GATHER_Q;
+    }
+}
@@ -0,0 +1,95 @@
+/*
+ *  Copyright (c) 2024, WSO2 LLC. (http://www.wso2.org) All Rights Reserved.
+ *
+ *  WSO2 LLC. licenses this file to you 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.
+ */
+
+package org.apache.synapse.config.xml;
+
+import org.apache.axiom.om.OMElement;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.synapse.Mediator;
+import org.apache.synapse.mediators.eip.Target;
+import org.apache.synapse.mediators.v2.ScatterGather;
+
+/**
+ * Serializer for {@link ScatterGather} instances.
+ */
+public class ScatterGatherMediatorSerializer extends AbstractMediatorSerializer {
+
+    public OMElement serializeSpecificMediator(Mediator m) {
+
+        ScatterGather scatterGatherMediator = null;
+        if (!(m instanceof ScatterGather)) {
+            handleException("Unsupported mediator passed in for serialization : " + m.getType());
+        } else {
+            scatterGatherMediator = (ScatterGather) m;
+        }
+
+        assert scatterGatherMediator != null;
+        OMElement scatterGatherElement = fac.createOMElement("scatter-gather", synNS);
+        saveTracingState(scatterGatherElement, scatterGatherMediator);
+
+        scatterGatherElement.addAttribute(fac.createOMAttribute(
+                "parallel-execution", nullNS, Boolean.toString(scatterGatherMediator.getParallelExecution())));
+        scatterGatherElement.addAttribute(fac.createOMAttribute(
+                "result-target", nullNS, scatterGatherMediator.getResultTarget()));
+        scatterGatherElement.addAttribute(fac.createOMAttribute(
+                "content-type", nullNS, scatterGatherMediator.getContentType()));
+        if (StringUtils.isNotBlank(scatterGatherMediator.getRootElementName())) {
+            scatterGatherElement.addAttribute(fac.createOMAttribute(
+                    "root-element", nullNS, scatterGatherMediator.getRootElementName()));
+        }
+
+        OMElement aggregationElement = fac.createOMElement("aggregation", synNS);
+
+        SynapsePathSerializer.serializePath(
+                scatterGatherMediator.getAggregationExpression(), aggregationElement, "expression");
+
+        if (scatterGatherMediator.getCorrelateExpression() != null) {
+            SynapsePathSerializer.serializePath(
+                    scatterGatherMediator.getAggregationExpression(), aggregationElement, "condition");
+        }
+
+        if (scatterGatherMediator.getCompletionTimeoutMillis() != 0) {
+            aggregationElement.addAttribute(fac.createOMAttribute(
+                    "timeout", nullNS, Long.toString(scatterGatherMediator.getCompletionTimeoutMillis())));
+        }
+        if (scatterGatherMediator.getMinMessagesToComplete() != null) {
+            new ValueSerializer().serializeValue(
+                    scatterGatherMediator.getMinMessagesToComplete(), "min-messages", aggregationElement);
+        }
+        if (scatterGatherMediator.getMaxMessagesToComplete() != null) {
+            new ValueSerializer().serializeValue(
+                    scatterGatherMediator.getMaxMessagesToComplete(), "max-messages", aggregationElement);
+        }
+        scatterGatherElement.addChild(aggregationElement);
+
+        for (Target target : scatterGatherMediator.getTargets()) {
+            if (target != null && target.getSequence() != null) {
+                SequenceMediatorSerializer serializer = new SequenceMediatorSerializer();
+                serializer.serializeAnonymousSequence(scatterGatherElement, target.getSequence());
+            }
+        }
+        serializeComments(scatterGatherElement, scatterGatherMediator.getCommentsList());
+
+        return scatterGatherElement;
+    }
+
+    public String getMediatorClassName() {
+
+        return ScatterGather.class.getName();
+    }
+}