websocket

fmhub.js

@@ -0,0 +1,60 @@
+export default class {
+  constructor() {
+    this.channels = new Map()
+    this.users = new Map()
+  }
+  订阅频道(fid, uid) {
+    console.log(`用户 ${uid} 订阅了 ${fid}`)
+    let channel = this.channels.get(fid) || new Map()
+    if (!channel) this.channels.set(fid, channel)
+    channel.set(uid, true)
+  }
+  取消订阅(fid, uid) {
+    console.log(`用户 ${uid} 取消订阅 ${fid} 频道`)
+    let channel = this.channels.get(fid)
+    channel.delete(uid)
+    // TODO: 如果此频道没有用户订阅了, 则将此频道也移除
+    // TODO: 如果此用户没有任何订阅了, 则将此用户也移除
+  }
+  增加会话(uid, ws) {
+    console.log(`用户 ${uid} 建立了新的会话连接`)
+    let user = this.users.get(uid) || new Map()
+    if (!user) this.users.set(uid, user)
+    user.set(ws, true)
+  }
+  移除会话(uid, ws) {
+    console.log(`用户 ${uid} 结束了当前会话连接`)
+    let user = this.users.get(uid)
+    user.delete(ws)
+    console.log(`此用户还剩 ${user.size} 个 ws 连接`)
+    if (user.size < 1) {
+      console.log(`由于用户 ${uid} 已经没有会话, 直接移除此用户记录`)
+      this.users.delete(uid)
+    }
+  }
+  发送消息(fm, uid, data) {
+    console.log("发送消息", fm, uid, data)
+    let msg = JSON.stringify({ fm, uid, data })
+    let channel = this.channels.get(fm) || new Map()
+    if (!channel) this.channels.set(fm, channel)
+    channel.forEach((value, userid) => {
+      console.log(userid, value)
+      let user = this.users.get(userid)
+      if (!user) return console.log("订阅频道的用户不在线, 应移除此订阅");
+      user.forEach((value, ws) => {
+        ws.send(msg)
+      })
+    })
+  }
+  移除用户(uid) {
+    console.log(`移除 ${uid} 的所有会话`)
+    let user = this.users.get(uid)
+    if (user) {
+      user.forEach((value, ws) => ws.close()) // 断开用户所有会话
+      this.users.delete(uid)                  // 删除用户(所有会话)
+    }
+    // TODO: 移除所有频道中的此用户
+    // TODO: 移除此用户的记录
+    // TODO: 断开所有此用户的连接
+  }
+}

index.js

@@ -39,78 +39,42 @@ const user_load = async (_id) => await new Promise(resolve => db('user').findOne
   return resolve(user)
 }))
 
-const HUB = class {
-  constructor() {
-    this.频道 = new Map() // fid: [uid]
-    this.用户 = new Map() // uid: [ws]
-  }
-  增加会话(uid, ws) {
-    this.用户(uid) ? this.用户(uid).push(ws) : this.用户.set(uid, [ws])
-  }
-  增加频道(fid) {
-    this.频道.set(fid, [])
-  }
-  订阅频道(fid, uid) {
-    this.频道(fid) ? this.频道(fid).push(uid) : this.频道.set(fid, [uid])
-  }
-  取消订阅(fid, uid) {
-    this.频道(fid).filter(item => item != uid)
-  }
-  发送消息(fid, uid, data) {
-    this.频道(fid).forEach(userid => {
-      this.用户(userid).forEach(ws => {
-        ws.send({ fm: fid, uid, data })
-      })
-    })
-  }
-  移除会话(uid, ws) {
-    this.用户(uid).filter(item => item != ws)
-  }
-  移除频道(id) {
-    this.频道.delete(id)
-  }
-  移除用户(uid) {
-    this.频道.filter(item => item != uid)        // 取消用户所有订阅
-    this.用户.get(uid).forEach(ws => ws.close()) // 断开用户所有会话
-    this.用户.delete(uid)                        // 删除用户(所有会话)
-  }
-}
 
+import HUB from './fmhub.js'
 const FM = new HUB()
 
 // 通讯频道 Frequency Modulation
 function websocketer(ws, req) {
-
-  // 验证身份已经登录
-  //if (!req.session.account) return ws.close()
+  console.log("new websocket link T !")
 
   // 游客使用公共账户 uid = 0
-  let uid = req.session?.account?.uid || 0
+  let uid = req.session?.account?.uid || "0"
 
-  // 当游客连接时, 为其使用访客公共订阅列表
+  console.log(`用户 ${uid} 连接了服务器`)
+
+  // 访客默认订阅的频道列表: 一般是所有公开的频道
+  let list = ["chat", "system"]
+  list.forEach(fid => {
+    FM.订阅频道(fid, uid)
+  })
+
+  // 当用户连接时, 读取其订阅列表
   if (req.session.account) {
-    // 当用户连接时, 读取其订阅列表
     db('user').findOne({ uid }, function (err, doc) {
       if (doc && Array.isArray(doc.fm)) {
         doc.fm.forEach(fid => FM.订阅频道(fid, uid))
       }
-
-      // 管理员额外订阅数据频道
-      // 实时统计信号应另外接口连接
     })
-  } else {
-    // 访客默认订阅的频道列表: 一般是所有公开的频道
-    [].forEach(fid => FM.订阅频道(fid, uid))
   }
 
-  // 将连接加入到列表
+  // 将连接加入到列表 ws
   FM.增加会话(uid, ws)
 
   // 收到消息时(只有频道消息)
   ws.on('message', function (msg) {
     // 可能需要检查权限, 是否可以向指定目标发送, 或者由客户端过滤
     // 还需要在 data 中附带上发送者信息
-    let { fm, data } = msg
+    let { fm, data } = JSON.parse(msg)
     FM.发送消息(fm, uid, data)
   })