Setup SSE Server to handle responses/feed requests back and forth

harness/helpers/helpers.ts

@@ -11,8 +11,10 @@ import {
     callVariableValue,
     sendCommand,
 } from './commands'
-import { config_sse } from '../mockData'
+import { config_sse, SSEMessage } from '../mockData'
 import { ProxyClientOptions } from './proxyClientOptions'
+import SseStream from 'ssestream'
+import { sseStreams } from '../mockServer'
 
 const oldFetch = fetch
 
@@ -439,6 +441,17 @@ export class LocalTestClient extends BaseTestClient {
     getValidConfig() {
         return config_sse(getMockServerUrl(), `/client/${this.clientId}/sse`)
     }
+
+    getSSEStream(): SseStream {
+        return sseStreams[this.clientId]
+    }
+    sendSSEMessage(message: SSEMessage) {
+        const sseStream = this.getSSEStream()
+        if (!sseStream) {
+            throw new Error('SSE Stream not initialized')
+        }
+        sseStream.write(JSON.stringify(message))
+    }
 }
 
 export class CloudTestClient extends BaseTestClient {

harness/mockServer/index.ts

@@ -2,14 +2,15 @@ import Koa from 'koa'
 import Router from '@koa/router'
 import axios from 'axios'
 import bodyParser from 'koa-bodyparser'
+import SseStream from 'ssestream'
 
 // NOTE: This file is excecuted by the jest environment, which does not share a memory space / module cache with the
 // jest test files. This means that any things you want to be able to access from test context
 // (like the unmatchedRequests) will need to be bound to the global scope inside the jest-environment and accessed
 // through that
 
 let unmatchedRequests = []
-
+export const sseStreams: Map<string, SseStream> = new Map()
 export const assertNoUnmatchedRequests = async (
     currentClientId,
     testNameMap,
@@ -48,6 +49,15 @@ export function initialize() {
         return next()
     }).all('/client/:clientId/sse', async (ctx) => {
         console.log(ctx.body)
+        let sseStream = sseStreams[ctx.clientId]
+        if (!sseStream) {
+            sseStream = new SseStream(ctx.req)
+            sseStreams[ctx.clientId] = sseStream
+        }
+        sseStream.pipe(ctx.res)
+        ctx.res.on('close', () => {
+            sseStream.unpipe(ctx.res)
+        })
     })
     router.all('/(.*)', async (ctx) => {
         const { headers, request } = ctx

package.json

@@ -25,8 +25,8 @@
         "@eresearchqut/jest-testcontainers": "^3.2.0",
         "@koa/router": "^12.0.1",
         "@types/jest": "^29.2.3",
-        "@types/koa__router": "^12.0.4",
         "@types/koa-bodyparser": "^4.3.12",
+        "@types/koa__router": "^12.0.4",
         "@typescript-eslint/eslint-plugin": "^5.44.0",
         "@typescript-eslint/parser": "^5.44.0",
         "axios": "^1.1.3",
@@ -52,6 +52,7 @@
     "dependencies": {
         "dotenv": "^16.0.3",
         "object-path-immutable": "^4.1.2",
+        "ssestream": "^1.1.0",
         "uuid": "^9.0.0"
     },
     "lint-staged": {

yarn.lock

@@ -559,6 +559,7 @@ __metadata:
     prettier: "npm:^3.2.5"
     prettier-eslint: "npm:^16.3.0"
     prettier-eslint-cli: "npm:^8.0.1"
+    ssestream: "npm:^1.1.0"
     ts-jest: "npm:^29.0.3"
     ts-node: "npm:^10.9.1"
     typescript: "npm:^4.9.3"
@@ -6672,6 +6673,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"ssestream@npm:^1.1.0":
+  version: 1.1.0
+  resolution: "ssestream@npm:1.1.0"
+  checksum: 10/29b0f313fd0a599360432a866ebf3d6355a3b137d3007f07fb32572f85dd3142575214264ff7f479fbb65e87f79c048c3af90506ad0303eecffa95c62b7656c5
+  languageName: node
+  linkType: hard
+
 "ssh-remote-port-forward@npm:^1.0.4":
   version: 1.0.4
   resolution: "ssh-remote-port-forward@npm:1.0.4"