Proper handle binary output

Author: patuwwy

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

@@ -67,7 +67,7 @@ export function createStreamHandlers(router: SequentialCeroRouter) {
                 ? `${type}; charset=${encoding}`
                 : type;
 
-            out.setEncoding(encoding);
+            logger.debug("encoding, cType, readableEncoding", encoding, cType, data.readableEncoding);
 
             res.setHeader("content-type", cType);
             res.setHeader("transfer-encoding", "chunked");

packages/host/src/lib/csi-controller.ts

@@ -496,6 +496,8 @@ export class CSIController extends TypedEmitter<Events> {
                 this.apiInputEnabled = false;
             }
 
+            this.upStreams[CC.OUT].setDefaultEncoding(pangData.outputEncoding || "utf-8");
+
             this.emit("pang", {
                 provides: this.provides,
                 requires: this.requires,
@@ -628,7 +630,8 @@ export class CSIController extends TypedEmitter<Events> {
         if (development()) {
             this.router.upstream("/monitoring", this.upStreams[CC.MONITORING]);
         }
-        this.router.upstream("/output", this.upStreams[CC.OUT]);
+
+        this.router.upstream("/output", this.upStreams[CC.OUT], { encoding: "binary" });
 
         this.router.downstream("/input", (req) => {
             if (this.apiInputEnabled) {

packages/host/src/lib/host.ts

@@ -628,12 +628,15 @@ export class Host implements IComponent {
     spaceMiddleware(req: ParsedMessage, res: ServerResponse) {
         const url = req.url!.replace(`${this.apiBase}/cpm/api/v1/`, "");
 
-        this.logger.info("SPACE REQUEST", req.url, url, this.apiBase, req.body);
+        this.logger.info("SPACE REQUEST", req.url, url, this.apiBase);
 
         const clientRequest = this.cpmConnector?.makeHttpRequestToCpm(req.method!, url, req.headers);
 
         if (clientRequest) {
             clientRequest.on("response", (response: IncomingMessage) => {
+                response.on("end", () => {
+                    this.logger.info("Space response ended", url, response.statusCode);
+                });
                 response.pipe(res);
             }).on("error", (error) => {
                 this.logger.error("Error requesting CPM", error);

packages/host/src/lib/socket-server.ts

@@ -4,6 +4,7 @@ import net from "net";
 
 import { isDefined, TypedEmitter } from "@scramjet/utility";
 import { ObjLogger } from "@scramjet/obj-logger";
+import { CommunicationChannel } from "@scramjet/symbols";
 
 type MaybeSocket = net.Socket | null
 type RunnerConnectionsInProgress = [
@@ -58,7 +59,9 @@ export class SocketServer extends TypedEmitter<Events> implements IComponent {
 
                 connection
                     .on("error", (err) => this.logger.error("Error on Instance in stream", id, channel, err))
-                    .on("end", () => this.logger.debug(`Channel [${id}:${channel}] ended`));
+                    .on("end", () => this.logger.debug(
+                        `Channel [${id}:${channel}] ended. tx/rx: ${connection.bytesWritten}/${connection.bytesRead}`
+                    ));
 
                 try {
                     await this.handleConnection(id, channel, connection);
@@ -90,8 +93,10 @@ export class SocketServer extends TypedEmitter<Events> implements IComponent {
         } else {
             throw new Error(`Runner(${id}) wanted to connect on already initialized channel ${channel}`);
         }
+
         if (runner.every(isDefined)) {
             this.runnerConnectionsInProgress.delete(id);
+            runner[CommunicationChannel.OUT]?.setDefaultEncoding("binary");
             this.emit("connect", id, runner as RunnerOpenConnections);
         }
     }

packages/runner/src/host-client.ts

@@ -65,6 +65,9 @@ class HostClient implements IHostClient {
         );
 
         this._streams = openConnections as HostOpenConnections;
+        this._streams[CC.OUT].on("end", () => {
+            this.logger.info("Total data written to instance output", (this.streams[CC.OUT] as net.Socket).bytesWritten);
+        });
 
         try {
             this.bpmux = new BPMux(this._streams[CC.PACKAGE]);

packages/runner/src/runner.ts

@@ -589,6 +589,13 @@ export class Runner<X extends AppConfig> implements IComponent {
                     stream instanceof StringStream || stream instanceof BufferStream
                 );
 
+                if (!shouldSerialize) {
+                    this.hostClient.outputStream.setDefaultEncoding("binary");
+                }
+
+                this.logger.info("Will Output be serialized?", shouldSerialize);
+                this.logger.info("Stream encoding is", stream.readableEncoding);
+
                 stream
                     .once("end", () => {
                         this.logger.debug("Sequence stream ended");
@@ -602,7 +609,8 @@ export class Runner<X extends AppConfig> implements IComponent {
                 MessageUtils.writeMessageOnStream(
                     [RunnerMessageCode.PANG, {
                         provides: intermediate.topic || "",
-                        contentType: intermediate.contentType || ""
+                        contentType: intermediate.contentType || "",
+                        outputEncoding: shouldSerialize ? "utf-8" : "binary"
                     }],
                     this.hostClient.monitorStream,
                 );

packages/types/src/messages/handshake.ts

@@ -10,7 +10,8 @@ export type HandshakeMessage = { msgCode: RunnerMessageCode.PING };
 export type PingMessageData = { ports?: Record<string, string> }
 
 export type PangMessageData = {
-    requires?: string,
-    contentType?: string,
-    provides?: string
+    requires?: string;
+    contentType?: string;
+    provides?: string;
+    outputEncoding?: BufferEncoding;
 }

packages/verser/src/lib/verser-client.ts

@@ -121,6 +121,7 @@ export class VerserClient extends TypedEmitter<Events> {
                 resolve({ res, socket });
             });
 
+            connectRequest.socket?.setNoDelay(true);
             connectRequest.flushHeaders();
         });
     }

packages/verser/src/lib/verser.ts

@@ -25,6 +25,7 @@ export class Verser extends TypedEmitter<Events> {
         this.server = server;
 
         this.server.on("connect", (req, socket: Socket) => {
+            socket.setNoDelay(true);
             this.logger.info("New connection:", req.url);
 
             const connection = new VerserConnection(req, socket);