feat: CLI SUBSCRIBE/MONITOR streaming mode

- New pubsub.ts: startSubscribe/stopSubscribe/startMonitor/stopMonitor
- IPC: redis:subscribe, redis:unsubscribe, redis:monitor, redis:monitorStop
- Streaming messages pushed via webContents.send
- Preload: onSubscribeMessage/onMonitorMessage callbacks
- CliView: detect SUBSCRIBE/PSUBSCRIBE/MONITOR, switch to streaming mode
- Stop button with pulsing indicator (green=subscribe, yellow=monitor)
- Real-time message display with timestamps
- Auto-cleanup on unmount
- i18n keys (zh-CN + en)

ROADMAP.md

@@ -20,7 +20,7 @@
 - [x] **多连接并行** — 同时打开多个连接，各自独立标签页
 - [x] **Monaco 编辑器** — 替换 StringEditor 的 textarea，支持 JSON 折叠/展开/语法高亮
 - [x] **Redis 命令自动补全** — CLI 输入时智能提示（基于 commands.js 命令数据库）
-- [ ] **CLI SUBSCRIBE/MONITOR** — 实时消息流 + MONITOR 命令流 + 停止按钮
+- [x] **CLI SUBSCRIBE/MONITOR** — 实时消息流 + MONITOR 命令流 + 停止按钮
 - [ ] **CLI MULTI/EXEC** — 事务模式支持
 - [ ] **Stream 消费者组** — XINFO GROUPS / XINFO CONSUMERS 展开表格
 

electron/main/ipc-handlers.ts

@@ -239,4 +239,30 @@ export function registerIpcHandlers(): void {
   ipcMain.handle('redis:clearCommandLog', () => {
     redis.clearLog()
   })
+
+  // Pub/Sub
+  ipcMain.handle('redis:subscribe', async (_e, id: string, channels: string[], isPattern: boolean) => {
+    try {
+      redis.startSubscribe(id, channels, isPattern, _e.sender)
+      return { success: true }
+    } catch (err: any) {
+      return { success: false, error: err.message }
+    }
+  })
+  ipcMain.handle('redis:unsubscribe', async (_e, id: string) => {
+    redis.stopSubscribe(id)
+  })
+
+  // Monitor
+  ipcMain.handle('redis:monitor', async (_e, id: string) => {
+    try {
+      redis.startMonitor(id, _e.sender)
+      return { success: true }
+    } catch (err: any) {
+      return { success: false, error: err.message }
+    }
+  })
+  ipcMain.handle('redis:monitorStop', async (_e, id: string) => {
+    redis.stopMonitor(id)
+  })
 }

electron/main/redis/index.ts

@@ -10,3 +10,4 @@ export * from './zset'
 export * from './stream'
 export * from './server'
 export * from './commandLogger'
+export * from './pubsub'

electron/main/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)
+}

electron/preload/index.d.ts

@@ -104,6 +104,12 @@ export interface ElectronAPI {
     isConnected: (id: string) => Promise<boolean>
     getCommandLog: () => Promise<any[]>
     clearCommandLog: () => Promise<void>
+    subscribe: (id: string, channels: string[], isPattern: boolean) => Promise<{ success: boolean; error?: string }>
+    unsubscribe: (id: string) => Promise<void>
+    onSubscribeMessage: (callback: (msg: any) => void) => void
+    monitor: (id: string) => Promise<{ success: boolean; error?: string }>
+    monitorStop: (id: string) => Promise<void>
+    onMonitorMessage: (callback: (msg: any) => void) => void
   }
   updater: {
     check: () => Promise<{ available: boolean; version?: string }>

electron/preload/index.ts

@@ -119,6 +119,22 @@ const electronAPI = {
       ipcRenderer.invoke('redis:getCommandLog'),
     clearCommandLog: (): Promise<void> =>
       ipcRenderer.invoke('redis:clearCommandLog'),
+    // Pub/Sub
+    subscribe: (id: string, channels: string[], isPattern: boolean): Promise<any> =>
+      ipcRenderer.invoke('redis:subscribe', id, channels, isPattern),
+    unsubscribe: (id: string): Promise<void> =>
+      ipcRenderer.invoke('redis:unsubscribe', id),
+    onSubscribeMessage: (callback: (msg: any) => void) => {
+      ipcRenderer.on('redis:subscribeMessage', (_e, msg) => callback(msg))
+    },
+    // Monitor
+    monitor: (id: string): Promise<any> =>
+      ipcRenderer.invoke('redis:monitor', id),
+    monitorStop: (id: string): Promise<void> =>
+      ipcRenderer.invoke('redis:monitorStop', id),
+    onMonitorMessage: (callback: (msg: any) => void) => {
+      ipcRenderer.on('redis:monitorMessage', (_e, msg) => callback(msg))
+    },
   },
   updater: {
     check: (): Promise<any> => ipcRenderer.invoke('updater:check'),

src/components/CliView.vue

@@ -1,16 +1,20 @@
 <script setup lang="ts">
 // src/renderer/src/components/CliView.vue
-import { ref, computed, nextTick, watch } from 'vue'
+import { ref, computed, nextTick, watch, onMounted, onUnmounted } from 'vue'
 import { useConnectionStore } from '@renderer/stores/connection'
+import { useI18n } from '@renderer/i18n'
 import { redisCommands } from '@renderer/data/commands'
 
 const connStore = useConnectionStore()
+const { t } = useI18n()
 
 const input = ref('')
 const history = ref<{ cmd: string; result: string; error?: boolean }[]>([])
 const historyIndex = ref(-1)
 const outputRef = ref<HTMLDivElement>()
 const suggestionIndex = ref(-1)
+const inputRef = ref<HTMLInputElement>()
+const streamingMode = ref<'none' | 'subscribe' | 'monitor'>('none')
 
 const showSuggestions = computed(() => {
   return suggestions.value.length > 0 && !input.value.includes(' ')
@@ -44,6 +48,36 @@ async function execute() {
   const args = parts.slice(1)
 
   try {
+    // Handle SUBSCRIBE / PSUBSCRIBE
+    if (command === 'SUBSCRIBE' || command === 'PSUBSCRIBE') {
+      const channels = args.filter(a => !a.startsWith('"'))
+      if (channels.length === 0) {
+        history.value.push({ cmd, result: 'ERROR: No channel specified', error: true })
+        return
+      }
+      const isPattern = command === 'PSUBSCRIBE'
+      const result = await window.electronAPI.redis.subscribe(connStore.activeId, channels, isPattern)
+      if (result.success) {
+        streamingMode.value = 'subscribe'
+        history.value.push({ cmd, result: t('cli.subscribedTo', { channels: channels.join(', ') }) })
+      } else {
+        history.value.push({ cmd, result: `ERROR: ${result.error}`, error: true })
+      }
+      return
+    }
+
+    // Handle MONITOR
+    if (command === 'MONITOR') {
+      const result = await window.electronAPI.redis.monitor(connStore.activeId)
+      if (result.success) {
+        streamingMode.value = 'monitor'
+        history.value.push({ cmd, result: t('cli.monitoring') })
+      } else {
+        history.value.push({ cmd, result: `ERROR: ${result.error}`, error: true })
+      }
+      return
+    }
+
     const result = await window.electronAPI.redis.execute(connStore.activeId, command, args)
     let formatted = ''
 
@@ -124,8 +158,6 @@ function onKeydown(e: KeyboardEvent) {
   }
 }
 
-const inputRef = ref<HTMLInputElement>()
-
 function categoryColor(category: string): string {
   switch (category) {
     case 'admin': return 'var(--color-red)'
@@ -136,6 +168,52 @@ function categoryColor(category: string): string {
     default: return 'var(--text-primary)'
   }
 }
+
+async function stopStreaming() {
+  if (!connStore.activeId) return
+  if (streamingMode.value === 'subscribe') {
+    await window.electronAPI.redis.unsubscribe(connStore.activeId)
+    history.value.push({ cmd: 'UNSUBSCRIBE', result: 'OK' })
+  } else if (streamingMode.value === 'monitor') {
+    await window.electronAPI.redis.monitorStop(connStore.activeId)
+    history.value.push({ cmd: 'MONITOR STOP', result: 'OK' })
+  }
+  streamingMode.value = 'none'
+  await nextTick()
+  if (outputRef.value) outputRef.value.scrollTop = outputRef.value.scrollHeight
+}
+
+// IPC listeners for streaming messages
+function onSubscribeMessage(msg: any) {
+  if (msg.error) {
+    history.value.push({ cmd: '', result: `SUBSCRIBE ERROR: ${msg.error}`, error: true })
+  } else {
+    const ts = new Date().toLocaleTimeString()
+    history.value.push({ cmd: '', result: `[${ts}] ${msg.channel}: ${msg.message}` })
+  }
+  nextTick(() => { if (outputRef.value) outputRef.value.scrollTop = outputRef.value.scrollHeight })
+}
+
+function onMonitorMessage(msg: any) {
+  if (msg.error) {
+    history.value.push({ cmd: '', result: `MONITOR ERROR: ${msg.error}`, error: true })
+  } else {
+    const ts = msg.time || new Date().toLocaleTimeString()
+    history.value.push({ cmd: '', result: `[${ts}] ${msg.args.join(' ')}` })
+  }
+  nextTick(() => { if (outputRef.value) outputRef.value.scrollTop = outputRef.value.scrollHeight })
+}
+
+onMounted(() => {
+  window.electronAPI.redis.onSubscribeMessage(onSubscribeMessage)
+  window.electronAPI.redis.onMonitorMessage(onMonitorMessage)
+})
+
+onUnmounted(() => {
+  if (connStore.activeId && streamingMode.value !== 'none') {
+    stopStreaming()
+  }
+})
 </script>
 
 <template>
@@ -149,17 +227,18 @@ function categoryColor(category: string): string {
         <pre class="cli-result" :class="{ error: item.error }">{{ item.result }}</pre>
       </div>
       <div v-if="history.length === 0" class="cli-welcome">
-        Redis CLI - type commands below
+        {{ t('cli.welcome') }}
       </div>
     </div>
     <div class="cli-input-bar">
+      <template v-if="streamingMode === 'none'">
         <span class="prompt">&gt;</span>
         <div class="cli-input-wrapper">
           <input
             ref="inputRef"
             v-model="input"
             class="cli-input"
-          placeholder="Enter Redis command..."
+            :placeholder="t('cli.placeholder')"
             @keydown="onKeydown"
             autofocus
           />
@@ -178,6 +257,14 @@ function categoryColor(category: string): string {
             </div>
           </div>
         </div>
+      </template>
+      <template v-else>
+        <span class="stream-indicator" :class="streamingMode" />
+        <span class="stream-text">{{ streamingMode === 'subscribe' ? t('cli.subscribed') : t('cli.monitoring') }}</span>
+        <el-button size="small" type="danger" @click="stopStreaming">
+          {{ streamingMode === 'subscribe' ? t('cli.stopSubscribe') : t('cli.stopMonitor') }}
+        </el-button>
+      </template>
     </div>
   </div>
 </template>
@@ -308,4 +395,31 @@ function categoryColor(category: string): string {
   font-size: 11px;
   color: var(--text-secondary);
 }
+
+.stream-indicator {
+  width: 8px;
+  height: 8px;
+  border-radius: 50%;
+  background: var(--success);
+  box-shadow: 0 0 8px var(--success);
+  animation: pulse 1.5s infinite;
+}
+
+.stream-indicator.monitor {
+  background: var(--warning);
+  box-shadow: 0 0 8px var(--warning);
+}
+
+@keyframes pulse {
+  0%, 100% { opacity: 1; }
+  50% { opacity: 0.4; }
+}
+
+.stream-text {
+  flex: 1;
+  font-size: 12px;
+  color: var(--text-secondary);
+  font-family: 'Cascadia Code', 'Fira Code', monospace;
+  margin-left: 6px;
+}
 </style>

src/i18n/locales/en.ts

@@ -137,6 +137,11 @@ export default {
     title: 'CLI',
     welcome: 'Redis CLI - type commands below',
     placeholder: 'Enter Redis command...',
+    stopSubscribe: 'Stop Subscribe',
+    stopMonitor: 'Stop Monitor',
+    subscribed: 'Subscribed, waiting for messages...',
+    monitoring: 'Monitoring, waiting for commands...',
+    subscribedTo: 'Subscribed to: {channels}',
   },
   slowlog: {
     title: 'Slow Log',

src/i18n/locales/zh-CN.ts

@@ -137,6 +137,11 @@ export default {
     title: 'CLI',
     welcome: 'Redis CLI - 在下方输入命令',
     placeholder: '输入 Redis 命令...',
+    stopSubscribe: '停止订阅',
+    stopMonitor: '停止监控',
+    subscribed: '已订阅，等待消息...',
+    monitoring: '监控中，等待命令...',
+    subscribedTo: '已订阅频道: {channels}',
   },
   slowlog: {
     title: '慢日志',