Update for planificador protocol changes

README.md

@@ -9,6 +9,20 @@ npm install
 npm run start
 ```
 
+## Endpoints
+
+### POST /run
+
+Runs a program and returns the output with an event stream.
+
+The stream always defines a `id` in each event. It may be:
+
+| id       | Description                                         |
+|----------|-----------------------------------------------------|
+| `stdout` | The event contains the output of the program.       |
+| `stderr` | The event contains the error output of the program. |
+| `exit`   | The event contains the exit code of the program.    |
+
 ## Bun support
 
 Pending [#5019](https://github.com/oven-sh/bun/issues/5019)

dprint.json

@@ -7,6 +7,6 @@
     "**/node_modules"
   ],
   "plugins": [
-    "https://plugins.dprint.dev/typescript-0.88.1.wasm"
+    "https://plugins.dprint.dev/typescript-0.88.9.wasm"
   ]
 }

package.json

@@ -16,7 +16,7 @@
     "@fastify/type-provider-typebox": "^4.0.0",
     "@sinclair/typebox": "^0.32.9",
     "fastify": "^4.25.2",
-    "nanoid": "^5.0.2",
-    "zeromq": "6.0.0-beta.17"
+    "nanoid": "^5.0.4",
+    "zeromq": "6.0.0-beta.19"
   }
 }

src/runner.ts

@@ -2,12 +2,21 @@ export const IMAGES = {
   moshell: 'ghcr.io/moshell-lang/moshell:master',
 };
 
+/**
+ * Prepares a buffer to be sent to the runner.
+ *
+ * @param jobId The job unique identifier.
+ * @param code The code to be executed.
+ * @param image The image to be used.
+ */
 export function allocateBuffer(jobId: string, code: string, image: string): Buffer {
-  const buffer = Buffer.allocUnsafe(jobId.length + image.length + code.length + 8);
-  buffer.write(jobId, 0);
-  buffer.writeUInt32BE(image.length, jobId.length);
-  buffer.writeUInt32BE(code.length, jobId.length + 4);
-  buffer.write(image, jobId.length + 8);
-  buffer.write(code, jobId.length + 8 + image.length);
+  let cur = 0;
+  const buffer = Buffer.allocUnsafe(jobId.length + image.length + code.length + 9);
+  cur = buffer.writeUInt8(0, cur);
+  cur += buffer.write(jobId, cur);
+  cur = buffer.writeUInt32BE(image.length, cur);
+  cur = buffer.writeUInt32BE(code.length, cur);
+  cur += buffer.write(image, cur);
+  buffer.write(code, cur);
   return buffer;
 }

src/server.ts

@@ -50,15 +50,36 @@ fastify.post('/run', {
 
 async function forwardOutput() {
   for await (const [buff] of receiver) {
-    const jobId = buff.subarray(0, 32).toString();
-    console.log(`Received ${jobId}`);
+    const messageType = buff.readInt8();
+    const jobId = buff.subarray(1, 33).toString();
     const reply = clients[jobId];
     if (!reply) {
       continue;
     }
-    reply.raw.write('event: message\n');
-    reply.raw.write(`data: ${encodeURIComponent(buff.subarray(32).toString())}\n`);
-    reply.raw.write('id: 1\n\n');
+    console.debug(`Forwarding message type ${messageType} for job ${jobId}`);
+    const raw = reply.raw;
+    switch (messageType) {
+      case 1:
+      case 2:
+        const text = encodeURIComponent(buff.subarray(37).toString());
+        raw.write('event: message\n');
+        if (messageType === 1) {
+          raw.write('id: stdout\n');
+        } else {
+          raw.write('id: stderr\n');
+        }
+        raw.write(`data: ${text}\n\n`);
+        break;
+      case 3:
+        const exitCode = buff.readUint32BE(33);
+        raw.write('event: message\n');
+        raw.write('id: exit\n');
+        raw.write(`data: ${exitCode}\n\n`);
+        raw.end();
+        break;
+      default:
+        console.error('Unknown message type', messageType);
+    }
   }
 }