.

package.json

@@ -54,6 +54,7 @@
     "crypto": "^1.0.1",
     "date-fns": "^2.29.3",
     "dedent": "^1.5.1",
+    "douyudm": "^2.1.1",
     "eventsource-parser": "^1.0.0",
     "express-basic-auth": "^1.2.1",
     "express-handlebars": "^7.0.6",

src/danmu/danmu.service.ts

@@ -18,13 +18,14 @@ import { getEncodeHeader } from '../utils/crypto'
 import * as WebSocket from 'ws'
 import { Proto } from './models/proto.model'
 import { Interval } from '@nestjs/schedule'
-
+import { fork, ChildProcess } from 'child_process'
 @Injectable()
 export class DanmuService implements OnModuleInit {
     private readonly logger = new Logger(DanmuService.name)
     client: Client
     biliApi: AxiosInstance
     biliWs: { [roomId: number]: WebSocket } = {}
+    douyuProcess: { [roomId: number]: ChildProcess } = {}
     constructor(
         @InjectRepository(Danmu)
         private readonly danmuRepository: Repository<Danmu>,
@@ -226,6 +227,45 @@ export class DanmuService implements OnModuleInit {
         }
     }
 
+    async handleDouyuMessage(data: any) {
+        this.logger.debug('收到弹幕')
+        const channel = `${data.rid}`
+        const platformUserId = `${data.uid}`
+        const name = data.nn
+        const avatar = `https://apic.douyucdn.cn/upload/${data.ic}_middle.jpg`
+        const msgId = data.cid
+        const room = await this.roomService.findActiveRoom(channel)
+        if (room) {
+            let danmuUser = await this.danmuUserRepository.findOne({
+                where: { roomId: room.id, platform: StreamPlatform.Douyu, platformUserId }
+            })
+            if (!danmuUser) {
+                danmuUser = await this.danmuUserRepository.save(
+                    new DanmuUser({
+                        roomId: room.id,
+                        platform: StreamPlatform.Douyu,
+                        platformUserId,
+                        name,
+                        avatar
+                    })
+                )
+            }
+            await this.danmuRepository.save(
+                new Danmu({
+                    platform: StreamPlatform.Douyu,
+                    danmuUserId: danmuUser.id,
+                    channel,
+                    platformRoomId: channel,
+                    platformUserId,
+                    content: data.txt,
+                    roomId: room.id,
+                    gameId: room.currentGameId,
+                    msgId
+                })
+            )
+        }
+    }
+
     async startDanmu(room: Room) {
         if (room.platform === StreamPlatform.Twitch) {
             try {
@@ -261,6 +301,10 @@ export class DanmuService implements OnModuleInit {
                 this.logger.error(`开启直播间失败: ${JSON.stringify(data)}`)
                 throw new InternalServerErrorException(data.message)
             }
+        } else if (room.platform === StreamPlatform.Douyu) {
+            const child = fork(path.join(__dirname, './douyu_process.js'), [room.channelId])
+            child.on('message', this.handleDouyuMessage.bind(this))
+            this.douyuProcess[room.id] = child
         }
     }
 
@@ -284,6 +328,11 @@ export class DanmuService implements OnModuleInit {
                 this.biliWs[room.id].close()
                 delete this.biliWs[room.id]
             }
+        } else if (room.platform === StreamPlatform.Douyu) {
+            if (this.douyuProcess[room.id]) {
+                this.douyuProcess[room.id].kill()
+                delete this.douyuProcess[room.id]
+            }
         }
     }
 

src/danmu/douyu_process.ts

@@ -0,0 +1,25 @@
+const douyudm = require('douyudm')
+const logger = require('@nestjs/common').Logger
+
+const roomId = parseInt(process.argv[2])
+
+const room = new douyudm(roomId)
+
+//系统事件
+room.on('connect', function () {
+    logger.log('[connect] roomId=%s', this.roomId)
+})
+room.on('disconnect', function () {
+    logger.log('[disconnect] roomId=%s', this.roomId)
+})
+room.on('error', function (err) {
+    logger.log('[error] roomId=%s', this.roomId)
+})
+
+//消息事件
+room.on('chatmsg', function (res) {
+    process.send(res)
+})
+ 
+//开始监听
+room.run()

yarn.lock

@@ -2971,7 +2971,7 @@ combined-stream@^1.0.6, combined-stream@^1.0.8, combined-stream@~1.0.6:
   dependencies:
     delayed-stream "~1.0.0"
 
-commander@4.1.1:
+commander@4.1.1, commander@^4.1.0:
   version "4.1.1"
   resolved "https://registry.npmjs.org/commander/-/commander-4.1.1.tgz"
   integrity sha512-NOKm8xhkzAjzFx8B2v5OAHT+u5pRQc2UCa2Vq9jYL/31o2wi9mxBA7LIFs3sV5VSC49z6pEhfbMULvShKj26WA==
@@ -3381,6 +3381,16 @@ dotenv@16.0.3, dotenv@^16.0.3:
   resolved "https://registry.npmjs.org/dotenv/-/dotenv-16.0.3.tgz"
   integrity sha512-7GO6HghkA5fYG9TYnNxi14/7K9f5occMlp3zXAuSxn7CKCxt9xbNWG7yF8hTCSUchlfWSe3uLmlPfigevRItzQ==
 
+douyudm@^2.1.1:
+  version "2.1.1"
+  resolved "https://registry.npmmirror.com/douyudm/-/douyudm-2.1.1.tgz#a8e6bc449fb5ee03f50c5fb17d582782656d6a66"
+  integrity sha512-2LBH+e/fGfpFqlSrm6pHvDNSojqoyweHFxgnP4JRw7eYSpq2jsOP0fKtTSTRfI2WGPWnxcxr+rV6U+C8wfHuYw==
+  dependencies:
+    commander "^4.1.0"
+    fast-text-encoding "^1.0.0"
+    lowdb "^1.0.0"
+    ws "^7.2.1"
+
 eastasianwidth@^0.2.0:
   version "0.2.0"
   resolved "https://registry.npmmirror.com/eastasianwidth/-/eastasianwidth-0.2.0.tgz#696ce2ec0aa0e6ea93a397ffcf24aa7840c827cb"
@@ -3921,6 +3931,11 @@ fast-safe-stringify@2.1.1, fast-safe-stringify@^2.1.1:
   resolved "https://registry.npmjs.org/fast-safe-stringify/-/fast-safe-stringify-2.1.1.tgz"
   integrity sha512-W+KJc2dmILlPplD/H4K9l9LcAHAfPtP6BY84uVLXQ6Evcz9Lcg33Y2z1IVblT6xdY54PXYVHEv+0Wpq8Io6zkA==
 
+fast-text-encoding@^1.0.0:
+  version "1.0.6"
+  resolved "https://registry.npmmirror.com/fast-text-encoding/-/fast-text-encoding-1.0.6.tgz#0aa25f7f638222e3396d72bf936afcf1d42d6867"
+  integrity sha512-VhXlQgj9ioXCqGstD37E/HBeqEGV/qOD/kmbVG8h5xKBYvM1L3lR1Zn4555cQ8GkYbJa8aJSipLPndE1k6zK2w==
+
 fastq@^1.6.0:
   version "1.15.0"
   resolved "https://registry.npmjs.org/fastq/-/fastq-1.15.0.tgz"
@@ -4397,7 +4412,7 @@ graceful-fs@^4.1.2, graceful-fs@^4.1.6, graceful-fs@^4.2.0, graceful-fs@^4.2.10,
   resolved "https://registry.npmjs.org/graceful-fs/-/graceful-fs-4.2.10.tgz"
   integrity sha512-9ByhssR2fPVsNZj478qUUbKfmL0+t5BDVyjShtyZZLiK7ZDAArFFfopyOTj0M05wE2tJPisA4iTnnXl2YoPvOA==
 
-graceful-fs@^4.2.11:
+graceful-fs@^4.1.3, graceful-fs@^4.2.11:
   version "4.2.11"
   resolved "https://registry.npmmirror.com/graceful-fs/-/graceful-fs-4.2.11.tgz#4183e4e8bf08bb6e05bbb2f7d2e0c8f712ca40e3"
   integrity sha512-RbJ5/jmFcNNCcDV5o9eTnBLJ/HszWV0P73bc+Ff4nS/rJj+YaS6IGyiOL0VoBYX+l1Wrl3k63h/KrH+nhJ0XvQ==
@@ -5041,6 +5056,11 @@ is-plain-object@^2.0.3, is-plain-object@^2.0.4:
   dependencies:
     isobject "^3.0.1"
 
+is-promise@^2.1.0:
+  version "2.2.2"
+  resolved "https://registry.npmmirror.com/is-promise/-/is-promise-2.2.2.tgz#39ab959ccbf9a774cf079f7b40c7a26f763135f1"
+  integrity sha512-+lP4/6lKUBfQjZ2pdxThZvLUAafmZb8OAxFb8XXtiQmS35INgr85hdOGoEs124ez1FCnZJt6jau/T+alh58QFQ==
+
 is-property@^1.0.2:
   version "1.0.2"
   resolved "https://registry.npmjs.org/is-property/-/is-property-1.0.2.tgz"
@@ -5910,7 +5930,7 @@ lodash.templatesettings@^4.0.0:
   dependencies:
     lodash._reinterpolate "^3.0.0"
 
-lodash@4.17.21, lodash@^4.17.19, lodash@^4.17.21:
+lodash@4, lodash@4.17.21, lodash@^4.17.19, lodash@^4.17.21:
   version "4.17.21"
   resolved "https://registry.npmjs.org/lodash/-/lodash-4.17.21.tgz"
   integrity sha512-v2kDEe57lecTulaDIuNTPy3Ry4gLGJ6Z1O3vE1krgXZNrsQ+LFTGHVxVjcXPs17LhbZVGedAJv8XZ1tvj5FvSg==
@@ -5965,6 +5985,17 @@ long@^5.2.1:
   resolved "https://registry.npmjs.org/long/-/long-5.2.1.tgz"
   integrity sha512-GKSNGeNAtw8IryjjkhZxuKB3JzlcLTwjtiQCHKvqQet81I93kXslhDQruGI/QsddO83mcDToBVy7GqGS/zYf/A==
 
+lowdb@^1.0.0:
+  version "1.0.0"
+  resolved "https://registry.npmmirror.com/lowdb/-/lowdb-1.0.0.tgz#5243be6b22786ccce30e50c9a33eac36b20c8064"
+  integrity sha512-2+x8esE/Wb9SQ1F9IHaYWfsC9FIecLOPrK4g17FGEayjUWH172H6nwicRovGvSE2CPZouc2MCIqCI7h9d+GftQ==
+  dependencies:
+    graceful-fs "^4.1.3"
+    is-promise "^2.1.0"
+    lodash "4"
+    pify "^3.0.0"
+    steno "^0.4.1"
+
 lru-cache@^5.1.1:
   version "5.1.1"
   resolved "https://registry.npmjs.org/lru-cache/-/lru-cache-5.1.1.tgz"
@@ -6959,6 +6990,11 @@ picomatch@^2.0.4, picomatch@^2.2.1, picomatch@^2.2.3, picomatch@^2.3.1:
   resolved "https://registry.npmjs.org/picomatch/-/picomatch-2.3.1.tgz"
   integrity sha512-JU3teHTNjmE2VCGFzuY8EXzCDVwEqB2a8fsIvwaStHhAWJEeVd1o1QD80CU6+ZdEXXSLbSsuLwJjkCBWqRQUVA==
 
+pify@^3.0.0:
+  version "3.0.0"
+  resolved "https://registry.npmmirror.com/pify/-/pify-3.0.0.tgz#e5a4acd2c101fdf3d9a4d07f0dbc4db49dd28176"
+  integrity sha512-C3FsVNH1udSEX48gGX1xfvwTWfsYWj5U+8/uK15BGzIGrKoUpghX8hWZwa/OFnakBiiVNmBvemTJR5mcy7iPcg==
+
 pirates@^4.0.4:
   version "4.0.5"
   resolved "https://registry.npmjs.org/pirates/-/pirates-4.0.5.tgz"
@@ -7845,6 +7881,13 @@ stdin-discarder@^0.1.0:
   dependencies:
     bl "^5.0.0"
 
+steno@^0.4.1:
+  version "0.4.4"
+  resolved "https://registry.npmmirror.com/steno/-/steno-0.4.4.tgz#071105bdfc286e6615c0403c27e9d7b5dcb855cb"
+  integrity sha512-EEHMVYHNXFHfGtgjNITnka0aHhiAlo93F7z2/Pwd+g0teG9CnM3JIINM7hVVB5/rhw9voufD7Wukwgtw2uqh6w==
+  dependencies:
+    graceful-fs "^4.1.3"
+
 stream-http@2.8.2:
   version "2.8.2"
   resolved "https://registry.npmmirror.com/stream-http/-/stream-http-2.8.2.tgz#4126e8c6b107004465918aa2fc35549e77402c87"
@@ -8773,6 +8816,11 @@ write-file-atomic@^4.0.2:
     imurmurhash "^0.1.4"
     signal-exit "^3.0.7"
 
+ws@^7.2.1:
+  version "7.5.9"
+  resolved "https://registry.npmmirror.com/ws/-/ws-7.5.9.tgz#54fa7db29f4c7cec68b1ddd3a89de099942bb591"
+  integrity sha512-F+P9Jil7UiSKSkppIiD94dN07AwvFixvLIj1Og1Rl9GGMuNipJnV9JzjD6XuqmAeiswGvUmNLjr5cFuXwNS77Q==
+
 ws@^8.14.2, ws@^8.2.0:
   version "8.14.2"
   resolved "https://registry.npmmirror.com/ws/-/ws-8.14.2.tgz#6c249a806eb2db7a20d26d51e7709eab7b2e6c7f"