diff --git a/ROADMAP.md b/ROADMAP.md index 6b790ee..0220f40 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -22,7 +22,7 @@ - [x] **Redis 命令自动补全** — CLI 输入时智能提示(基于 commands.js 命令数据库) - [x] **CLI SUBSCRIBE/MONITOR** — 实时消息流 + MONITOR 命令流 + 停止按钮 - [x] **CLI MULTI/EXEC** — 事务模式支持 -- [ ] **Stream 消费者组** — XINFO GROUPS / XINFO CONSUMERS 展开表格 +- [x] **Stream 消费者组** — XINFO GROUPS / XINFO CONSUMERS 展开表格 ## P2 — 重要功能增强 diff --git a/electron/main/ipc-handlers.ts b/electron/main/ipc-handlers.ts index f5b2ea0..59324e2 100644 --- a/electron/main/ipc-handlers.ts +++ b/electron/main/ipc-handlers.ts @@ -232,6 +232,23 @@ export function registerIpcHandlers(): void { return redis.streamDel(id, key, ...ids) })) + // Redis - Stream Consumer Groups + ipcMain.handle('redis:streamGroups', async (_e, id: string, key: string) => { + return redis.streamGroups(id, key) + }) + ipcMain.handle('redis:streamConsumers', async (_e, id: string, key: string, group: string) => { + return redis.streamConsumers(id, key, group) + }) + ipcMain.handle('redis:streamGroupCreate', withLog('XGROUP CREATE', async (_e, id: string, key: string, group: string, startId: string, mkstream: boolean) => { + return redis.streamGroupCreate(id, key, group, startId, mkstream) + })) + ipcMain.handle('redis:streamGroupDestroy', withLog('XGROUP DESTROY', async (_e, id: string, key: string, group: string) => { + return redis.streamGroupDestroy(id, key, group) + })) + ipcMain.handle('redis:streamAck', withLog('XACK', async (_e, id: string, key: string, group: string, ids: string[]) => { + return redis.streamAck(id, key, group, ...ids) + })) + // Command Log ipcMain.handle('redis:getCommandLog', () => { return redis.getLog() diff --git a/electron/main/redis/stream.ts b/electron/main/redis/stream.ts index 7cf8115..6d700cf 100644 --- a/electron/main/redis/stream.ts +++ b/electron/main/redis/stream.ts @@ -35,3 +35,38 @@ export async function streamDel(id: string, key: string, ...ids: string[]): Prom if (!client) throw new Error(`Client ${id} not found`) return client.xdel(key, ...ids) } + +// Consumer group operations +export async function streamGroups(id: string, key: string): Promise { + const client = getClient(id) + if (!client) throw new Error(`Client ${id} not found`) + return client.xinfo('GROUPS', key) as Promise +} + +export async function streamConsumers(id: string, key: string, group: string): Promise { + const client = getClient(id) + if (!client) throw new Error(`Client ${id} not found`) + return client.xinfo('CONSUMERS', key, group) as Promise +} + +export async function streamGroupCreate( + id: string, key: string, group: string, startId: string, mkstream: boolean +): Promise { + const client = getClient(id) + if (!client) throw new Error(`Client ${id} not found`) + const args: any[] = [key, group, startId] + if (mkstream) args.push('MKSTREAM') + return (client as any).xgroup('CREATE', ...args) +} + +export async function streamGroupDestroy(id: string, key: string, group: string): Promise { + const client = getClient(id) + if (!client) throw new Error(`Client ${id} not found`) + return (client as any).xgroup('DESTROY', key, group) +} + +export async function streamAck(id: string, key: string, group: string, ...ids: string[]): Promise { + const client = getClient(id) + if (!client) throw new Error(`Client ${id} not found`) + return (client as any).xack(key, group, ...ids) +} diff --git a/electron/preload/index.d.ts b/electron/preload/index.d.ts index 098b9b1..d643b27 100644 --- a/electron/preload/index.d.ts +++ b/electron/preload/index.d.ts @@ -98,6 +98,11 @@ export interface ElectronAPI { streamAdd: (id: string, key: string, streamId: string, fieldValues: string[]) => Promise streamTrim: (id: string, key: string, maxLen: number) => Promise streamDel: (id: string, key: string, ids: string[]) => Promise + streamGroups: (id: string, key: string) => Promise + streamConsumers: (id: string, key: string, group: string) => Promise + streamGroupCreate: (id: string, key: string, group: string, startId: string, mkstream: boolean) => Promise + streamGroupDestroy: (id: string, key: string, group: string) => Promise + streamAck: (id: string, key: string, group: string, ids: string[]) => Promise renameKey: (id: string, oldKey: string, newKey: string) => Promise existsKey: (id: string, key: string) => Promise ping: (id: string) => Promise diff --git a/electron/preload/index.ts b/electron/preload/index.ts index 3205785..7411165 100644 --- a/electron/preload/index.ts +++ b/electron/preload/index.ts @@ -105,6 +105,17 @@ const electronAPI = { ipcRenderer.invoke('redis:streamTrim', id, key, maxLen), streamDel: (id: string, key: string, ids: string[]): Promise => ipcRenderer.invoke('redis:streamDel', id, key, ids), + // Stream consumer groups + streamGroups: (id: string, key: string): Promise => + ipcRenderer.invoke('redis:streamGroups', id, key), + streamConsumers: (id: string, key: string, group: string): Promise => + ipcRenderer.invoke('redis:streamConsumers', id, key, group), + streamGroupCreate: (id: string, key: string, group: string, startId: string, mkstream: boolean): Promise => + ipcRenderer.invoke('redis:streamGroupCreate', id, key, group, startId, mkstream), + streamGroupDestroy: (id: string, key: string, group: string): Promise => + ipcRenderer.invoke('redis:streamGroupDestroy', id, key, group), + streamAck: (id: string, key: string, group: string, ids: string[]): Promise => + ipcRenderer.invoke('redis:streamAck', id, key, group, ids), // Key ops renameKey: (id: string, oldKey: string, newKey: string): Promise => ipcRenderer.invoke('redis:renameKey', id, oldKey, newKey), diff --git a/src/components/StreamEditor.vue b/src/components/StreamEditor.vue index 43c3791..047bba7 100644 --- a/src/components/StreamEditor.vue +++ b/src/components/StreamEditor.vue @@ -15,6 +15,19 @@ interface StreamEntry { fields: Record } +interface ConsumerGroup { + name: string + consumers: number + pending: number + lastDeliveredId: string +} + +interface Consumer { + name: string + pending: number + idle: number +} + const entries = ref([]) const loading = ref(false) const showAddDialog = ref(false) @@ -23,6 +36,15 @@ const newFields = ref<{ key: string; value: string }[]>([{ key: '', value: '' }] const trimming = ref(false) const trimMaxLen = ref(1000) +const groups = ref([]) +const expandedGroup = ref(null) +const consumers = ref>({}) +const showGroupDialog = ref(false) +const newGroupName = ref('') +const newGroupStartId = ref('$') +const newGroupMkstream = ref(false) +const groupsLoading = ref(false) + async function loadEntries() { if (!connStore.activeId || !keyStore.selectedKey) return loading.value = true @@ -47,9 +69,109 @@ function parseFields(raw: any[]): Record { return result } +function parseGroupInfo(raw: any[]): ConsumerGroup { + const map: Record = {} + for (let i = 0; i < raw.length; i += 2) { + map[raw[i]] = raw[i + 1] + } + return { + name: map['name'] as string, + consumers: Number(map['consumers'] ?? 0), + pending: Number(map['pending'] ?? 0), + lastDeliveredId: map['last-delivered-id'] as string, + } +} + +function parseConsumerInfo(raw: any[]): Consumer { + const map: Record = {} + for (let i = 0; i < raw.length; i += 2) { + map[raw[i]] = raw[i + 1] + } + return { + name: map['name'] as string, + pending: Number(map['pending'] ?? 0), + idle: Number(map['idle'] ?? 0), + } +} + +async function loadGroups() { + if (!connStore.activeId || !keyStore.selectedKey) return + groupsLoading.value = true + try { + const rawGroups = await window.electronAPI.redis.streamGroups( + connStore.activeId, keyStore.selectedKey + ) + groups.value = (rawGroups as any[]).map(parseGroupInfo) + } catch { + groups.value = [] + } finally { + groupsLoading.value = false + } +} + +async function toggleGroup(groupName: string) { + if (expandedGroup.value === groupName) { + expandedGroup.value = null + return + } + expandedGroup.value = groupName + if (!consumers.value[groupName]) { + try { + const rawConsumers = await window.electronAPI.redis.streamConsumers( + connStore.activeId!, keyStore.selectedKey!, groupName + ) + consumers.value[groupName] = (rawConsumers as any[]).map(parseConsumerInfo) + } catch { + consumers.value[groupName] = [] + } + } +} + +async function createGroup() { + if (!connStore.activeId || !keyStore.selectedKey || !newGroupName.value) return + try { + await window.electronAPI.redis.streamGroupCreate( + connStore.activeId, keyStore.selectedKey, + newGroupName.value, newGroupStartId.value || '$', newGroupMkstream.value + ) + showGroupDialog.value = false + newGroupName.value = '' + newGroupStartId.value = '$' + newGroupMkstream.value = false + await loadGroups() + } catch (err: any) { + ElMessage.error(err.message) + } +} + +async function destroyGroup(groupName: string) { + if (!connStore.activeId || !keyStore.selectedKey) return + try { + await ElMessageBox.confirm( + `Destroy consumer group "${groupName}"?`, 'Confirm', + { type: 'warning' } + ) + await window.electronAPI.redis.streamGroupDestroy( + connStore.activeId, keyStore.selectedKey, groupName + ) + if (expandedGroup.value === groupName) { + expandedGroup.value = null + delete consumers.value[groupName] + } + await loadGroups() + } catch (err: any) { + if (err !== 'cancel') ElMessage.error(err.message) + } +} + watch( () => keyStore.selectedKey, - () => { if (keyStore.selectedType === 'stream') loadEntries() }, + () => { + if (keyStore.selectedType === 'stream') { + loadEntries() + loadGroups() + } + }, { immediate: true } ) @@ -133,7 +255,36 @@ async function trimStream() { {{ t('editor.trim') }} - {{ t('common.refresh') }} + {{ t('common.refresh') }} + + +
+
+ {{ t('editor.consumerGroups') }} + {{ groups.length }} + {{ t('editor.createGroup') }} +
+
+
+
+ + {{ group.name }} + {{ t('editor.consumers') }}: {{ group.consumers }} + {{ t('editor.pending') }}: {{ group.pending }} + {{ group.lastDeliveredId }} + {{ t('editor.destroyGroup') }} +
+
+
+ {{ consumer.name }} + {{ t('editor.pending') }}: {{ consumer.pending }} + {{ t('editor.consumers') }}: {{ Math.round(consumer.idle / 1000) }}s +
+
{{ t('editor.empty') }}
+
+
+
{{ t('editor.noGroups') }}
+
@@ -175,6 +326,24 @@ async function trimStream() { + + + + + + + + + + {{ t('editor.mkstream') }} + + + + + @@ -214,6 +383,120 @@ async function trimStream() { color: var(--text-muted); } +.groups-panel { + max-height: 200px; + overflow-y: auto; + border-bottom: 1px solid var(--border-color); + background: var(--bg-secondary); + flex-shrink: 0; +} + +.groups-header { + display: flex; + align-items: center; + gap: 8px; + padding: 8px 12px; + border-bottom: 1px solid var(--border-color); +} + +.groups-title { + font-size: 12px; + font-weight: 600; + color: var(--text-primary); +} + +.groups-count { + font-size: 10px; + color: var(--text-muted); + background: var(--bg-tertiary); + padding: 1px 6px; + border-radius: 8px; + margin-right: auto; +} + +.groups-list { + padding: 4px 0; +} + +.group-item { + border-bottom: 1px solid var(--border-subtle); +} + +.group-row { + display: flex; + align-items: center; + gap: 8px; + padding: 6px 12px; + cursor: pointer; + font-size: 12px; +} + +.group-row:hover { + background: var(--bg-hover); +} + +.group-arrow { + font-size: 10px; + color: var(--text-muted); + transition: transform 0.15s ease; + flex-shrink: 0; + width: 12px; + text-align: center; +} + +.group-arrow.expanded { + transform: rotate(90deg); +} + +.group-name { + font-weight: 600; + color: var(--text-primary); + min-width: 80px; +} + +.group-stat { + font-size: 11px; + color: var(--text-secondary); +} + +.group-last-id { + font-size: 10px; + font-family: 'Cascadia Code', monospace; + color: var(--text-muted); + margin-left: auto; +} + +.consumers-list { + padding: 2px 0 4px 32px; + background: var(--bg-tertiary); +} + +.consumer-row { + display: flex; + align-items: center; + gap: 12px; + padding: 4px 8px; + font-size: 11px; +} + +.consumer-name { + color: var(--text-primary); + font-family: 'Cascadia Code', monospace; + min-width: 100px; +} + +.consumer-stat { + color: var(--text-secondary); +} + +.no-consumers, +.no-groups { + padding: 12px; + text-align: center; + color: var(--text-muted); + font-size: 12px; +} + .stream-content { flex: 1; overflow-y: auto; diff --git a/src/i18n/locales/en.ts b/src/i18n/locales/en.ts index 90970b3..88f6ac9 100644 --- a/src/i18n/locales/en.ts +++ b/src/i18n/locales/en.ts @@ -131,7 +131,19 @@ export default { trimConfirm: 'Trim stream?', entryId: 'ID (use * for auto)', fields: 'Fields', - maxLength: 'Max length', + maxLength: 'Max Length', + consumerGroups: 'Consumer Groups', + group: 'Group', + consumers: 'Consumers', + pending: 'Pending', + createGroup: 'Create Group', + destroyGroup: 'Destroy Group', + groupName: 'Group Name', + startId: 'Start ID', + mkstream: 'Auto-create stream', + ack: 'ACK', + ackConfirm: 'ACK selected entries?', + noGroups: 'No consumer groups', }, cli: { title: 'CLI', diff --git a/src/i18n/locales/zh-CN.ts b/src/i18n/locales/zh-CN.ts index 8a4b42f..8cb7334 100644 --- a/src/i18n/locales/zh-CN.ts +++ b/src/i18n/locales/zh-CN.ts @@ -132,6 +132,18 @@ export default { entryId: 'ID (使用 * 自动生成)', fields: '字段', maxLength: '最大长度', + consumerGroups: '消费者组', + group: '组名', + consumers: '消费者', + pending: '待处理', + createGroup: '创建组', + destroyGroup: '删除组', + groupName: '组名称', + startId: '起始 ID', + mkstream: '自动创建流', + ack: 'ACK', + ackConfirm: '确认 ACK 选中的条目?', + noGroups: '暂无消费者组', }, cli: { title: 'CLI',