Env Function

README.md

@@ -1725,9 +1725,67 @@ All JSONata functions are available in Stated.
 ## Stated functions
 Stated provides many functions not provided out of the box by JSONata. 
 
-### $timeout
+### $env
+`$env` is a function used to read environment variables for Stated templates running in node.js. A default value can 
+be specified. If there is no default provided and no such variable exists in node `process.env` an error is returned.
+```json
+> .init -f example/ex24.json
+{
+  "myVar": "${ $env('MY_VAR', 'default value') }"
+}
+> .out
+{
+  "myVar": "default value"
+}
+
+```
+
+### $setTimeout
+`$setTimeout` is the JavaScript `setTimeout` function. It receives a function and an timeout
+expressed as milliseconds. The function is called after the timeout. Example `ex15.json` demonstrates an 
+intersting use of `$setTimout`. The `counter` variable is updates, which causes `upCount$` to re-evaluate due to
+its reactive dependency on `counter`. In this way, each time `counter` changes, it schedules its next change 1 second
+in the future.:
+```json
+{
+  "counter": 0,
+  "upCount$": "counter<10 ? ($setTimeout(function(){$set('/counter', counter+1)}, 1000);'counting'):'done'"
+}
+```
+
+```json ["data.counter=3"]
+> .init -f example/ex15.json --tail "/ until counter=3"
+Started tailing... Press Ctrl+C to stop.
+{
+  "counter": 3,
+  "upCount$": "counting"
+}
+>
+```
 
-### $interval
+### $setInterval
+`$setInterval` is the JavaScript `setInterval` function. It receives a function and an interval
+expressed as milliseconds. The function is called every inteval. Example `ex14.json` shows how reaction
+to `counter` changing is used to clear the interval when count exceeds 10.:
+```json
+{
+  "incr": "${function(){$set('/counter',counter+1)}}",
+  "counter": 0,
+  "upCount": "${  $setInterval(incr, 1000)  }",
+  "status": "${(counter>10?($clearInterval(upCount);'done'):'counting')}"
+}
+```
+```json ["data.counter=3"]
+> .init -f example/ex14.json --tail "/ until counter=3"
+Started tailing... Press Ctrl+C to stop.
+{
+  "incr": "{function:}",
+  "counter": 3,
+  "upCount": "--interval/timeout--",
+  "status": "counting"
+}
+
+```
 
 ### $fetch
 JSONata provides the standard JS fetch function. You can use it exactly as you would in JS, except that

README.test.js

@@ -15,6 +15,11 @@
 import {parseMarkdownAndTestCodeblocks } from './dist/src/TestUtils.js';
 import CliCore from './dist/src/CliCore.js';
 
-
-parseMarkdownAndTestCodeblocks('./README.md', new CliCore());
+(async () => {
+    try {
+        await parseMarkdownAndTestCodeblocks('./README.md', new CliCore());
+    } catch (error) {
+        console.error('Error:', error);
+    }
+})();
 

example/ex24.json

@@ -0,0 +1,3 @@
+{
+  "myVar": "${ $env('MY_VAR', 'default value') }"
+}

src/CliCore.ts

@@ -44,9 +44,9 @@ export default class CliCore {
         this.currentDirectory = process.cwd();
         this.onInit = ()=>{};
     }
-    public close(){
+    public async close(){
         if(this.templateProcessor){
-            this.templateProcessor.close();
+            await this.templateProcessor.close();
         }
         if(this.server){
             this.server.close();
@@ -421,7 +421,7 @@ export default class CliCore {
                 countDown--;
             }
 
-            if(countDown === 0 || await compiledExpr.evaluate(data)===true){
+            if(countDown === 0 || await compiledExpr.evaluate(data)===true){ //check if the expression in the 'until' argument (the stop tailing condition) has evaluated to true
                 done = true;
                 unplug();
                 resolve(); //resolve the latch promise

src/ExecutionStatus.ts

@@ -29,7 +29,7 @@ export class ExecutionStatus {
         return stringifyTemplateJSON(this.toJsonObject());
     }
 
-    public toJsonObject():object{
+    public getForkMap():Map<string,Fork>{
         const outputsByForkId = new  Map<string, Fork>();
         Array.from(this.statuses).forEach((mutationPlan:Plan)=>{
             const {forkId, output, forkStack}= mutationPlan;
@@ -39,11 +39,16 @@ export class ExecutionStatus {
             });
 
         });
+        return outputsByForkId;
+    }
+
+    public toJsonObject():object{
+
         const snapshot = {
             template: this.tp.input,
             output: this.tp.output,
             options: this.tp.options,
-            mvcc:Array.from(outputsByForkId.values()),
+            mvcc:Array.from(this.getForkMap().values()),
             metaInfoByJsonPointer: this.metaInfosToJSON(this.metaInfoByJsonPointer),
             plans: Array.from(this.statuses).map(this.mutationPlanToJSON)
         };
@@ -121,8 +126,14 @@ export class ExecutionStatus {
         for (const mutationPlan of this.statuses) {
             // we initialize all functions/timeouts for each plan
             await tp.createRestorePlan(mutationPlan);
-            tp.executePlan(mutationPlan); // restart the restored plan asynchronously
-        };
+            //we don't await here. In fact, it is critical NOT to await here because the presence of multiple mutationPlan
+            //means that there was concurrent ($forked) execution and we don't want to serialize what was intended to
+            //run concurrently
+            tp.executePlan(mutationPlan).catch(error => {
+                console.error(`Error executing plan for mutation: ${this.mutationPlanToJSON(mutationPlan)}`, error);
+            }); // restart the restored plan asynchronously
+
+        }
     }
 
     /**

src/TemplateProcessor.ts

@@ -29,6 +29,7 @@ import {rateLimit} from "./utils/rateLimit.js"
 import {ExecutionStatus} from "./ExecutionStatus.js";
 import {Sleep} from "./utils/Sleep.js";
 import {saferFetch} from "./utils/FetchWrapper.js";
+import {env} from "./utils/env.js"
 import * as jsonata from "jsonata";
 import StatedREPL from "./StatedREPL.js";
 
@@ -212,19 +213,23 @@ export default class TemplateProcessor {
      * @static
      * @type {{
      *   fetch: typeof fetch,
-     *   setInterval: typeof setInterval,
      *   clearInterval: typeof clearInterval,
      *   setTimeout: typeof setTimeout,
-     *   console: Console
+     *   setInterval: typeof setInterval,
+     *   console: Console,
+     *   debounce: typeof debounce
+     *   Date: Date
+     *   rateLimit: typeof rateLimit
+     *   env: typeof env
      * }}
      */
     static DEFAULT_FUNCTIONS = {
         fetch:saferFetch,
-        setTimeout,
         console,
         debounce,
         Date,
-        rateLimit
+        rateLimit,
+        env
     }
 
     private static _isNodeJS = typeof process !== 'undefined' && process.release && process.release.name === 'node';
@@ -308,7 +313,7 @@ export default class TemplateProcessor {
 
     public executionStatus: ExecutionStatus;
 
-
+    private isClosed = false;
 
 
 
@@ -358,6 +363,7 @@ export default class TemplateProcessor {
 
     // resetting template means that we are resetting all data holders and set up new template
     private resetTemplate(template:object) {
+        this.executionQueue.length = 0; //empty the execution queue - it can contain lingering plans that mustn't infect this template
         this.input = JSON.parse(JSON.stringify(template));
         this.output = template; //initial output is input template
         this.templateMeta = JSON.parse(JSON.stringify(template));// Copy the given template to `initialize the templateMeta
@@ -373,24 +379,29 @@ export default class TemplateProcessor {
             TemplateProcessor.DEFAULT_FUNCTIONS,
             {"save": (output:object)=>{ //default implementation of save just logs the execution status
                     if (this.isEnabled("debug")){
-                        console.debug(this.executionStatus.toJsonString());
+                        this.logger.debug(this.executionStatus.toJsonString());
                     }
                     return output;
             }}, //note that save is before context, by design, so context can override save as needed
             context,
             {"set": this.setData},
-            {"sleep": new Sleep(this.timerManager).sleep}
+            {"sleep": new Sleep(this.timerManager).sleep},
+            {"setTimeout": this.timerManager.setTimeout},
+            {"clearTimeout": this.timerManager.clearTimeout},
+
         );
         const safe = this.withErrorHandling.bind(this);
         for (const key in this.context) {
             if (typeof this.context[key] === 'function') {
+                /*
                 if (key === "setTimeout" || key === "setInterval") {
                     //replace with wrappers that allow us to ensure we kill all prior timers when template re-inits
-                    // this.context[key] = this.timerManager[key].bind(this.timerManager);
+                    this.context[key] = this.timerManager[key].bind(this.timerManager);
                     //TODO: remove it after migrating to generated function
-                } else {
+                    //TODO: ^^^^ sergey please explain this comment
+                } else { */
                     this.context[key] = safe(this.context[key]);
-                }
+                //}
             }
         }
     }
@@ -435,7 +446,7 @@ export default class TemplateProcessor {
         }
 
         if (jsonPtr === "/" && this.isInitializing) {
-            console.error("-----Initialization '/' is already in progress. Ignoring concurrent call to initialize!!!! Strongly consider checking your JS code for errors.-----");
+            this.logger.error("-----Initialization '/' is already in progress. Ignoring concurrent call to initialize!!!! Strongly consider checking your JS code for errors.-----");
             return;
         }
 
@@ -470,7 +481,7 @@ export default class TemplateProcessor {
             }else{
                 compilationTarget = importedSubtemplate; //the case where we already initialized once, and now we are initializing an imported sub-template
             }
-            // Recretaing the meta info if execution status is not provided
+            // Recreating the meta info if execution status is not provided
             if (executionStatusSnapshot === undefined) {
                 const metaInfos = await this.createMetaInfos(compilationTarget , parsedJsonPtr);
                 this.metaInfoByJsonPointer[jsonPtr] = metaInfos; //dictionary for importedSubtemplate meta info, by import path (jsonPtr)
@@ -479,8 +490,10 @@ export default class TemplateProcessor {
                 this.setupDependees(metaInfos); //dependency <-> dependee is now bidirectional
                 this.propagateTags(metaInfos);
                 this.tempVars = [...this.tempVars, ...this.cacheTmpVarLocations(metaInfos)];
+                this.isClosed = false; //open the execution queue for processing
                 await this.evaluateInitialPlan(jsonPtr);
             } else {
+                this.isClosed = false;
                 await this.executionStatus.restore(this);
             }
             await this.postInitialize();
@@ -492,7 +505,10 @@ export default class TemplateProcessor {
         }
     }
 
-    close():void{
+    async close():Promise<void>{
+        this.isClosed = true;
+        this.executionQueue.length = 0; //nuke execution queue
+        await this.drainExecutionQueue();
         this.timerManager.clearAll();
         this.changeCallbacks.clear();
         this.executionStatus.clear();
@@ -1086,6 +1102,9 @@ export default class TemplateProcessor {
      * @returns {Promise<<JsonPointerString[]>} A promise with the list of json pointers touched by the plan
      */
     async setData(jsonPtr:JsonPointerString, data:any=null, op:Op="set"):Promise<JsonPointerString[]> {
+        if(this.isClosed){
+            throw new Error("Attempt to setData on a closed TemplateProcessor.")
+        }
         this.isEnabled("debug") && this.logger.debug(`setData on ${jsonPtr} for TemplateProcessor uid=${this.uniqueId}`)
         //get all the jsonPtrs we need to update, including this one, to percolate the change
         const fromPlan = [...this.from(jsonPtr)]; //defensive copy
@@ -1114,6 +1133,9 @@ export default class TemplateProcessor {
      * @param planStep
      */
     public async setDataForked(planStep:PlanStep):Promise<JsonPointerString[]>{
+        if(this.isClosed){
+            throw new Error("Attempt to setData on a closed TemplateProcessor.")
+        }
         const {jsonPtr} = planStep;
         this.isEnabled("debug") && this.logger.debug(`setData on ${jsonPtr} for TemplateProcessor uid=${this.uniqueId}`)
         const fromPlan = [...this.from(jsonPtr)]; //defensive copy
@@ -1123,7 +1145,7 @@ export default class TemplateProcessor {
     }
 
     private async drainExecutionQueue(){
-        while (this.executionQueue.length > 0) {
+        while (this.executionQueue.length > 0 && !this.isClosed) {
             try {
                 const plan: Plan | SnapshotPlan = this.executionQueue[0];
                 if (plan.op === "snapshot") {
@@ -1167,7 +1189,7 @@ export default class TemplateProcessor {
                 output = planStep.output; // forked/joined will change the output so we have to record it to pass to next step
             }
         } catch (e) {
-            console.error(`failed to initialize restore plan, error=${e}`);
+            this.logger.error(`failed to initialize restore plan, error=${e}`);
             throw e;
         }
     }

src/TestUtils.ts

@@ -26,8 +26,8 @@ export type CommandAndResponse = {
   jsonataExpr: string;
 };
 
-export function parseMarkdownAndTestCodeblocks(md:string, cliCore:CliCore, printFunction:(k:any, v:any)=>any = stringifyTemplateJSON){
-  runMarkdownTests(parseMarkdownTests(md, cliCore), cliCore, printFunction);
+export async function parseMarkdownAndTestCodeblocks(md:string, cliCore:CliCore, printFunction:(k:any, v:any)=>any = stringifyTemplateJSON){
+  await runMarkdownTests(parseMarkdownTests(md, cliCore), cliCore, printFunction);
 }
 
 /**
@@ -91,7 +91,7 @@ export function parseMarkdownTests(markdownPath:string, cliInstance:CliCore):Com
  * @param {object} cliCore An instance of the CLI class that has the methods to be tested.
  * @param {function} printFunction The function is used to print response output to compare with expected response.
  */
-function runMarkdownTests(testData: CommandAndResponse[], cliCore:CliCore, printFunction = stringifyTemplateJSON) {
+async function runMarkdownTests(testData: CommandAndResponse[], cliCore:CliCore, printFunction = stringifyTemplateJSON):Promise<void> {
   try {
     afterAll(async () => {
       if (cliCore) {
@@ -132,6 +132,6 @@ function runMarkdownTests(testData: CommandAndResponse[], cliCore:CliCore, print
       }, 100000); // set timeout to 100 seconds for each test
     });
   }finally {
-    cliCore.close();
+    await cliCore.close();
   }
 }