feat: pass value through activate to worker threads (#2842)

Author: idoros

packages/runtime-node/src/micro-rpc.ts

@@ -35,12 +35,12 @@ export function rpcPost(worker: Worker, type: string, value?: unknown) {
     worker.postMessage(outgoingMessage);
 }
 
-export function bindRpcListener<T>(type: string, customFetcher: () => Promise<T> | T) {
+export function bindRpcListener<T>(type: string, customFetcher: (value: unknown) => Promise<T> | T) {
     const handler = async (message: unknown) => {
         if (isValidRpcMessage(message) && message.type === type) {
             const outgoingMessage = {
                 id: message.id,
-                value: await customFetcher(),
+                value: await customFetcher(message.value),
             };
             if (parentPort) {
                 parentPort.postMessage(outgoingMessage);
@@ -60,7 +60,7 @@ export function bindRpcListener<T>(type: string, customFetcher: () => Promise<T>
     };
 }
 
-export function isValidRpcMessage(message: unknown): message is { type: string; id: string } {
+export function isValidRpcMessage(message: unknown): message is { type: string; id: string; value?: unknown } {
     return !!(
         message &&
         typeof message === 'object' &&

packages/runtime-node/src/node-env-manager.ts

@@ -19,7 +19,7 @@ export interface RunningNodeEnvironment {
     id: string;
     dispose(): Promise<void>;
     getMetrics(): Promise<PerformanceMetrics>;
-    activate?(): Promise<void>;
+    activate?(value?: unknown): Promise<void>;
 }
 
 export interface NodeEnvConfig extends Pick<AnyEnvironment, 'env' | 'endpointType'> {
@@ -126,11 +126,11 @@ export class NodeEnvManager implements IDisposable {
         return { port };
     }
 
-    async activateEnvs() {
+    async activateEnvs(value?: unknown) {
         const activatedEnvs: Promise<void>[] = [];
         for (const env of this.openEnvironments.values()) {
             if (!env.activate) continue;
-            activatedEnvs.push(env.activate());
+            activatedEnvs.push(env.activate(value));
         }
         await Promise.all(activatedEnvs);
     }

packages/runtime-node/src/worker-thread-initializer2.ts

@@ -9,7 +9,7 @@ import type { RunningNodeEnvironment } from './node-env-manager.js';
 export interface WorkerThreadInitializer2 extends RunningNodeEnvironment {
     initialize: () => Promise<void>;
     preLoad: () => void;
-    activate: () => Promise<void>;
+    activate: (value?: unknown) => Promise<void>;
 }
 
 export interface WorkerThreadInitializerOptions2 {
@@ -78,8 +78,8 @@ export function workerThreadInitializer2({
         });
     };
 
-    const activate = async () => {
-        rpcPost(worker, 'activate');
+    const activate = async (value?: unknown) => {
+        rpcPost(worker, 'activate', value);
         await envIsReady;
     };
 

packages/runtime-node/test-kit/entrypoints/a.node.ts

@@ -13,6 +13,11 @@ if (verbose) {
     console.log(`[${env.env}: Started with options: `, options);
 }
 
+let activateValue: unknown;
+export function getActivateValue() {
+    return activateValue;
+}
+
 export function runEnv({
     Feature = TestFeature,
     topLevelConfig = [],
@@ -39,7 +44,8 @@ export function runEnv({
 if (workerData) {
     const unbindMetricsListener = bindMetricsListener();
     let running: ReturnType<typeof runEnv>;
-    const unbindActivateListener = bindRpcListener('activate', () => {
+    const unbindActivateListener = bindRpcListener('activate', (value: unknown) => {
+        activateValue = value;
         unbindActivateListener();
         running = runEnv();
     });

packages/runtime-node/test-kit/entrypoints/b.node.ts

@@ -13,6 +13,11 @@ if (verbose) {
     console.log(`[${env.env}: Started with options: `, options);
 }
 
+let activateValue: unknown;
+export function getActivateValue() {
+    return activateValue;
+}
+
 export function runEnv({
     Feature = TestFeature,
     topLevelConfig = [],
@@ -39,7 +44,8 @@ export function runEnv({
 if (workerData) {
     const unbindMetricsListener = bindMetricsListener();
     let running: ReturnType<typeof runEnv>;
-    const unbindActivateListener = bindRpcListener('activate', () => {
+    const unbindActivateListener = bindRpcListener('activate', (value: unknown) => {
+        activateValue = value;
         unbindActivateListener();
         running = runEnv();
     });

packages/runtime-node/test-kit/feature/test-feature.a.env.ts

@@ -1,5 +1,6 @@
 import { aEnv } from './envs.js';
 import TestFeature from './test-feature.js';
+import { getActivateValue } from '../entrypoints/a.node.js';
 
 TestFeature.setup(aEnv, ({ echoBService }) => {
     return {
@@ -8,6 +9,7 @@ TestFeature.setup(aEnv, ({ echoBService }) => {
             echoChained: async () => {
                 return echoBService.echo();
             },
+            getActivateValue,
         },
     };
 });

packages/runtime-node/test-kit/feature/test-feature.b.env.ts

@@ -1,5 +1,6 @@
 import { bEnv } from './envs.js';
 import TestFeature from './test-feature.js';
+import { getActivateValue } from '../entrypoints/b.node.js';
 
 TestFeature.setup(bEnv, ({ echoAService }) => {
     return {
@@ -8,6 +9,7 @@ TestFeature.setup(bEnv, ({ echoAService }) => {
             echoChained: async () => {
                 return echoAService.echo();
             },
+            getActivateValue,
         },
     };
 });

packages/runtime-node/test-kit/feature/types.ts

@@ -1,4 +1,5 @@
 export type EchoService = {
     echo: () => string;
     echoChained: () => Promise<string>;
+    getActivateValue: () => unknown;
 };

packages/runtime-node/test/node-env.manager.unit.ts

@@ -193,6 +193,36 @@ describe('NodeEnvManager', () => {
 
             expect(await api.echo()).to.equal('a');
         });
+
+        it('should pass activate value to the worker', async () => {
+            const featureEnvironmentsMapping: NodeEnvsFeatureMapping = {
+                featureToEnvironments: {
+                    'test-feature': [aEnv.env, bEnv.env],
+                },
+                availableEnvironments: {
+                    a: {
+                        env: aEnv.env,
+                        endpointType: 'single',
+                        envType: 'node',
+                    },
+                    b: {
+                        env: bEnv.env,
+                        endpointType: 'single',
+                        envType: 'node',
+                    },
+                },
+            };
+
+            const manager = disposeAfterTest(new NodeEnvManager(meta, featureEnvironmentsMapping));
+            const { port } = await manager.autoLaunch(new Map([['feature', 'test-feature']]), {}, true);
+            const communication = disposeAfterTest(getClientCom(port));
+            const api = communication.apiProxy<EchoService>({ id: aEnv.env }, { id: 'test-feature.echoAService' });
+
+            const activateData = { hello: 'world' };
+            await manager.activateEnvs(activateData);
+
+            expect(await api.getActivateValue()).to.deep.equal(activateData);
+        });
     });
 
     describe('NodeEnvManager with connection handlers', () => {