fix OpenHausIO#51

bootstrap.js

@@ -79,9 +79,47 @@ function bootstrap() {
                 }
             });
 
+        }),
+
+
+        // connect to websocket events
+        new Promise((resolve, reject) => {
+
+            let uri = new url.URL(process.env.BACKEND_URL);
+            uri.protocol = (process.env.BACKEND_PROTOCOL === "https" ? "wss" : "ws");
+            uri.pathname = "/api/system/connector";
+            uri.search = `x-auth-token=${process.env.AUTH_TOKEN}`;
+
+            let ws = new WebSocket(uri);
+
+            ws.on("open", () => {
+
+                logger.debug(`WebSocket connected to: ${ws.url}`);
+
+                resolve(ws);
+
+            });
+
+            ws.on("error", (err) => {
+                //console.error("Websocket", err);
+                reject(err);
+            });
+
+            ws.on("close", (code) => {
+                if (code === 1006) {
+
+                    retry();
+
+                } else {
+
+                    console.warn("WebSocket (event) conneciton closed", code);
+
+                }
+            });
+
         })
 
-    ]).then(([map, ws]) => {
+    ]).then(([map, ws, connector]) => {
 
         // reset flags
         counter = 0;
@@ -91,6 +129,7 @@ function bootstrap() {
 
         require("./forwarder.js");
         require("./handler.js")(map, ws);
+        require("./socketize.js")(map, connector);
 
     }).catch((err) => {
         if (err.code === "ECONNREFUSED") {

bridge2.js

@@ -0,0 +1,90 @@
+#! /usr/bin/node
+
+const { EOL } = require("os");
+const {
+    isMainThread,
+    workerData,
+    parentPort
+} = require("worker_threads");
+
+const WebSocket = require("ws");
+const minimist = require("minimist");
+
+
+const argv = minimist(process.argv, {
+    boolean: ["help"],
+    default: {
+        upstream: "",
+        socket: "tcp",
+        host: "",
+        port: ""
+    }
+});
+
+
+if (!isMainThread) {
+
+    // arguments passed via workerData
+    // path argv object with passed data
+    Object.assign(argv, workerData);
+
+}
+
+
+// check arguments if used as cli client or spawend via worker thread
+if ((isMainThread && (!argv.upstream || !argv.host)) || argv.help || (!argv.port && argv.socket !== "raw")) {
+    console.group("Usage of bridge.js as cli tool:", EOL);
+    console.log(`bridge.js --upstream="ws://example.com" --host="127.0.0.1" --port="8080" --socket="tcp"`);
+    console.log(`bridge.js --upstream="ws://open-haus.lan/api/foo/bar" --host="172.16.0.11" --socket="udp" --port="53"`);
+    console.log(`bridge.js --upstream="ws://127.0.0.1:8080/api/devices/663fc49985397fe02064d60d/interfaces/663fc4a06a1e907dd8e86f0e" --socket="tcp" --host="127.0.0.1" --port="8123"`);
+    console.log(`bridge.js --upstream="ws://127.0.0.1:8080/api/devices/663fc4b0490a00181d03486c/interfaces/663fc4b5d6cf46265f713ba4" --host="192.168.2.1" --socket="raw"`, EOL);
+    console.log("--upstream\tWebSocket upstream endpoint");
+    console.log("--socket\tNetwork socket type: tcp|udp|raw");
+    console.log("--host\tHost to connect to");
+    console.log("--port\tHost port to connect to");
+    console.log("");
+    process.exit(1);
+}
+
+//console.log(`bridge2.js --upstream="${argv.upstream}" --host="${argv.host}" --port="${argv.port}" --socket="${argv.socket}"`);
+
+// bridge the websocket stream to underlaying network socket
+let ws = new WebSocket(argv.upstream);
+
+ws.once("error", (err) => {
+
+    console.error(err);
+    process.exit(10);
+
+});
+
+ws.once("close", (code) => {
+    console.log("Closed with code", code);
+    process.exit();
+});
+
+ws.once("open", () => {
+
+    let upstream = WebSocket.createWebSocketStream(ws);
+
+    let socket = require(`./sockets/${argv.socket}.js`)({
+        host: argv.host,
+        port: argv.port
+    });
+
+    upstream.pipe(socket);
+    socket.pipe(upstream);
+
+    if (!isMainThread) {
+        parentPort.on("message", (msg) => {
+            if (msg === "disconnect") {
+
+                ws.close(() => {
+                    process.exit(0);
+                });
+
+            }
+        });
+    }
+
+});

package.json

@@ -42,6 +42,7 @@
     "dateformat": "^4.6.3",
     "dotenv": "^16.0.0",
     "minimist": "^1.2.6",
+    "raw-socket": "^1.8.1",
     "ws": "^8.5.0",
     "xml-js": "^1.6.11"
   },
@@ -52,4 +53,4 @@
     "nodemon": "^2.0.15",
     "pkg": "^5.8.0"
   }
-}
+}

socketize.js

@@ -0,0 +1,49 @@
+const { URLSearchParams } = require("url");
+const { Worker } = require("worker_threads");
+
+
+module.exports = (map, ws) => {
+    ws.on("message", (msg) => {
+
+        console.log("Message from backend for bidrigin interface", msg.toString());
+
+        // parse data
+        msg = JSON.parse(msg);
+
+        let sp = new URLSearchParams();
+
+        sp.set("socket", "true");
+        sp.set("uuid", msg.uuid);
+        sp.set("type", "response");
+        sp.set("x-auth-token", process.env.AUTH_TOKEN);
+
+
+        console.log("interface url mapping", map);
+
+        let upstream = `${process.env.BACKEND_URL}/api/devices/${msg.device}/interfaces/${msg.interface}`;
+        let { host, port, socket } = map.get(upstream).settings;
+
+        let worker = new Worker("./bridge2.js", {
+            workerData: {
+                upstream: `${upstream.replace("http", "ws")}?${sp.toString()}`,
+                host,
+                port,
+                socket
+            },
+            env: process.env
+        });
+
+        worker.once("online", () => {
+            console.log("Worker spawend for url %s", upstream);
+        });
+
+        worker.once("exit", (code) => {
+            console.log("Worker exited with code %d: %s", code, upstream)
+        });
+
+        worker.once("error", (err) => {
+            console.error("Worker died", err, upstream);
+        });
+
+    });
+};

sockets/raw.js

@@ -1,6 +1,58 @@
+const { Duplex } = require("stream");
+const raw = require("raw-socket");
+
 // this file handles raw network sockets
 // the protocol implemtnation is done on the server side
+const logger = require("../system/logger.js");
 
 module.exports = ({ host, port }, options) => {
-    console.log(`TO BE IMPLEMENTED - raw://${host}:${port}`, options);
+
+    let socket = raw.createSocket({
+        protocol: raw.Protocol.ICMP
+    });
+
+    let stream = new Duplex({
+        write(chunk, encoding, cb) {
+
+            console.log("Write to device", `raw://${host}:${port}`, chunk);
+
+            socket.send(chunk, 0, chunk.length, host, (error, bytes) => {
+                console.log("Writen to devoce")
+                if (error)
+                    console.log(error.toString());
+            });
+
+
+        },
+        read(size) {
+            logger.verbose(`raw://${host}:${port} Read called`, size);
+        },
+        end(chunk) {
+            if (chunk) {
+                socket.send(chunk, 0, chunk.length, host, (error, bytes) => {
+                    if (error)
+                        console.log(error.toString());
+                });
+            }
+            socket.close();
+        }
+    });
+
+    socket.on("error", (err) => {
+        logger.error(`[error] raw://${host}:${port}`, err);
+    });
+
+    socket.on("close", () => {
+        logger.debug(`[closed] raw://${host}:${port}`);
+    });
+
+    socket.on("message", (buffer, source) => {
+        if (source === host) {
+            console.log("received " + buffer.length + " bytes from " + source);
+            stream.push(buffer);
+        }
+    });
+
+    return stream;
+
 };

system/bridge.js

@@ -85,7 +85,7 @@ function bridge(uri, settings, options) {
             });
 
             stream.on("error", (err) => {
-                if (["ECONNRESET", "ECONNREFUSED"].includes(err.code) && process.env.ALLOW_HALF_OPEN === "true") {
+                if (["ECONNRESET", "ECONNREFUSED", "ETIMEDOUT"].includes(err.code) && process.env.ALLOW_HALF_OPEN === "true") {
 
                     console.log("Coult not connect, half open:", process.env.ALLOW_HALF_OPEN, err);
 

system/socket.js

@@ -0,0 +1,16 @@
+const WebSocket = require("ws");
+
+module.exports = ({ upstream, socket, host, port }) => {
+
+    let ws = new WebSocket(upstream);
+    let wsStream = WebSocket.createWebSocketStream(ws);
+
+    let stream = require(`../sockets/${socket}.js`)({
+        host,
+        port
+    });
+
+    wsStream.pipe(stream);
+    stream.pipe(wsStream);
+
+};