Merge branch 'devel' into fix/remove-cpu-features

Author: patuwwy

bdd/data/sequences/event-sequence/index.js

@@ -0,0 +1,28 @@
+/* eslint-disable no-console */
+
+// eslint-disable-next-line valid-jsdoc
+/**
+ * Simple test event sequence.
+ *
+ * @param {never} _input - unused
+ * @param {string} inputEvent - input
+ * @param {string} outputEvent - output
+ * @returns {void}
+ * @this {import("@scramjet/types").AppContext<{}, {}>} - context
+ */
+module.exports = async function(_input, inputEvent = "in", outputEvent = "out") {
+    this.logger.info("started");
+    return new Promise((res) => {
+        this.on(inputEvent, async (msg) => {
+            const ev = JSON.parse(msg);
+
+            console.log("event", JSON.stringify(ev));
+            this.emit(outputEvent, JSON.stringify({ test: ev.test + 1 }));
+
+            await new Promise(res2 => setTimeout(res2, 100));
+
+            res();
+        });
+    });
+};
+

bdd/data/sequences/event-sequence/package.json

@@ -0,0 +1,25 @@
+{
+  "name": "@scramjet/event-sequence",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "predeploy": "mkdir -p dist/ && cp index.js package.json dist/ && (cd dist && npm i --omit=dev)"
+  },
+  "engines": {
+    "node": ">=16"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/scramjetorg/create-sequence.git"
+  },
+  "bugs": {
+    "url": "https://github.com/scramjetorg/create-sequence/issues"
+  },
+  "homepage": "https://github.com/scramjetorg/create-sequence#readme",
+  "devDependencies": {
+    "@scramjet/types": "^0.34.4"
+  },
+  "author": "",
+  "license": "ISC"
+}

bdd/features/hub/HUB-002-host-iac.feature

@@ -28,3 +28,21 @@ Feature: HUB-002 Host started in Infrastructure as Code mode
         And wait for "500" ms
         And host is running
         * exit hub process
+
+    @ci-hub @starts-host
+    Scenario: HUB-002 TC-004 Event forwarding works between sequences
+        When hub process is started with random ports and parameters "--sequences-root=data/sequences/ --instance-lifetime-extension-delay=10 --identify-existing --runtime-adapter=process"
+        And host is running
+        And I get list of instances
+        And start Instance by name "event-sequence" with JSON arguments '["event-one", "event-two"]'
+        * remember last instance as "first"
+        And start Instance by name "event-sequence" with JSON arguments '["event-two", "event-three"]'
+        * remember last instance as "second"
+        * switch to instance "first"
+        And send event "event-one" to instance with message '{"test": 1}'
+        # * wait for "100" ms
+        Then "stdout" starts with 'event {"test":1}'
+        * switch to instance "second"
+        Then "stdout" starts with 'event {"test":2}'
+        And host is running
+        * exit hub process

bdd/lib/utils.ts

@@ -218,23 +218,25 @@ export async function waitUntilStreamContains(stream: Readable, expected: string
     ]);
 }
 
-export async function waitUntilStreamEquals(stream: Readable, expected: string): Promise<string> {
+export async function waitUntilStreamEquals(stream: Readable, expected: string, timeout = 10000): Promise<string> {
     let response = "";
 
     await Promise.race([
         (async () => {
-            for await (const chunk of stream.pipe(new PassThrough({ encoding: undefined }))) {
-                response += chunk.toString();
+            for await (const chunk of stream.pipe(new PassThrough({ encoding: "utf-8" }))) {
+                response += chunk;
+
+                // eslint-disable-next-line no-console
+                console.log(response, chunk);
 
                 if (response === expected) return expected;
                 if (response.length >= expected.length) {
-                    assert.equal(response, expected);
+                    return assert.equal(response, expected);
                 }
             }
-            assert.equal(response, expected, "End of stream reached");
-
-            return "passed";
-        })()
+            throw new Error("End of stream reached");
+        })(),
+        defer(timeout).then(() => { assert.equal(response, expected, "timeout"); })
     ]);
 
     return response;
@@ -361,6 +363,27 @@ export function spawnSiInit(
     });
 }
 
+export async function waitUntilStreamStartsWith(stream: Readable, expected: string, timeout = 10000): Promise<string> {
+    let response = "";
+
+    await Promise.race([
+        (async () => {
+            for await (const chunk of stream.pipe(new PassThrough({ encoding: undefined }))) {
+                response += chunk.toString();
+
+                if (response === expected) return expected;
+                if (response.length >= expected.length) {
+                    return assert.equal(response.substring(0, expected.length), expected);
+                }
+            }
+            throw new Error("End of stream reached");
+        })(),
+        defer(timeout).then(() => { assert.equal(response, expected, "timeout"); })
+    ]);
+
+    return response;
+}
+
 export function isTemplateCreated(templateType: string, workingDirectory: string) {
     return new Promise<boolean>((resolve, reject) => {
         // eslint-disable-next-line complexity

bdd/step-definitions/e2e/host-steps.ts

@@ -7,6 +7,7 @@ import {
     removeBoundaryQuotes,
     defer,
     waitUntilStreamEquals,
+    waitUntilStreamStartsWith,
     waitUntilStreamContains,
     removeProfile,
     createProfile,
@@ -307,6 +308,37 @@ When(
 
 When("instance started with arguments {string}", { timeout: 25000 }, startWith);
 
+When("start Instance by name {string}", async function(this: CustomWorld, name: string) {
+    this.resources.sequence = hostClient.getSequenceClient(name);
+    this.resources.instance = await this.resources.sequence!.start({
+        appConfig: {}
+    });
+});
+
+When("start Instance by name {string} with JSON arguments {string}", async function(this: CustomWorld, name: string, args: string) {
+    const instanceArgs: any = JSON.parse(args);
+
+    if (!Array.isArray(instanceArgs)) throw new Error("Args must be an array");
+
+    this.resources.sequence = hostClient.getSequenceClient(name);
+    this.resources.instance = await this.resources.sequence!.start({
+        appConfig: {},
+        args: instanceArgs
+    });
+});
+
+When("remember last instance as {string}", function(this: CustomWorld, seq: string) {
+    if (!this.resources.instance) throw new Error("No instance client set");
+
+    this.resources.instanceList[seq] = this.resources.instance;
+});
+
+When("switch to instance {string}", function(this: CustomWorld, seq: string) {
+    if (!this.resources.instanceList[seq]) throw new Error(`No instance "${seq}"`);
+
+    this.resources.instance = this.resources.instanceList[seq];
+});
+
 When("start Instance with output topic name {string}", async function(this: CustomWorld, topicOut: string) {
     this.resources.instance = await this.resources.sequence!.start({
         appConfig: {},
@@ -754,6 +786,13 @@ When("send {string} to stdin", async function(this: CustomWorld, str) {
     await this.resources.instance?.sendStream("stdin", Readable.from(str));
 });
 
+Then("{string} starts with {string}", async function(this: CustomWorld, stream, text) {
+    const result = await this.resources.instance?.getStream(stream);
+
+    await waitUntilStreamStartsWith(result!, text);
+    if (!result) assert.fail(`No data in ${stream}!`);
+});
+
 Then("{string} is {string}", async function(this: CustomWorld, stream, text) {
     const result = await this.resources.instance?.getStream(stream);
     const response = await waitUntilStreamEquals(result!, text);

bdd/step-definitions/world.ts

@@ -16,14 +16,17 @@ export class CustomWorld implements World {
     resources: {
         [key: string]: any;
         hub?: ChildProcess;
+        instanceList: {[key: string]: InstanceClient};
         instance?: InstanceClient;
         instance1?: InstanceClient;
         instance2?: InstanceClient;
         sequence?: SequenceClient;
         sequence1?: SequenceClient;
         sequence2?: SequenceClient;
         outStream?: Readable;
-    } = {};
+    } = {
+            instanceList: {}
+        };
 
     cliResources: {
         stdio?: [stdout: string, stderr: string, statusCode: any];
@@ -48,7 +51,6 @@ export class CustomWorld implements World {
         if (setDefaultResultOrder) {
             setDefaultResultOrder("ipv4first");
         }
-
         this.attach = attach;
         this.log = log;
         this.parameters = parameters;

packages/api-client/src/host-client.ts

@@ -1,4 +1,4 @@
-import { ClientProvider, ClientUtils, HttpClient } from "@scramjet/client-utils";
+import { ClientProvider, ClientUtils, Headers, HttpClient } from "@scramjet/client-utils";
 import { STHRestAPI } from "@scramjet/types";
 import { InstanceClient } from "./instance-client";
 import { SequenceClient } from "./sequence-client";
@@ -193,10 +193,13 @@ export class HostClient implements ClientProvider {
     async sendTopic<T>(
         topic: string,
         stream: Parameters<HttpClient["sendStream"]>[1],
-        requestInit?: RequestInit,
+        requestInit: RequestInit = {},
         contentType: string = "application/x-ndjson",
         end?: boolean
     ) {
+        requestInit.headers ||= {} as Headers;
+        (requestInit.headers as Headers).expect = "100-continue";
+
         return this.client.sendStream<T>(`topic/${topic}`, stream, requestInit, { type: contentType, end: end });
     }
 

packages/api-server/src/handlers/op.ts

@@ -182,6 +182,10 @@ export function createOperationHandler(router: SequentialCeroRouter): APIRoute["
         const handler: Middleware = async (req, res, next) => {
             logger.trace("Request", req.method, req.url);
 
+            if (req.headers.expect === "100-continue") {
+                res.writeContinue();
+            }
+
             try {
                 if (typeof message === "function") {
                     return await opDataHandler(req, res, message, { rawBody });

packages/api-server/src/handlers/stream.ts

@@ -1,4 +1,4 @@
-import { StreamConfig, StreamInput, StreamOutput } from "@scramjet/types";
+import { ParsedMessage, StreamConfig, StreamInput, StreamOutput } from "@scramjet/types";
 import { IncomingHttpHeaders, IncomingMessage, ServerResponse } from "http";
 import { Writable, Readable, Duplex } from "stream";
 import { DuplexStream } from "../lib/duplex-stream";
@@ -106,16 +106,27 @@ export function createStreamHandlers(router: SequentialCeroRouter) {
     const downstream = (
         path: string | RegExp,
         stream: StreamOutput,
-        { json = false, text = false, end: _end = false, encoding = "utf-8", checkContentType = true, checkEndHeader = true, method = "post" }: StreamConfig = {}
+        { json = false, text = false, end: _end = false, encoding = "utf-8", checkContentType = true, checkEndHeader = true, method = "post", postponeContinue = false }: StreamConfig = {}
     ): void => {
-        router[method](path, async (req, res, next) => {
+        // eslint-disable-next-line complexity
+        router[method](path, async (req: ParsedMessage, res, next) => {
             try {
                 if (checkContentType) {
                     checkAccepts(req.headers["content-type"], text, json);
                 }
 
                 if (req.headers.expect === "100-continue") {
-                    res.writeContinue();
+                    if (!postponeContinue) {
+                        res.writeContinue();
+                    }
+                }
+
+                if (postponeContinue && req.headers.expect !== "100-continue") {
+                    res.writeHead(400, "Bad Request", { "Content-type": "application/json" });
+                    res.write(JSON.stringify({ error: "Missing 'expect' header" }));
+                    res.end();
+
+                    return;
                 }
 
                 // Explicit pause causes next `on('data')` not to resume stream automatically.
@@ -177,11 +188,14 @@ export function createStreamHandlers(router: SequentialCeroRouter) {
     };
     const duplex = (
         path: string | RegExp,
-        callback: (stream: Duplex, headers: IncomingHttpHeaders) => void
+        callback: (stream: Duplex, headers: IncomingHttpHeaders) => void,
+        { postponeContinue = false }: StreamConfig = {}
     ): void => {
         router.post(path, (req, res, next) => {
             if (req.headers.expect === "100-continue") {
-                res.writeContinue();
+                if (!postponeContinue) {
+                    res.writeContinue();
+                }
             }
 
             try {

packages/api-server/src/index.ts

@@ -100,8 +100,20 @@ export function createServer(conf: ServerConfig = {}): APIExpose {
     };
     const { server: srv, router } = cero({ server: createCeroServerConfig(conf), router: sequentialRouter(config) });
 
+    // Disable auto sending "100 Continue".
+    srv.on("checkContinue", (request, response) => {
+        request.writeContinue = () => {
+            response.writeContinue();
+        };
+
+        srv.emit("request", request, response);
+    });
+
     router.use("/", async (req, res, next) => {
+        req.writeContinue ||= () => {};
+
         next();
+
         // TODO: fix - this should log on errors.
         log.write({ date: Date.now(), method: req.method, url: req.url, status: await new Promise(s => res.on("finish", () => s(res.statusCode))) } as any);
     });