调整通信

packages/message/README.md

@@ -6,5 +6,11 @@
 
 - 从脚本发起的GM请求，需要层层传递到service_worker/offscreen进行处理，有的GM只需要进行一次调用获取一次结果，有的需要进行
   多次调用获取多次结果，使用connect的方式实现
-- 从service_woker/offscreen发起的请求，类似消息队列，其它页面进行监听，触发后广播给所有页面，使用connect方式实现
-- 从扩展页面发起的请求，需要传递到service_worker/offscreen进行处理，如果只是单次调用，获取一次结果，使用message方式实现
+- 从service_worker/offscreen发起的请求，类似消息队列，其它页面进行监听，触发后广播给所有页面，使用sendMessage方式实现
+- 从扩展页面发起的请求，需要传递到service_worker/offscreen进行处理，如果只是单次调用，获取一次结果，使用sendMessage方式实现，如果需要
+  多次调用获取多次结果，使用connect方式实现
+
+## 注意点
+
+- service_worker和offscreen之间可以使用postMessage的方式进行通信，避免同时监听message与connect导致冲突的问题
+- service_worker会变为不活动的状态，尽量避免与service_worker建立长连接

packages/message/extension_message.ts

@@ -35,6 +35,9 @@ export class ExtensionMessage extends ExtensionMessageSend implements Message {
   // 注意chrome.runtime.onMessage.addListener的回调函数需要返回true才能处理异步请求
   onMessage(callback: (data: any, sendResponse: (data: any) => void) => void) {
     chrome.runtime.onMessage.addListener((msg, sender, sendResponse) => {
+      if (msg.action === "messageQueue") {
+        return false;
+      }
       return callback(msg, sendResponse);
     });
   }

packages/message/message_queue.ts

@@ -1,98 +1,57 @@
 import EventEmitter from "eventemitter3";
-import { ApiFunction, MessageConnect, MessageSend, Server } from "./server";
-import { sendMessage } from "./client";
 import Logger from "@App/app/logger/logger";
 import LoggerCore from "@App/app/logger/core";
 
 export type SubscribeCallback = (message: any) => void;
-
-export class Broker {
-  constructor(private msg: MessageSend) {}
-
-  // 订阅
-  async subscribe(topic: string, handler: SubscribeCallback): Promise<MessageConnect> {
-    LoggerCore.getInstance().logger({ service: "messageQueue" }).debug("subscribe", { topic });
-    const con = await this.msg.connect({ action: "messageQueue", data: { action: "subscribe", topic } });
-    con.onMessage((msg: { action: string; topic: string; message: any }) => {
-      if (msg.action === "message") {
-        handler(msg.message);
-      }
-    });
-    return con;
-  }
-
-  // 发布
-  publish(topic: string, message: any) {
-    sendMessage(this.msg, "messageQueue", { action: "publish", topic, message });
-  }
-}
+// 释放订阅
+export type Unsubscribe = () => void;
 
 // 消息队列
 export class MessageQueue {
-  topicConMap: Map<string, { name: string; con: MessageConnect }[]> = new Map();
-
   private EE: EventEmitter = new EventEmitter();
 
-  logger: Logger;
-  constructor(api: Server) {
-    api.on("messageQueue", this.handler());
-    this.logger = LoggerCore.getInstance().logger({ service: "messageQueue" });
+  constructor() {
+    chrome.runtime.onMessage.addListener((msg) => {
+      if (msg.action === "messageQueue") {
+        this.handler(msg.data);
+      }
+    });
   }
 
-  handler(): ApiFunction {
-    return ({ action, topic, message }: { action: string; topic: string; message: any }, con) => {
-      this.logger.trace("messageQueueHandler", { action, topic, message });
-      if (!con) {
-        throw new Error("con is required");
-      }
-      if (!topic) {
-        throw new Error("topic is required");
-      }
-      switch (action) {
-        case "subscribe":
-          this.subscribe(topic, con as MessageConnect);
-          break;
-        case "publish":
-          this.publish(topic, message);
-          break;
-        default:
-          throw new Error("action not found");
-      }
+  handler({ action, topic, message }: { action: string; topic: string; message: any }) {
+    LoggerCore.getInstance()
+      .logger({ service: "messageQueue" })
+      .trace("messageQueueHandler", { action, topic, message });
+    if (!topic) {
+      throw new Error("topic is required");
+    }
+    switch (action) {
+      case "message":
+        this.EE.emit(topic, message);
+        break;
+      default:
+        throw new Error("action not found");
+    }
+  }
+
+  subscribe(topic: string, handler: SubscribeCallback): Unsubscribe {
+    this.EE.on(topic, handler);
+    return () => {
+      this.EE.off(topic, handler);
     };
   }
 
-  private subscribe(topic: string, con: MessageConnect) {
-    let list = this.topicConMap.get(topic);
-    if (!list) {
-      list = [];
-      this.topicConMap.set(topic, list);
-    }
-    list.push({ name: topic, con });
-    con.onDisconnect(() => {
-      let list = this.topicConMap.get(topic);
-      // 移除断开连接的con
-      list = list!.filter((item) => item.con !== con);
-      this.topicConMap.set(topic, list);
-      this.logger.debug("disconnect", { topic });
-    });
-  }
-
   publish(topic: string, message: any) {
-    const list = this.topicConMap.get(topic);
-    list?.forEach((item) => {
-      item.con.sendMessage({ action: "message", topic, message });
+    chrome.runtime.sendMessage({
+      action: "messageQueue",
+      data: { action: "message", topic, message },
     });
     this.EE.emit(topic, message);
-    this.logger.trace("publish", { topic, message, list: list?.length });
+    LoggerCore.getInstance().logger({ service: "messageQueue" }).trace("publish", { topic, message });
   }
 
   // 只发布给当前环境
   emit(topic: string, message: any) {
     this.EE.emit(topic, message);
   }
-
-  // 同环境下使用addListener
-  addListener(topic: string, handler: (message: any) => void) {
-    this.EE.on(topic, handler);
-  }
 }

packages/message/server.ts

@@ -28,15 +28,21 @@ export class Server {
 
   private logger = LoggerCore.getInstance().logger({ service: "messageServer" });
 
-  constructor(message: Message) {
+  constructor(prefix: string, message: Message) {
     message.onConnect((msg: any, con: MessageConnect) => {
       this.logger.trace("server onConnect", { msg });
-      this.connectHandle(msg.action, msg.data, con);
+      if (msg.action.startsWith(prefix)) {
+        return this.connectHandle(msg.action.slice(prefix.length + 1), msg.data, con);
+      }
+      return false;
     });
 
-    message.onMessage((msg, sendResponse) => {
-      this.logger.trace("server onMessage", { msg });
-      return this.messageHandle(msg.action, msg.data, sendResponse);
+    message.onMessage((msg: { action: string; data: any }, sendResponse) => {
+      this.logger.trace("server onMessage", { msg: msg as any });
+      if (msg.action.startsWith(prefix)) {
+        return this.messageHandle(msg.action.slice(prefix.length + 1), msg.data, sendResponse);
+      }
+      return false;
     });
   }
 
@@ -98,10 +104,11 @@ export class Group {
 }
 
 // 转发消息
-export function forwardMessage(path: string, from: Server, to: MessageSend) {
+export function forwardMessage(prefix: string, path: string, from: Server, to: MessageSend) {
   from.on(path, (params, fromCon) => {
+    console.log("forwardMessage", path, prefix, params);
     if (fromCon) {
-      to.connect({ action: path, data: params }).then((toCon) => {
+      to.connect({ action: prefix + "/" + path, data: params }).then((toCon) => {
         fromCon.onMessage((data) => {
           toCon.sendMessage(data);
         });
@@ -116,7 +123,7 @@ export function forwardMessage(path: string, from: Server, to: MessageSend) {
         });
       });
     } else {
-      return to.sendMessage({ action: path, data: params });
+      return to.sendMessage({ action: prefix + "/" + path, data: params });
     }
   });
 }

packages/message/window_message.ts

@@ -171,7 +171,6 @@ export class ServiceWorkerMessageSend implements MessageSend {
     const list = await self.clients.matchAll({ includeUncontrolled: true, type: "window" });
     this.target = list[0];
     self.addEventListener("message", (e) => {
-      console.log("serviceWorker", e);
       this.messageHandle(e.data);
     });
   }