feat: Stream consumer groups UI
- stream.ts: streamGroups, streamConsumers, streamGroupCreate, streamGroupDestroy, streamAck - IPC: redis:streamGroups, streamConsumers, streamGroupCreate, streamGroupDestroy, streamAck - Preload bridge + types for all new methods - StreamEditor.vue: collapsible consumer groups panel - XINFO GROUPS table with name/consumers/pending/last-delivered-id - Click group to expand XINFO CONSUMERS - Create group dialog (name, start ID, MKSTREAM) - Destroy group with confirm - i18n keys (zh-CN + en) - P1 complete!
This commit is contained in:
+1
-1
@@ -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 — 重要功能增强
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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<any[]> {
|
||||
const client = getClient(id)
|
||||
if (!client) throw new Error(`Client ${id} not found`)
|
||||
return client.xinfo('GROUPS', key) as Promise<any[]>
|
||||
}
|
||||
|
||||
export async function streamConsumers(id: string, key: string, group: string): Promise<any[]> {
|
||||
const client = getClient(id)
|
||||
if (!client) throw new Error(`Client ${id} not found`)
|
||||
return client.xinfo('CONSUMERS', key, group) as Promise<any[]>
|
||||
}
|
||||
|
||||
export async function streamGroupCreate(
|
||||
id: string, key: string, group: string, startId: string, mkstream: boolean
|
||||
): Promise<string> {
|
||||
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<number> {
|
||||
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<number> {
|
||||
const client = getClient(id)
|
||||
if (!client) throw new Error(`Client ${id} not found`)
|
||||
return (client as any).xack(key, group, ...ids)
|
||||
}
|
||||
|
||||
Vendored
+5
@@ -98,6 +98,11 @@ export interface ElectronAPI {
|
||||
streamAdd: (id: string, key: string, streamId: string, fieldValues: string[]) => Promise<string>
|
||||
streamTrim: (id: string, key: string, maxLen: number) => Promise<number>
|
||||
streamDel: (id: string, key: string, ids: string[]) => Promise<number>
|
||||
streamGroups: (id: string, key: string) => Promise<any[]>
|
||||
streamConsumers: (id: string, key: string, group: string) => Promise<any[]>
|
||||
streamGroupCreate: (id: string, key: string, group: string, startId: string, mkstream: boolean) => Promise<any>
|
||||
streamGroupDestroy: (id: string, key: string, group: string) => Promise<number>
|
||||
streamAck: (id: string, key: string, group: string, ids: string[]) => Promise<number>
|
||||
renameKey: (id: string, oldKey: string, newKey: string) => Promise<void>
|
||||
existsKey: (id: string, key: string) => Promise<boolean>
|
||||
ping: (id: string) => Promise<string>
|
||||
|
||||
@@ -105,6 +105,17 @@ const electronAPI = {
|
||||
ipcRenderer.invoke('redis:streamTrim', id, key, maxLen),
|
||||
streamDel: (id: string, key: string, ids: string[]): Promise<number> =>
|
||||
ipcRenderer.invoke('redis:streamDel', id, key, ids),
|
||||
// Stream consumer groups
|
||||
streamGroups: (id: string, key: string): Promise<any[]> =>
|
||||
ipcRenderer.invoke('redis:streamGroups', id, key),
|
||||
streamConsumers: (id: string, key: string, group: string): Promise<any[]> =>
|
||||
ipcRenderer.invoke('redis:streamConsumers', id, key, group),
|
||||
streamGroupCreate: (id: string, key: string, group: string, startId: string, mkstream: boolean): Promise<any> =>
|
||||
ipcRenderer.invoke('redis:streamGroupCreate', id, key, group, startId, mkstream),
|
||||
streamGroupDestroy: (id: string, key: string, group: string): Promise<number> =>
|
||||
ipcRenderer.invoke('redis:streamGroupDestroy', id, key, group),
|
||||
streamAck: (id: string, key: string, group: string, ids: string[]): Promise<number> =>
|
||||
ipcRenderer.invoke('redis:streamAck', id, key, group, ids),
|
||||
// Key ops
|
||||
renameKey: (id: string, oldKey: string, newKey: string): Promise<void> =>
|
||||
ipcRenderer.invoke('redis:renameKey', id, oldKey, newKey),
|
||||
|
||||
@@ -15,6 +15,19 @@ interface StreamEntry {
|
||||
fields: Record<string, string>
|
||||
}
|
||||
|
||||
interface ConsumerGroup {
|
||||
name: string
|
||||
consumers: number
|
||||
pending: number
|
||||
lastDeliveredId: string
|
||||
}
|
||||
|
||||
interface Consumer {
|
||||
name: string
|
||||
pending: number
|
||||
idle: number
|
||||
}
|
||||
|
||||
const entries = ref<StreamEntry[]>([])
|
||||
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<ConsumerGroup[]>([])
|
||||
const expandedGroup = ref<string | null>(null)
|
||||
const consumers = ref<Record<string, Consumer[]>>({})
|
||||
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<string, string> {
|
||||
return result
|
||||
}
|
||||
|
||||
function parseGroupInfo(raw: any[]): ConsumerGroup {
|
||||
const map: Record<string, any> = {}
|
||||
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<string, any> = {}
|
||||
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() {
|
||||
<el-button size="small" :loading="trimming">{{ t('editor.trim') }}</el-button>
|
||||
</template>
|
||||
</el-popconfirm>
|
||||
<el-button size="small" @click="loadEntries" :loading="loading">{{ t('common.refresh') }}</el-button>
|
||||
<el-button size="small" @click="loadEntries(); loadGroups()" :loading="loading || groupsLoading">{{ t('common.refresh') }}</el-button>
|
||||
</div>
|
||||
|
||||
<div class="groups-panel">
|
||||
<div class="groups-header">
|
||||
<span class="groups-title">{{ t('editor.consumerGroups') }}</span>
|
||||
<span class="groups-count">{{ groups.length }}</span>
|
||||
<el-button size="small" type="primary" @click="showGroupDialog = true">{{ t('editor.createGroup') }}</el-button>
|
||||
</div>
|
||||
<div class="groups-list">
|
||||
<div v-for="group in groups" :key="group.name" class="group-item">
|
||||
<div class="group-row" @click="toggleGroup(group.name)">
|
||||
<span class="group-arrow" :class="{ expanded: expandedGroup === group.name }">▶</span>
|
||||
<span class="group-name">{{ group.name }}</span>
|
||||
<span class="group-stat">{{ t('editor.consumers') }}: {{ group.consumers }}</span>
|
||||
<span class="group-stat">{{ t('editor.pending') }}: {{ group.pending }}</span>
|
||||
<span class="group-last-id">{{ group.lastDeliveredId }}</span>
|
||||
<el-button size="small" type="danger" text @click.stop="destroyGroup(group.name)">{{ t('editor.destroyGroup') }}</el-button>
|
||||
</div>
|
||||
<div v-if="expandedGroup === group.name" class="consumers-list">
|
||||
<div v-for="consumer in consumers[group.name] || []" :key="consumer.name" class="consumer-row">
|
||||
<span class="consumer-name">{{ consumer.name }}</span>
|
||||
<span class="consumer-stat">{{ t('editor.pending') }}: {{ consumer.pending }}</span>
|
||||
<span class="consumer-stat">{{ t('editor.consumers') }}: {{ Math.round(consumer.idle / 1000) }}s</span>
|
||||
</div>
|
||||
<div v-if="!(consumers[group.name] && consumers[group.name].length)" class="no-consumers">{{ t('editor.empty') }}</div>
|
||||
</div>
|
||||
</div>
|
||||
<div v-if="groups.length === 0 && !groupsLoading" class="no-groups">{{ t('editor.noGroups') }}</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="stream-content">
|
||||
@@ -175,6 +326,24 @@ async function trimStream() {
|
||||
</template>
|
||||
</el-dialog>
|
||||
|
||||
<el-dialog v-model="showGroupDialog" :title="t('editor.createGroup')" width="400px">
|
||||
<el-form label-position="top">
|
||||
<el-form-item :label="t('editor.groupName')">
|
||||
<el-input v-model="newGroupName" />
|
||||
</el-form-item>
|
||||
<el-form-item :label="t('editor.startId')">
|
||||
<el-input v-model="newGroupStartId" placeholder="$" />
|
||||
</el-form-item>
|
||||
<el-form-item>
|
||||
<el-checkbox v-model="newGroupMkstream">{{ t('editor.mkstream') }}</el-checkbox>
|
||||
</el-form-item>
|
||||
</el-form>
|
||||
<template #footer>
|
||||
<el-button @click="showGroupDialog = false">{{ t('common.cancel') }}</el-button>
|
||||
<el-button type="primary" @click="createGroup">{{ t('common.confirm') }}</el-button>
|
||||
</template>
|
||||
</el-dialog>
|
||||
|
||||
<el-dialog v-model="trimming" :title="t('editor.trim')" width="300px">
|
||||
<el-form-item :label="t('editor.maxLength')">
|
||||
<el-input-number v-model="trimMaxLen" :min="1" />
|
||||
@@ -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;
|
||||
|
||||
+13
-1
@@ -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',
|
||||
|
||||
@@ -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',
|
||||
|
||||
Reference in New Issue
Block a user