update

electron/redis/pubsub.ts

@@ -0,0 +1,79 @@
+// electron/main/redis/pubsub.ts
+// 发布订阅 + MONITOR 流式接口
+import Redis from 'ioredis'
+import { getClient } from './connection'
+import type { WebContents } from 'electron'
+
+const subClients = new Map<string, Redis>()
+const monitorClients = new Map<string, Redis>()
+
+export function startSubscribe(
+  id: string,
+  channels: string[],
+  isPattern: boolean,
+  sender: WebContents
+): void {
+  stopSubscribe(id)
+  const existing = getClient(id)
+  if (!existing) throw new Error('No active connection')
+
+  const sub = new Redis(existing.options)
+  subClients.set(id, sub)
+
+  if (isPattern) {
+    sub.psubscribe(...channels)
+    sub.on('pmessage', (_pattern, channel, message) => {
+      sender.send('redis:subscribeMessage', { channel, message, pattern: _pattern })
+    })
+  } else {
+    sub.subscribe(...channels)
+    sub.on('message', (channel, message) => {
+      sender.send('redis:subscribeMessage', { channel, message })
+    })
+  }
+
+  sub.on('error', (err) => {
+    sender.send('redis:subscribeMessage', { error: err.message })
+  })
+}
+
+export function stopSubscribe(id: string): void {
+  const sub = subClients.get(id)
+  if (sub) {
+    sub.unsubscribe()
+    sub.punsubscribe()
+    sub.disconnect()
+    subClients.delete(id)
+  }
+}
+
+export function startMonitor(id: string, sender: WebContents): void {
+  stopMonitor(id)
+  const existing = getClient(id)
+  if (!existing) throw new Error('No active connection')
+
+  const monitor = new Redis(existing.options)
+  monitorClients.set(id, monitor)
+
+  monitor.monitor().then((mon) => {
+    mon.on('monitor', (time: string, args: string[], source: string, database: number) => {
+      sender.send('redis:monitorMessage', { time, args, source, database })
+    })
+    mon.on('error', (err: Error) => {
+      sender.send('redis:monitorMessage', { error: err.message })
+    })
+  })
+}
+
+export function stopMonitor(id: string): void {
+  const mon = monitorClients.get(id)
+  if (mon) {
+    mon.disconnect()
+    monitorClients.delete(id)
+  }
+}
+
+export function stopAllPubSub(id: string): void {
+  stopSubscribe(id)
+  stopMonitor(id)
+}