Fix: websocket app

.github/workflows/ci.yml

@@ -15,7 +15,7 @@ jobs:
     strategy:
       matrix:
         operating-system: [macos-latest, ubuntu-latest]
-        node-version: [16, 18, 20]
+        node-version: [16, 18, 20, 22, 24]
     name: Node.js ${{ matrix.node-version }} Test on ${{ matrix.operating-system }}
 
     steps:

eslint.config.mjs

@@ -24,7 +24,7 @@ export default [...compat.extends("eslint:recommended"), {
             beforeEach: true,
         },
 
-        ecmaVersion: 2018,
+        ecmaVersion: 2020,
         sourceType: "commonjs",
 
         parserOptions: {

examples/websocket.server.js

@@ -1,3 +1,4 @@
+const { debug } = require('@axiosleo/cli-tool');
 const { WebSocketApplication } = require('../src/apps');
 const root = require('./api.router');
 
@@ -11,4 +12,10 @@ const app = new WebSocketApplication({
   }
 });
 
+setInterval(() => {
+  debug.log('send message');
+  const res = app.broadcast('Hello, world!', 'ok', 0, null);
+  debug.log('send message result:', res);
+}, 1000);
+
 app.start();

index.d.ts

@@ -481,6 +481,8 @@ interface KoaContext<
   TBody = any,
   TQuery = any,
 > extends AppContext<TParams, TBody, TQuery> {
+  /** Application instance */
+  app: KoaApplication;
   /** Route parameters */
   params?: TParams;
   /** Application configuration */
@@ -533,6 +535,10 @@ export interface SocketContext<
   TBody = any,
   TQuery = any,
 > extends AppContext<TParams, TBody, TQuery> {
+  /** Application instance */
+  app: SocketApplication;
+  /** Connection ID */
+  connection_id: string;
   /** Route parameters */
   params?: TParams;
   /** Application configuration */
@@ -580,6 +586,10 @@ export interface WebSocketContext<
   TBody = any,
   TQuery = any,
 > extends AppContext<TParams, TBody, TQuery> {
+  /** Application instance */
+  app: WebSocketApplication;
+  /** Connection ID */
+  connection_id: string;
   /** Route parameters */
   params?: TParams;
   /** Application configuration */
@@ -1236,7 +1246,7 @@ export declare abstract class Application extends EventEmitter {
   /** Application identifier */
   app_id: string;
   /** Application configuration */
-  config: Configuration;
+  config: AppConfiguration;
 
   constructor(config: AppConfiguration);
 
@@ -1301,6 +1311,7 @@ export declare class SocketClient {
  * Socket-based application
  */
 export declare class SocketApplication extends Application {
+  config: AppConfiguration;
   constructor(config: SocketAppConfiguration);
 
   /**
@@ -1322,6 +1333,53 @@ export declare class SocketApplication extends Application {
     code?: number,
     connections?: Socket[],
   ): void;
+
+  /**
+   * Send data to a specific connection
+   * @param connection Connection to send to
+   * @param data Data to send
+   * @param msg Message
+   * @param code Status code
+   */
+  send(connection: Socket, data?: any, msg?: string, code?: number): boolean;
+
+  /**
+   * Send data to a specific connection by connection ID
+   * @param connection_id Connection ID to send to
+   * @param data Data to send
+   * @param msg Message
+   * @param code Status code
+   */
+  sendByConnectionId(
+    connection_id: string,
+    data?: any,
+    msg?: string,
+    code?: number,
+  ): boolean;
+
+  /**
+   * Close a specific connection
+   * @param connection Connection to close
+   */
+  close(connection: Socket): boolean;
+
+  /**
+   * Close a specific connection by connection ID
+   */
+  closeByConnectionId(connection_id: string): boolean;
+
+  /**
+   * Get a specific connection by connection ID
+   * @param connection_id Connection ID to get
+   * @returns Connection or null if not found
+   */
+  getConnection(connection_id: string): Socket | null;
+
+  /**
+   * Ping a specific connection
+   * @param connection_id Connection ID to ping
+   */
+  ping(connection_id: string): boolean;
 }
 
 /**
@@ -1349,6 +1407,53 @@ export declare class WebSocketApplication extends Application {
     code?: number,
     connections?: WebSocket[],
   ): void;
+
+  /**
+   * Send data to a specific connection
+   * @param connection Connection to send to
+   * @param data Data to send
+   * @param msg Message
+   * @param code Status code
+   */
+  send(connection: WebSocket, data?: any, msg?: string, code?: number): boolean;
+
+  /**
+   * Send data to a specific connection by connection ID
+   * @param connection_id Connection ID to send to
+   * @param data Data to send
+   * @param msg Message
+   * @param code Status code
+   */
+  sendByConnectionId(
+    connection_id: string,
+    data?: any,
+    msg?: string,
+    code?: number,
+  ): boolean;
+
+  /**
+   * Close a specific connection
+   * @param connection Connection to close
+   */
+  close(connection: WebSocket): boolean;
+
+  /**
+   * Close a specific connection by connection ID
+   */
+  closeByConnectionId(connection_id: string): boolean;
+
+  /**
+   * Get a specific connection by connection ID
+   * @param connection_id Connection ID to get
+   * @returns Connection or null if not found
+   */
+  getConnection(connection_id: string): WebSocket | null;
+
+  /**
+   * Ping a specific connection
+   * @param connection_id Connection ID to ping
+   */
+  ping(connection_id: string): boolean;
 }
 
 // ========================================

src/apps/socket.js

@@ -11,7 +11,7 @@ const operator = require('../workflows/socket.workflow');
 const { _assign } = require('@axiosleo/cli-tool/src/helper/obj');
 const { _sleep } = require('@axiosleo/cli-tool/src/helper/cmd');
 
-const dispatcher = ({ app, app_id, workflow, connection }) => {
+const dispatcher = ({ app, app_id, workflow, connection, connection_id }) => {
   return async (ctx) => {
     let context = initContext({
       app,
@@ -21,6 +21,7 @@ const dispatcher = ({ app, app_id, workflow, connection }) => {
       app_id,
     });
     context.socket = connection;
+    context.connection_id = connection_id;
     context.query = ctx.query || {};
     context.body = ctx.body || {};
     try {
@@ -100,7 +101,8 @@ class SocketApplication extends Application {
               app: self,
               app_id: self.app_id,
               workflow: self.workflow,
-              connection
+              connection,
+              connection_id
             });
             process.nextTick(callback, context);
           } catch (err) {
@@ -157,6 +159,57 @@ class SocketApplication extends Application {
       Object.keys(connections).map((id) => connections[id].write(data));
     }
   }
+
+  send(connection = null, data = '', msg = 'ok', code = 0) {
+    if (connection) {
+      data = JSON.stringify({
+        request_id: _uuid_salt(this.app_id),
+        timestamp: (new Date()).getTime(),
+        code,
+        message: msg,
+        data: data
+      });
+      connection.write(data);
+      return true;
+    }
+    return false;
+  }
+
+  sendByConnectionId(connection_id = null, data = '', msg = 'ok', code = 0) {
+    if (connection_id && this.connections[connection_id]) {
+      return this.send(this.connections[connection_id], data, msg, code);
+    }
+    return false;
+  }
+
+  close(connection = null) {
+    if (connection) {
+      connection.end();
+      return true;
+    }
+    return false;
+  }
+
+  closeByConnectionId(connection_id = null) {
+    if (connection_id && this.connections[connection_id]) {
+      return this.close(this.connections[connection_id]);
+    }
+    return false;
+  }
+
+  getConnection(connection_id = null) {
+    if (connection_id && this.connections[connection_id]) {
+      return this.connections[connection_id];
+    }
+    return null;
+  }
+
+  ping(connection_id = null) {
+    if (connection_id && this.connections[connection_id]) {
+      return this.send(this.connections[connection_id], 'ping', 'ok', 0);
+    }
+    return false;
+  }
 }
 
 module.exports = SocketApplication;

src/apps/websocket.js

@@ -16,16 +16,19 @@ const { _sleep } = require('@axiosleo/cli-tool/src/helper/cmd');
  * @param {{request: import('http').IncomingMessage}} param0 
  * @returns 
  */
-const dispatcher = ({ app, app_id, workflow, connection, request }) => {
+const dispatcher = ({ app, app_id, workflow, connection_id, connection, request }) => {
   return async (ctx) => {
-    const url = new URL(request.url, `ws://localhost:${app.port}`);
+    const host = request.headers.host || `localhost:${app.port}`;
+    const protocol = request.headers['x-forwarded-proto'] === 'https' ? 'wss' : 'ws';
+    const url = new URL(request.url, `${protocol}://${host}`);
     let context = initContext({
       app,
       method: request.method ? request.method.toUpperCase() : 'GET',
       pathinfo: url.pathname,
       app_id,
     });
     context.socket = connection;
+    context.connection_id = connection_id;
     context.query = Object.fromEntries(url.searchParams);
     context.body = ctx || {};
     context.headers = request.headers;
@@ -116,6 +119,7 @@ class WebSocketApplication extends Application {
             app_id: self.app_id,
             workflow: self.workflow,
             connection: ws,
+            connection_id,
             request
           });
           process.nextTick(callback, context);
@@ -168,6 +172,57 @@ class WebSocketApplication extends Application {
       Object.keys(connections).map((id) => connections[id].send(data));
     }
   }
+
+  send(connection = null, data = '', msg = 'ok', code = 0) {
+    if (connection) {
+      data = JSON.stringify({
+        request_id: _uuid_salt(this.app_id),
+        timestamp: (new Date()).getTime(),
+        code,
+        message: msg,
+        data: data
+      });
+      connection.send(data);
+      return true;
+    }
+    return false;
+  }
+
+  sendByConnectionId(connection_id = null, data = '', msg = 'ok', code = 0) {
+    if (connection_id && this.connections[connection_id]) {
+      return this.send(this.connections[connection_id], data, msg, code);
+    }
+    return false;
+  }
+
+  close(connection = null) {
+    if (connection) {
+      connection.close();
+      return true;
+    }
+    return false;
+  }
+
+  closeByConnectionId(connection_id = null) {
+    if (connection_id && this.connections[connection_id]) {
+      return this.close(this.connections[connection_id]);
+    }
+    return false;
+  }
+
+  getConnection(connection_id = null) {
+    if (connection_id && this.connections[connection_id]) {
+      return this.connections[connection_id];
+    }
+    return null;
+  }
+
+  ping(connection_id = null) {
+    if (connection_id && this.connections[connection_id]) {
+      return this.send(this.connections[connection_id], 'ping', 'ok', 0);
+    }
+    return false;
+  }
 }
 
 module.exports = WebSocketApplication;