/* This file is part of GNU Taler (C) 2023 Taler Systems S.A. GNU Taler is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 3, or (at your option) any later version. GNU Taler is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with GNU Taler; see the file COPYING. If not, see */ import * as net from "node:net"; import * as fs from "node:fs"; import { Logger } from "./logging.js"; import { bytesToString, typedArrayConcat } from "./taler-crypto.js"; import type { RpcConnectArgs, RpcServerArgs } from "./twrpc.js"; interface ReadLinewiseArgs { onLine(lineData: Uint8Array): void; sock: net.Socket; } const logger = new Logger("twrpc-impl.node.ts"); function readStreamLinewise(args: ReadLinewiseArgs): void { let chunks: Uint8Array[] = []; args.sock.on("data", (buf: Uint8Array) => { // Process all newlines in the newly received buffer while (1) { const newlineIdx = buf.indexOf("\n".charCodeAt(0)); if (newlineIdx >= 0) { let left = buf.subarray(0, newlineIdx + 1); let right = buf.subarray(newlineIdx + 1); chunks.push(left); const line = typedArrayConcat(chunks); args.onLine(line); chunks = []; buf = right; } else { chunks.push(buf); break; } } }); } export async function connectRpc(args: RpcConnectArgs): Promise { let sockFilename = args.socketFilename; return new Promise((resolve, reject) => { const client = net.createConnection(sockFilename); client.on("error", (e) => { reject(e); }); client.on("connect", () => { let parsingBody: string | undefined = undefined; let bodyChunks: string[] = []; logger.info("connected!"); client.write("%hello-from-client\n"); const res = args.onEstablished({ sendMessage(m) { client.write("%request\n"); client.write(JSON.stringify(m)); client.write("\n"); client.write("%end\n"); }, close() { client.destroy(); }, }); readStreamLinewise({ sock: client, onLine(line) { const lineStr = bytesToString(line); // Are we currently parsing the body of a request? if (!parsingBody) { const strippedLine = lineStr.trim(); if (strippedLine == "%message") { parsingBody = "message"; } else if (strippedLine == "%hello-from-server") { } else if (strippedLine.startsWith("%error:")) { client.end(); res.onDisconnect(); } else { logger.warn("got unknown request"); client.write("%error: invalid message\n"); client.end(); } } else if (parsingBody == "message") { const strippedLine = lineStr.trim(); if (strippedLine == "%end") { let req = bodyChunks.join(""); let reqJson: any = undefined; try { reqJson = JSON.parse(req); } catch (e) { logger.warn("JSON message from server was invalid"); logger.info(`message was: ${req}`); } if (reqJson !== undefined) { res.onMessage(reqJson); } else { client.write("%error: invalid JSON"); client.end(); } bodyChunks = []; parsingBody = undefined; } else { bodyChunks.push(lineStr); } } else { logger.info("invalid parser state"); client.write("%error: internal error\n"); client.end(); } }, }); client.on("close", () => { res.onDisconnect(); }); client.on("data", () => {}); resolve(res.result); }); }); } export async function runRpcServer(args: RpcServerArgs): Promise { let sockFilename = args.socketFilename; try { fs.unlinkSync(sockFilename); } catch (e) { // Do nothing! } return new Promise((resolve, reject) => { const server = net.createServer((sock) => { // Are we currently parsing the body of a request? let parsingBody: string | undefined = undefined; let bodyChunks: string[] = []; sock.write("%hello-from-server\n"); const handlers = args.onConnect({ sendResponse(message) { sock.write("%message\n"); sock.write(JSON.stringify(message)); sock.write("\n"); sock.write("%end\n"); }, }); sock.on("error", (err) => { logger.error(`connection error: ${err}`); }); function processLine(line: Uint8Array) { const lineStr = bytesToString(line); if (!parsingBody) { const strippedLine = lineStr.trim(); if (strippedLine == "%request") { parsingBody = "request"; } else if (strippedLine === "%hello-from-client") { // Nothing to do, ignore hello } else if (strippedLine.startsWith("%error:")) { logger.warn("got error from client"); sock.end(); handlers.onDisconnect(); } else { logger.info("got unknown request"); sock.write("%error: invalid request\n"); sock.end(); } } else if (parsingBody == "request") { const strippedLine = lineStr.trim(); if (strippedLine == "%end") { let req = bodyChunks.join(""); let reqJson: any = undefined; try { reqJson = JSON.parse(req); } catch (e) { logger.warn("JSON request from client was invalid"); } if (reqJson !== undefined) { handlers.onMessage(reqJson); } else { sock.write("%error: invalid JSON"); sock.end(); } bodyChunks = []; parsingBody = undefined; } else { bodyChunks.push(lineStr); } } else { logger.error("invalid parser state"); sock.write("%error: internal error\n"); sock.end(); } } readStreamLinewise({ sock, onLine: processLine, }); sock.on("close", (hadError: boolean) => { logger.trace(`connection closed, hadError=${hadError}`); handlers.onDisconnect(); }); }); server.listen(args.socketFilename); }); }