Add support to stop streaing messages.

nishants · Mar 28, 2021 · 33bc6d2 · 33bc6d2
1 parent d57cc37
commit 33bc6d2
Show file tree

Hide file tree

Showing 4 changed files with 52 additions and 7 deletions.
diff --git a/core/src/app.js b/core/src/app.js
@@ -19,8 +19,27 @@ module.exports = {
     const dataFiles  = TemplateReader.create({configPath});
 
     const mappingResolver = await endpointDataFileResolver.createResolvers({endpoints, mappings, templates: dataFiles});
+    const globalState = {
+      streamingContexts: {}
+    };
+
+    const globalActions = {
+      "stopStreaming@": async (arg) => {
+        const contextId = arg["context@"];
+        const streamingContext = globalState.streamingContexts[contextId]
+        if(!contextId){
+          throw new Error("stopStreaming@: Please provide context@ to stop.")
+        }
+        if(!streamingContext){
+          throw new Error(`stopStreaming@: No streaming context found for context@=${contextId}`)
+        }
+        delete globalState.streamingContexts[contextId]
+        await streamingContext.stop();
+      }
+    }
+
     const controllers = await Promise.all(endpoints.map((endpoint) => {
-      return Controllers.create(endpoint, mappingResolver, dataFiles)
+      return Controllers.create(endpoint, mappingResolver, dataFiles, globalState, globalActions)
     }));
 
     const client     = recording ? await Client.create({host : remoteHost, port : remotePort, endpoints}) : null;

diff --git a/core/src/controllers/duplex-streaming-controller.js b/core/src/controllers/duplex-streaming-controller.js
@@ -1,7 +1,7 @@
 const DEFAULT_STREAMING_DELAY = 1000;
 
 module.exports = {
-  create: async ({endpoint, mappingResolver, dataFiles}) => {
+  create: async ({endpoint, mappingResolver, dataFiles, globalState, globalActions}) => {
     return {
       endpointId: endpoint.getId(),
       canHandle: (endpointId) => {
@@ -18,6 +18,29 @@ module.exports = {
           const template = await dataFiles.get(responseFile);
           const response = template.getResponse().compile({request : {body: request}});
 
+          // ** Start of of streaming context handling ****************************************
+          const stop = async () => {
+            keepStreaming = false;
+          };
+
+          // Set streaming context
+          if(response.hasOwnProperty("context")){
+            globalState.streamingContexts[response.context] = {stop};
+          }
+
+          // Check for actions
+          if(response.actions){
+            for(const action in response.actions){
+              const globalAction = globalActions[action];
+              if(!globalAction){
+                throw new Error(`Undefined actions : ${action}`)
+              }
+              await globalAction(response.actions[action]);
+              console.log("Stopped streaming for messages")
+            }
+          }
+          // ** End of streaming context handling ****************************************
+
           if(!response){
             return callContext.end();
           }
@@ -27,11 +50,12 @@ module.exports = {
           let keepStreaming = !response.doNotRepeat ;
 
           for(let i =0; i < responses.length || keepStreaming; i++){
-            await new Promise((resolve => setTimeout(resolve, streamingDelay)));
             const nextMessage = responses[i%responses.length];
             console.log(`Sending response for ${endpoint.getId()} from ${responseFile} `, nextMessage);
             callContext.write(nextMessage);
+            await new Promise((resolve => setTimeout(resolve, streamingDelay)));
           }
+          console.log("End of streaming for ", request)
         };
 
         callContext.on('data', (response) => {

diff --git a/core/src/controllers/index.js b/core/src/controllers/index.js
@@ -4,7 +4,7 @@ const ClientStreamController = require('./duplex-streaming-controller');
 const DuplexStreamController = require('./duplex-streaming-controller');
 
 module.exports = {
-  create: (endpoint, mappingResolver, dataFiles) => {
+  create: (endpoint, mappingResolver, dataFiles, globalState, globalActions) => {
     const clientStream = endpoint.isStreamingRequest();
     const serverStream = endpoint.isStreamingResponse();
 
@@ -17,13 +17,13 @@ module.exports = {
     }
 
     if(isServerStream){
-      return ServerStreamController.create({endpoint, mappingResolver, dataFiles});
+      return ServerStreamController.create({endpoint, mappingResolver, dataFiles, globalState, globalActions});
     }
 
     if(isClientStream){
-      return ClientStreamController.create({endpoint, mappingResolver, dataFiles});
+      return ClientStreamController.create({endpoint, mappingResolver, dataFiles, globalState, globalActions});
     }
 
-    return DuplexStreamController.create({endpoint, mappingResolver, dataFiles});
+    return DuplexStreamController.create({endpoint, mappingResolver, dataFiles, globalState, globalActions});
   }
 }
diff --git a/core/src/mappingTemplate.js b/core/src/mappingTemplate.js
@@ -14,6 +14,8 @@ module.exports = {
             if(response['stream@']){
               return {
                 stream: compiled(response['stream@'], variables),
+                context: compiled(response['context@'], variables),
+                actions: compiled(response['actions@'], variables),
                 doNotRepeat: !!response['doNotRepeat@'],
                 streamInterval: response['streamInterval@'],
               }