remove remnants of discarded flows

Author: paed01

dist/activity/Activity.js

@@ -435,7 +435,7 @@ Activity.prototype.resume = function resume() {
 };
 
 /**
- * Discard the activity. Stops execution if running and discards outbound flows.
+ * Discard the activity. Stops execution if running; the activity leaves without taking any outbound flow.
  * @param {Record<string, any>} [discardContent] Optional content propagated with the discard
  * @returns {void}
  */
@@ -707,16 +707,11 @@ Activity.prototype._onInbound = function onInbound(routingKey, message) {
         message: content.message,
         inbound
       });
-    case 'flow.discard':
     case 'activity.discard':
       {
-        let discardSequence;
-        if (content.discardSequence) discardSequence = content.discardSequence.slice();
-        const context = {
-          inbound,
-          discardSequence
-        };
-        return this[K_FLAGS].isParallelGateway ? this.run(context) : this._runDiscard(context);
+        return this._runDiscard({
+          inbound
+        });
       }
   }
 };
@@ -768,7 +763,6 @@ Activity.prototype._onInboundEvent = function onInboundEvent(routingKey, message
       }
     case 'association.take':
     case 'flow.take':
-    case 'flow.discard':
       return inboundQ.queueMessage(fields, (0, _messageHelper.cloneContent)(content), properties);
   }
 };
@@ -972,7 +966,7 @@ Activity.prototype._onExecutionMessage = function onExecutionMessage(routingKey,
   switch (routingKey) {
     case 'execution.outbound.take':
       {
-        return this._doOutbound(message, false, (err, outbound) => {
+        return this._doOutbound(message, (err, outbound) => {
           message.ack();
           if (err) return this.emitFatal(err, content);
           broker.publish('run', 'run.execute.passthrough', (0, _messageHelper.cloneContent)(content, {
@@ -1027,13 +1021,13 @@ Activity.prototype._doRunLeave = function doRunLeave(message, isDiscarded, onOut
     properties
   } = message;
   const correlationId = properties.correlationId;
-  if (content.ignoreOutbound) {
+  if (isDiscarded || content.ignoreOutbound) {
     this.broker.publish('run', 'run.leave', (0, _messageHelper.cloneContent)(content), {
       correlationId
     });
     return onOutbound();
   }
-  return this._doOutbound((0, _messageHelper.cloneMessage)(message), isDiscarded, (err, outbound) => {
+  return this._doOutbound((0, _messageHelper.cloneMessage)(message), (err, outbound) => {
     if (err) {
       return this._publishEvent('error', {
         ...content,
@@ -1054,16 +1048,12 @@ Activity.prototype._doRunLeave = function doRunLeave(message, isDiscarded, onOut
 };
 
 /** @internal */
-Activity.prototype._doOutbound = function doOutbound(fromMessage, isDiscarded, callback) {
+Activity.prototype._doOutbound = function doOutbound(fromMessage, callback) {
   const outboundSequenceFlows = this[K_FLOWS].outboundSequenceFlows;
   if (!outboundSequenceFlows.length) return callback(null, []);
   const fromContent = fromMessage.content;
   let outboundFlows;
-  if (isDiscarded) {
-    outboundFlows = outboundSequenceFlows.map(flow => (0, _outboundEvaluator.formatFlowAction)(flow, {
-      action: 'discard'
-    }));
-  } else if (fromContent.outbound?.length) {
+  if (fromContent.outbound?.length) {
     outboundFlows = outboundSequenceFlows.map(flow => (0, _outboundEvaluator.formatFlowAction)(flow, fromContent.outbound.filter(f => f.id === flow.id).pop()));
   }
   if (outboundFlows) {

dist/flows/SequenceFlow.js

@@ -87,8 +87,9 @@ SequenceFlow.prototype.take = function take(content) {
 };
 
 /**
- * Discard the flow and publish flow.discard. Detects loops via discardSequence and emits
- * flow.looped instead when the target id is already in the sequence.
+ * Discard the flow and publish flow.discard.
+ *
+ * @deprecated The execution runtime no longer discards sequence flows, so this is a no-op during a run. It will be removed in a future version.
  * @param {Record<string, any>} [content]
  */
 SequenceFlow.prototype.discard = function discard(content = {}) {

dist/gateways/ParallelGateway.js

@@ -37,8 +37,7 @@ function ParallelGateway(activityDef, context) {
   const cachedPeers = context.getShakenPeers(id);
   if (cachedPeers) {
     for (const [flowId, sourceIds] of cachedPeers) {
-      let peer = peers.get(flowId);
-      if (!peer) peers.set(flowId, peer = new Set());
+      const peer = peers.get(flowId);
       for (const sourceId of sourceIds) peer.add(sourceId);
     }
     activity[K_PEERS_DISCOVERED] = true;
@@ -162,19 +161,15 @@ ParallelGatewayBehaviour.prototype._onPeerEnterMessage = function onPeerEnterMes
   if (peer) this.peerMonitor.running.set(message.content.id, peer);
 };
 ParallelGatewayBehaviour.prototype._complete = function complete() {
-  const take = this.peerMonitor.inbound.some(({
-    action
-  }) => action === 'take');
   this.broker.cancel('_converging-inbound', false);
   this._stop();
-  const state = take ? 'completed' : 'discard';
-  this.activity.logger.debug(`<${this.executionId} (${this.id})> completed monitoring with state: ${state}`);
+  this.activity.logger.debug(`<${this.executionId} (${this.id})> completed monitoring`);
   const content = (0, _messageHelper.cloneContent)(this[_constants.K_EXECUTE_MESSAGE].content, {
     isRootScope: true,
-    state
+    state: 'completed'
   });
   content.inbound = this.peerMonitor.inbound;
-  return this.broker.publish('execution', `execute.${state}`, content);
+  return this.broker.publish('execution', 'execute.completed', content);
 };
 ParallelGatewayBehaviour.prototype._stop = function stop() {
   this.broker.cancel('_converging-inbound');

dist/messageHelper.js

@@ -17,7 +17,6 @@ exports.unshiftParent = unshiftParent;
  */
 function cloneContent(content, extend) {
   const {
-    discardSequence,
     inbound,
     outbound,
     parent,
@@ -32,9 +31,6 @@ function cloneContent(content, extend) {
   if (parent) {
     clone.parent = cloneParent(parent);
   }
-  if (discardSequence) {
-    clone.discardSequence = discardSequence.slice();
-  }
   if (inbound) {
     clone.inbound = inbound.map(c => cloneContent(c));
   }

dist/process/ProcessExecution.js

@@ -712,7 +712,6 @@ ProcessExecution.prototype._onChildMessage = function onChildMessage(routingKey,
         if (!prevMsg) return message.ack();
         break;
       }
-    case 'flow.looped':
     case 'activity.leave':
       return this._onChildCompleted(message);
   }

src/activity/Activity.js

@@ -428,7 +428,7 @@ Activity.prototype.resume = function resume() {
 };
 
 /**
- * Discard the activity. Stops execution if running and discards outbound flows.
+ * Discard the activity. Stops execution if running; the activity leaves without taking any outbound flow.
  * @param {Record<string, any>} [discardContent] Optional content propagated with the discard
  * @returns {void}
  */
@@ -677,12 +677,8 @@ Activity.prototype._onInbound = function onInbound(routingKey, message) {
         message: content.message,
         inbound,
       });
-    case 'flow.discard':
     case 'activity.discard': {
-      let discardSequence;
-      if (content.discardSequence) discardSequence = content.discardSequence.slice();
-      const context = { inbound, discardSequence };
-      return this[K_FLAGS].isParallelGateway ? this.run(context) : this._runDiscard(context);
+      return this._runDiscard({ inbound });
     }
   }
 };
@@ -721,7 +717,6 @@ Activity.prototype._onInboundEvent = function onInboundEvent(routingKey, message
     }
     case 'association.take':
     case 'flow.take':
-    case 'flow.discard':
       return inboundQ.queueMessage(fields, cloneContent(content), properties);
   }
 };
@@ -918,7 +913,7 @@ Activity.prototype._onExecutionMessage = function onExecutionMessage(routingKey,
 
   switch (routingKey) {
     case 'execution.outbound.take': {
-      return this._doOutbound(message, false, (err, outbound) => {
+      return this._doOutbound(message, (err, outbound) => {
         message.ack();
         if (err) return this.emitFatal(err, content);
         broker.publish('run', 'run.execute.passthrough', cloneContent(content, { outbound }));
@@ -958,12 +953,12 @@ Activity.prototype._ackRunExecuteMessage = function ackRunExecuteMessage() {
 Activity.prototype._doRunLeave = function doRunLeave(message, isDiscarded, onOutbound) {
   const { content, properties } = message;
   const correlationId = properties.correlationId;
-  if (content.ignoreOutbound) {
+  if (isDiscarded || content.ignoreOutbound) {
     this.broker.publish('run', 'run.leave', cloneContent(content), { correlationId });
     return onOutbound();
   }
 
-  return this._doOutbound(cloneMessage(message), isDiscarded, (err, outbound) => {
+  return this._doOutbound(cloneMessage(message), (err, outbound) => {
     if (err) {
       return this._publishEvent('error', { ...content, error: err }, { correlationId });
     }
@@ -982,16 +977,14 @@ Activity.prototype._doRunLeave = function doRunLeave(message, isDiscarded, onOut
 };
 
 /** @internal */
-Activity.prototype._doOutbound = function doOutbound(fromMessage, isDiscarded, callback) {
+Activity.prototype._doOutbound = function doOutbound(fromMessage, callback) {
   const outboundSequenceFlows = this[K_FLOWS].outboundSequenceFlows;
   if (!outboundSequenceFlows.length) return callback(null, []);
 
   const fromContent = fromMessage.content;
 
   let outboundFlows;
-  if (isDiscarded) {
-    outboundFlows = outboundSequenceFlows.map((flow) => formatFlowAction(flow, { action: 'discard' }));
-  } else if (fromContent.outbound?.length) {
+  if (fromContent.outbound?.length) {
     outboundFlows = outboundSequenceFlows.map((flow) => formatFlowAction(flow, fromContent.outbound.filter((f) => f.id === flow.id).pop()));
   }
 

src/flows/SequenceFlow.js

@@ -68,8 +68,9 @@ SequenceFlow.prototype.take = function take(content) {
 };
 
 /**
- * Discard the flow and publish flow.discard. Detects loops via discardSequence and emits
- * flow.looped instead when the target id is already in the sequence.
+ * Discard the flow and publish flow.discard.
+ *
+ * @deprecated The execution runtime no longer discards sequence flows, so this is a no-op during a run. It will be removed in a future version.
  * @param {Record<string, any>} [content]
  */
 SequenceFlow.prototype.discard = function discard(content = {}) {

src/gateways/ParallelGateway.js

@@ -177,20 +177,16 @@ ParallelGatewayBehaviour.prototype._onPeerEnterMessage = function onPeerEnterMes
 };
 
 ParallelGatewayBehaviour.prototype._complete = function complete() {
-  const take = this.peerMonitor.inbound.some(({ action }) => action === 'take');
-
   this.broker.cancel('_converging-inbound', false);
 
   this._stop();
 
-  const state = take ? 'completed' : 'discard';
-
-  this.activity.logger.debug(`<${this.executionId} (${this.id})> completed monitoring with state: ${state}`);
+  this.activity.logger.debug(`<${this.executionId} (${this.id})> completed monitoring`);
 
-  const content = cloneContent(this[K_EXECUTE_MESSAGE].content, { isRootScope: true, state });
+  const content = cloneContent(this[K_EXECUTE_MESSAGE].content, { isRootScope: true, state: 'completed' });
   content.inbound = this.peerMonitor.inbound;
 
-  return this.broker.publish('execution', `execute.${state}`, content);
+  return this.broker.publish('execution', 'execute.completed', content);
 };
 
 ParallelGatewayBehaviour.prototype._stop = function stop() {

src/messageHelper.js

@@ -5,7 +5,7 @@
  * @returns cloned content
  */
 export function cloneContent(content, extend) {
-  const { discardSequence, inbound, outbound, parent, sequence } = content;
+  const { inbound, outbound, parent, sequence } = content;
 
   /** @type {import('#types').ElementMessageContent} */
   const clone = {
@@ -16,9 +16,6 @@ export function cloneContent(content, extend) {
   if (parent) {
     clone.parent = cloneParent(parent);
   }
-  if (discardSequence) {
-    clone.discardSequence = discardSequence.slice();
-  }
   if (inbound) {
     clone.inbound = inbound.map((c) => cloneContent(c));
   }

src/process/ProcessExecution.js

@@ -710,7 +710,6 @@ ProcessExecution.prototype._onChildMessage = function onChildMessage(routingKey,
       if (!prevMsg) return message.ack();
       break;
     }
-    case 'flow.looped':
     case 'activity.leave':
       return this._onChildCompleted(message);
   }