This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 5f92965f0 feat(topic): add send payload preflight (#3064)
5f92965f0 is described below
commit 5f92965f0091d57f269816979df4810fc59e2468
Author: coder999o <[email protected]>
AuthorDate: Fri Sep 4 14:52:30 2026 +0800
feat(topic): add send payload preflight (#3064)
---
.../pages/instance/__tests__/TopicPage.test.tsx | 88 ++++-
web/src/pages/instance/topic.tsx | 211 ++++++++++-
web/src/utils/messagePayloadPreview.test.ts | 256 +++++++++++++
web/src/utils/messagePayloadPreview.ts | 410 +++++++++++++++++++++
4 files changed, 943 insertions(+), 22 deletions(-)
diff --git a/web/src/pages/instance/__tests__/TopicPage.test.tsx
b/web/src/pages/instance/__tests__/TopicPage.test.tsx
index 8dc37744b..b230c8ef7 100644
--- a/web/src/pages/instance/__tests__/TopicPage.test.tsx
+++ b/web/src/pages/instance/__tests__/TopicPage.test.tsx
@@ -16,10 +16,10 @@
*/
import { describe, it, expect, vi, beforeAll, beforeEach, afterEach } from
'vitest';
-import { fireEvent, render, screen, waitFor, within } from
'@testing-library/react';
+import { cleanup, fireEvent, render, screen, waitFor, within } from
'@testing-library/react';
import userEvent from '@testing-library/user-event';
import { MemoryRouter, Route, Routes } from 'react-router-dom';
-import { App } from 'antd';
+import { App, Modal } from 'antd';
import { LangProvider } from '../../../i18n/LangContext';
import type { BrokerRoute, Topic } from '../../../api/metadata';
import { parseMessageProperties } from '../../../utils/messageProperties';
@@ -132,6 +132,13 @@ const getTableBody = () => {
return tableBody as HTMLElement;
};
+const getSendDialog = async () => {
+ const title = await screen.findByText('发送消息到 topic-01');
+ const dialog = title.closest('[role="dialog"]');
+ expect(dialog).not.toBeNull();
+ return dialog as HTMLElement;
+};
+
describe('TopicPage', () => {
beforeEach(() => {
mockTopicsList(buildTopics(25));
@@ -175,6 +182,11 @@ describe('TopicPage', () => {
page: 1,
pageSize: 20,
});
+ topicServiceMocks.sendTopicMessage.mockResolvedValue({
+ msgId: 'MSG-0001',
+ sendTime: '2026-01-02T00:00:00Z',
+ offsetMsgId: 'OFFSET-0001',
+ });
instanceServiceMocks.listInstances.mockResolvedValue([
{
id: 5,
@@ -191,6 +203,8 @@ describe('TopicPage', () => {
});
afterEach(() => {
+ Modal.destroyAll();
+ cleanup();
vi.clearAllMocks();
});
@@ -644,6 +658,76 @@ describe('TopicPage', () => {
});
});
+ it('previews the send payload and submits the normalized properties', async
() => {
+ const user = userEvent.setup();
+ mockTopicsList([buildTopics(1)[0]]);
+ renderWithProviders();
+
+ await user.click(await screen.findByRole('button', { name: /发送/ }));
+ const dialog = await getSendDialog();
+ fireEvent.change(within(dialog).getByLabelText('Tag'), { target: { value:
' paid ' } });
+ fireEvent.change(within(dialog).getByLabelText('Key'), { target: { value:
' order-1 ' } });
+ fireEvent.change(within(dialog).getByLabelText('消息体 Body'), {
+ target: { value: '{"orderId":"order-1","amount":128}' },
+ });
+ await user.click(within(dialog).getByRole('button', { name: /添加属性/ }));
+ fireEvent.change(within(dialog).getByPlaceholderText('属性名'), {
+ target: { value: 'traceId' },
+ });
+ fireEvent.change(within(dialog).getByPlaceholderText('属性值'), {
+ target: { value: 'trace-1' },
+ });
+
+ await waitFor(() =>
expect(within(dialog).getByText('可以发送')).toBeInTheDocument());
+ expect(within(dialog).getByText('JSON Object')).toBeInTheDocument();
+ expect(within(dialog).getByText('Tag paid')).toBeInTheDocument();
+ expect(within(dialog).getByText('Key order-1')).toBeInTheDocument();
+ expect(within(dialog).getByText('traceId=trace-1')).toBeInTheDocument();
+
+ await user.click(within(dialog).getByRole('button', { name: /发\s*送/ }));
+
+ await waitFor(() =>
expect(topicServiceMocks.sendTopicMessage).toHaveBeenCalledTimes(1));
+ expect(topicServiceMocks.sendTopicMessage).toHaveBeenCalledWith({
+ topic: 'topic-01',
+ instanceId: 'instance-proxy-1',
+ tag: 'paid',
+ key: 'order-1',
+ body: '{"orderId":"order-1","amount":128}',
+ properties: { traceId: 'trace-1' },
+ });
+ });
+
+ it('blocks duplicate form properties in the send payload preflight', async
() => {
+ const user = userEvent.setup();
+ mockTopicsList([buildTopics(1)[0]]);
+ renderWithProviders();
+
+ await user.click(await screen.findByRole('button', { name: /发送/ }));
+ const dialog = await getSendDialog();
+ fireEvent.change(within(dialog).getByLabelText('消息体 Body'), {
+ target: { value: '{"event":"created"}' },
+ });
+ await user.click(within(dialog).getByRole('button', { name: /添加属性/ }));
+ await user.click(within(dialog).getByRole('button', { name: /添加属性/ }));
+ const propertyKeys = within(dialog).getAllByPlaceholderText('属性名');
+ const propertyValues = within(dialog).getAllByPlaceholderText('属性值');
+ fireEvent.change(propertyKeys[0], { target: { value: 'traceId' } });
+ fireEvent.change(propertyValues[0], { target: { value: 'first' } });
+ fireEvent.change(propertyKeys[1], { target: { value: 'traceId' } });
+ fireEvent.change(propertyValues[1], { target: { value: 'second' } });
+
+ await waitFor(() =>
expect(within(dialog).getByText('阻止发送')).toBeInTheDocument());
+ expect(within(dialog).getByText('属性名重复')).toBeInTheDocument();
+
expect(within(dialog).getByText('重复属性会覆盖前面的值：traceId')).toBeInTheDocument();
+
+ await user.click(within(dialog).getByRole('button', { name: /发\s*送/ }));
+
+ await waitFor(() =>
+ expect(screen.getByText(/发送前预检未通过：属性名重复/)).toBeInTheDocument(),
+ );
+ expect(topicServiceMocks.sendTopicMessage).not.toHaveBeenCalled();
+ });
+
it('renders unavailable Topic consumer metrics distinctly from zero', async
() => {
const user = userEvent.setup();
mockTopicsList([buildTopics(1)[0]]);
diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx
index 12d696ee7..8ec7335f8 100644
--- a/web/src/pages/instance/topic.tsx
+++ b/web/src/pages/instance/topic.tsx
@@ -86,7 +86,6 @@ import {
type ResourceImportRow,
} from '../../utils/resourceCsvImport';
import { downloadCsv } from '../../utils/download';
-import { parseMessageProperties } from '../../utils/messageProperties';
import { tableScrollX } from '../../utils/table';
import {
analyzeTopicRoutes,
@@ -94,6 +93,13 @@ import {
type RouteDiagnosticStatus,
type RouteDistribution,
} from '../../utils/topicRouteDiagnostics';
+import {
+ analyzeMessagePayloadPreview,
+ type MessageBodyFormat,
+ type MessagePayloadIssue,
+ type MessagePayloadPreviewStatus,
+ type MessagePropertyInput,
+} from '../../utils/messagePayloadPreview';
const { Text } = Typography;
@@ -150,6 +156,15 @@ const TOPIC_TYPE_CARDS = [
// ─── Perm label ───────────────────────────────────────────────────
const PERM_LABEL: Record<string, string> = { RW: '读写', RO: '只读', WO: '只写' };
+type SendMessageFormValues = {
+ topic: string;
+ tag?: string;
+ key?: string;
+ body: string;
+ propsText?: string;
+ properties?: MessagePropertyInput[];
+};
+
const visibleTopics = (
topics: Topic[],
selectedInstanceId: string | undefined,
@@ -299,6 +314,35 @@ const ISSUE_SEVERITY_COLOR:
Record<RouteDiagnosticIssue['severity'], string> = {
const formatPercent = (value: number) => `${value.toFixed(value % 1 === 0 ? 0
: 1)}%`;
+const formatBytes = (bytes: number): string => {
+ if (bytes >= 1048576) return `${(bytes / 1048576).toFixed(2)} MB`;
+ if (bytes >= 1024) return `${(bytes / 1024).toFixed(2)} KB`;
+ return `${bytes} B`;
+};
+
+const BODY_FORMAT_LABEL: Record<MessageBodyFormat, string> = {
+ empty: '空 Body',
+ 'json-object': 'JSON Object',
+ 'json-array': 'JSON Array',
+ 'json-scalar': 'JSON 标量',
+ 'plain-text': '文本',
+};
+
+const PAYLOAD_STATUS_META: Record<
+ MessagePayloadPreviewStatus,
+ { label: string; color: string; alertType: 'success' | 'warning' | 'error' }
+> = {
+ ready: { label: '可以发送', color: 'success', alertType: 'success' },
+ warning: { label: '建议检查', color: 'warning', alertType: 'warning' },
+ error: { label: '阻止发送', color: 'error', alertType: 'error' },
+};
+
+const PAYLOAD_ISSUE_COLOR: Record<MessagePayloadIssue['severity'], string> = {
+ info: 'blue',
+ warning: 'warning',
+ error: 'error',
+};
+
// ═══════════════════════════════════════════════════════════════════
const TopicPage = () => {
const { t } = useLang();
@@ -344,6 +388,12 @@ const TopicPage = () => {
const [sending, setSending] = useState(false);
const [sendForm] = Form.useForm();
const [propsMode, setPropsMode] = useState<'form' | 'text'>('form');
+ const sendTagValue = Form.useWatch('tag', sendForm);
+ const sendKeyValue = Form.useWatch('key', sendForm);
+ const sendBodyValue = Form.useWatch('body', sendForm);
+ const sendPropsTextValue = Form.useWatch('propsText', sendForm);
+ const sendPropertiesValue = Form.useWatch('properties', sendForm) as
+ MessagePropertyInput[] | undefined;
const { modal } = App.useApp();
const importInputRef = useRef<HTMLInputElement>(null);
const [importModalOpen, setImportModalOpen] = useState(false);
@@ -359,6 +409,28 @@ const TopicPage = () => {
const consumersRequestIdRef = useRef(0);
const createInFlightRef = useRef(false);
+ const sendPayloadPreview = useMemo(
+ () =>
+ analyzeMessagePayloadPreview({
+ topic: sendTopic?.name,
+ tag: sendTagValue,
+ key: sendKeyValue,
+ body: sendBodyValue,
+ propsMode,
+ propsText: sendPropsTextValue,
+ properties: sendPropertiesValue,
+ }),
+ [
+ propsMode,
+ sendBodyValue,
+ sendKeyValue,
+ sendPropertiesValue,
+ sendPropsTextValue,
+ sendTagValue,
+ sendTopic?.name,
+ ],
+ );
+
useEffect(() => {
if (!selectedInstanceId) {
topicRequestIdRef.current += 1;
@@ -1184,9 +1256,102 @@ const TopicPage = () => {
},
];
+ const renderPayloadIssues = (issues: MessagePayloadIssue[]) => {
+ if (issues.length === 0) {
+ return <Text type="secondary">未发现阻止发送的问题</Text>;
+ }
+ return (
+ <Space direction="vertical" size={6} style={{ width: '100%' }}>
+ {issues.map((item, index) => (
+ <Flex
+ key={`${item.code}-${item.names?.join(',') ?? index}`}
+ align="flex-start"
+ gap={8}
+ wrap="nowrap"
+ >
+ <Tag color={PAYLOAD_ISSUE_COLOR[item.severity]} style={{
marginTop: 1 }}>
+ {item.severity === 'error' ? '阻止' : item.severity === 'warning'
? '关注' : '提示'}
+ </Tag>
+ <div style={{ minWidth: 0 }}>
+ <Text strong>{item.title}</Text>
+ <Text type="secondary" style={{ display: 'block' }}>
+ {item.description}
+ </Text>
+ </div>
+ </Flex>
+ ))}
+ </Space>
+ );
+ };
+
+ const renderSendPayloadPreview = () => {
+ const statusMeta = PAYLOAD_STATUS_META[sendPayloadPreview.status];
+ const propertyPreview = sendPayloadPreview.propertyEntries.slice(0, 6);
+ const hiddenPropertyCount = sendPayloadPreview.propertyEntries.length -
propertyPreview.length;
+
+ return (
+ <Space direction="vertical" size={12} style={{ width: '100%' }}>
+ <Alert
+ showIcon
+ type={statusMeta.alertType}
+ message={
+ <Flex gap={8} align="center" wrap>
+ <span>发送前预检</span>
+ <Tag color={statusMeta.color}>{statusMeta.label}</Tag>
+
<Tag>{BODY_FORMAT_LABEL[sendPayloadPreview.summary.bodyFormat]}</Tag>
+ </Flex>
+ }
+ description={
+ sendPayloadPreview.blockingIssues.length > 0
+ ? `发现 ${sendPayloadPreview.blockingIssues.length} 个阻止发送的问题。`
+ : '将按下方摘要发送到 RocketMQ，发送前可继续调整 Body、Tag、Key 和自定义属性。'
+ }
+ />
+
+ <Flex gap={8} wrap>
+ <Tag>Body {formatBytes(sendPayloadPreview.summary.bodyBytes)}</Tag>
+ <Tag>属性 {sendPayloadPreview.summary.propertyCount}</Tag>
+ <Tag>属性大小
{formatBytes(sendPayloadPreview.summary.propertyBytes)}</Tag>
+ <Tag color={sendPayloadPreview.normalized.tag ? 'blue' : undefined}>
+ Tag {sendPayloadPreview.normalized.tag || '-'}
+ </Tag>
+ <Tag color={sendPayloadPreview.normalized.key ? 'blue' : undefined}>
+ Key {sendPayloadPreview.normalized.key || '-'}
+ </Tag>
+ </Flex>
+
+ <Descriptions bordered size="small" column={1}>
+ <Descriptions.Item label="Topic">
+ <Text code>{sendPayloadPreview.normalized.topic || sendTopic?.name
|| '-'}</Text>
+ </Descriptions.Item>
+ <Descriptions.Item label="Body 类型">
+ {BODY_FORMAT_LABEL[sendPayloadPreview.summary.bodyFormat]} /{' '}
+ {formatBytes(sendPayloadPreview.summary.bodyBytes)}
+ </Descriptions.Item>
+ <Descriptions.Item label="自定义属性">
+ {propertyPreview.length === 0 ? (
+ <Text type="secondary">无</Text>
+ ) : (
+ <Space size={[4, 4]} wrap>
+ {propertyPreview.map((entry) => (
+ <Tag key={entry.key} color={entry.reserved ? 'warning' :
undefined}>
+ {entry.key}={entry.value || '""'}
+ </Tag>
+ ))}
+ {hiddenPropertyCount > 0 && <Tag>+{hiddenPropertyCount}</Tag>}
+ </Space>
+ )}
+ </Descriptions.Item>
+ </Descriptions>
+
+ {renderPayloadIssues(sendPayloadPreview.issues)}
+ </Space>
+ );
+ };
+
// ─── Send message modal submit ────────────────────────────────
const handleSend = async () => {
- let values;
+ let values: SendMessageFormValues;
try {
values = await sendForm.validateFields();
} catch {
@@ -1195,27 +1360,28 @@ const TopicPage = () => {
}
setSending(true);
try {
- // Build properties: batch-paste text mode or key-value form rows
- let props: Record<string, string> = {};
- if (propsMode === 'text') {
- const parsed = parseMessageProperties(values.propsText || '');
- if (parsed.errors.length > 0) {
- message.error(`消息属性格式错误：${parsed.errors.join('；')}`);
- return;
- }
- props = parsed.properties;
- } else if (values.properties && Array.isArray(values.properties)) {
- values.properties.forEach((p: { key?: string; value?: string }) => {
- if (p.key) props[p.key] = p.value || '';
- });
+ const payloadPreview = analyzeMessagePayloadPreview({
+ topic: values.topic,
+ tag: values.tag,
+ key: values.key,
+ body: values.body,
+ propsMode,
+ propsText: values.propsText,
+ properties: values.properties,
+ });
+ if (payloadPreview.blockingIssues.length > 0) {
+ message.error(
+ `发送前预检未通过：${payloadPreview.blockingIssues.map((item) =>
item.title).join('；')}`,
+ );
+ return;
}
const result = await sendTopicMessage({
- topic: values.topic,
+ topic: payloadPreview.normalized.topic,
instanceId: selectedInstanceId || undefined,
- tag: values.tag || undefined,
- key: values.key || undefined,
- body: values.body,
- properties: props,
+ tag: payloadPreview.normalized.tag,
+ key: payloadPreview.normalized.key,
+ body: payloadPreview.normalized.body,
+ properties: payloadPreview.properties,
});
// Keep the modal open for consecutive sends
message.success(`消息发送成功！MsgId: ${result.msgId}`);
@@ -1787,6 +1953,11 @@ const TopicPage = () => {
)}
</Form.List>
)}
+
+ <Divider style={{ margin: '20px 0 16px' }} orientation="left" plain>
+ 发送前预检
+ </Divider>
+ {renderSendPayloadPreview()}
</Form>
</Modal>
diff --git a/web/src/utils/messagePayloadPreview.test.ts
b/web/src/utils/messagePayloadPreview.test.ts
new file mode 100644
index 000000000..813f56d43
--- /dev/null
+++ b/web/src/utils/messagePayloadPreview.test.ts
@@ -0,0 +1,256 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { describe, expect, it } from 'vitest';
+import {
+ analyzeMessagePayloadPreview,
+ buildMessagePropertiesFromRows,
+ isReservedMessageProperty,
+} from './messagePayloadPreview';
+
+describe('messagePayloadPreview', () => {
+ it('summarizes a ready JSON object payload with normalized tag and key', ()
=> {
+ const preview = analyzeMessagePayloadPreview({
+ topic: ' orders-topic ',
+ tag: ' paid ',
+ key: ' order-1 ',
+ body: '{"orderId":"order-1","amount":128}',
+ propsMode: 'form',
+ properties: [
+ { key: 'traceId', value: 'trace-1' },
+ { key: 'tenant', value: 'retail' },
+ ],
+ });
+
+ expect(preview.status).toBe('ready');
+ expect(preview.blockingIssues).toEqual([]);
+ expect(preview.normalized).toEqual({
+ topic: 'orders-topic',
+ tag: 'paid',
+ key: 'order-1',
+ body: '{"orderId":"order-1","amount":128}',
+ });
+ expect(preview.properties).toEqual({ traceId: 'trace-1', tenant: 'retail'
});
+ expect(preview.summary.bodyFormat).toBe('json-object');
+ expect(preview.summary.propertyCount).toBe(2);
+ expect(preview.issues.map((issue) => issue.code)).toEqual(['TRIMMED_TAG',
'TRIMMED_KEY']);
+ });
+
+ it('reports empty bodies as blocking issues', () => {
+ const preview = analyzeMessagePayloadPreview({
+ topic: 'orders-topic',
+ body: ' ',
+ propsMode: 'form',
+ properties: [],
+ });
+
+ expect(preview.status).toBe('error');
+ expect(preview.summary.bodyFormat).toBe('empty');
+ expect(preview.blockingIssues).toHaveLength(1);
+ expect(preview.blockingIssues[0]).toMatchObject({
+ code: 'EMPTY_BODY',
+ field: 'body',
+ severity: 'error',
+ });
+ });
+
+ it('flags default RocketMQ body size overflow as a blocking issue', () => {
+ const preview = analyzeMessagePayloadPreview(
+ {
+ topic: 'orders-topic',
+ body: 'rocketmq',
+ propsMode: 'form',
+ properties: [],
+ },
+ { maxBodyBytes: 4 },
+ );
+
+ expect(preview.status).toBe('error');
+ expect(preview.blockingIssues.map((issue) =>
issue.code)).toEqual(['BODY_SIZE_LIMIT']);
+ expect(preview.summary.bodyBytes).toBe(8);
+ });
+
+ it('keeps plain text sendable while describing the body format', () => {
+ const preview = analyzeMessagePayloadPreview({
+ topic: 'orders-topic',
+ body: 'plain event body',
+ propsMode: 'form',
+ properties: [],
+ });
+
+ expect(preview.status).toBe('ready');
+ expect(preview.summary.bodyFormat).toBe('plain-text');
+ expect(preview.issues.map((issue) =>
issue.code)).toEqual(['PLAIN_TEXT_BODY']);
+ });
+
+ it('detects JSON arrays and scalar JSON bodies separately', () => {
+ const arrayPreview = analyzeMessagePayloadPreview({
+ topic: 'orders-topic',
+ body: '[{"id":1}]',
+ propsMode: 'form',
+ properties: [],
+ });
+ const scalarPreview = analyzeMessagePayloadPreview({
+ topic: 'orders-topic',
+ body: '"ready"',
+ propsMode: 'form',
+ properties: [],
+ });
+
+ expect(arrayPreview.summary.bodyFormat).toBe('json-array');
+ expect(arrayPreview.status).toBe('ready');
+ expect(scalarPreview.summary.bodyFormat).toBe('json-scalar');
+ expect(scalarPreview.issues.map((issue) =>
issue.code)).toEqual(['SCALAR_JSON_BODY']);
+ });
+
+ it('reports duplicate form properties before they can overwrite earlier
values', () => {
+ const result = buildMessagePropertiesFromRows([
+ { key: 'traceId', value: 'first' },
+ { key: 'tenant', value: 'demo' },
+ { key: 'traceId', value: 'second' },
+ ]);
+
+ expect(result.entries.map((entry) => [entry.key, entry.value])).toEqual([
+ ['traceId', 'first'],
+ ['tenant', 'demo'],
+ ]);
+ expect(result.issues).toHaveLength(1);
+ expect(result.issues[0]).toMatchObject({
+ code: 'DUPLICATE_PROPERTY_KEY',
+ severity: 'error',
+ names: ['traceId'],
+ });
+ });
+
+ it('reports form property values that do not have a key', () => {
+ const result = buildMessagePropertiesFromRows([{ key: ' ', value:
'orphan-value' }]);
+
+ expect(result.entries).toEqual([]);
+ expect(result.issues[0]).toMatchObject({
+ code: 'EMPTY_PROPERTY_KEY',
+ severity: 'error',
+ });
+ });
+
+ it('ignores sparse form property rows emitted by dynamic form lists', () => {
+ const result = buildMessagePropertiesFromRows([
+ undefined as unknown as { key?: string; value?: string },
+ { key: 'traceId', value: 'trace-1' },
+ ]);
+
+ expect(result.entries.map((entry) => [entry.key, entry.value])).toEqual([
+ ['traceId', 'trace-1'],
+ ]);
+ expect(result.issues).toEqual([]);
+ });
+
+ it('uses the batch text parser and preserves values containing equals
signs', () => {
+ const preview = analyzeMessagePayloadPreview({
+ topic: 'orders-topic',
+ body: '{}',
+ propsMode: 'text',
+ propsText: 'traceId=abc\nsignature=part-a=part-b',
+ });
+
+ expect(preview.status).toBe('ready');
+ expect(preview.properties).toEqual({
+ traceId: 'abc',
+ signature: 'part-a=part-b',
+ });
+ expect(preview.blockingIssues).toEqual([]);
+ });
+
+ it('maps malformed batch property lines to blocking preview issues', () => {
+ const preview = analyzeMessagePayloadPreview({
+ topic: 'orders-topic',
+ body: '{}',
+ propsMode: 'text',
+ propsText: 'traceId=abc\ntenant\ntraceId=duplicate',
+ });
+
+ expect(preview.status).toBe('error');
+ expect(preview.blockingIssues.map((issue) => issue.code)).toEqual([
+ 'INVALID_PROPERTY_FORMAT',
+ 'INVALID_PROPERTY_FORMAT',
+ ]);
+ expect(preview.properties).toEqual({ traceId: 'abc' });
+ });
+
+ it('warns when user properties use names reserved by RocketMQ message
metadata', () => {
+ const preview = analyzeMessagePayloadPreview({
+ topic: 'orders-topic',
+ body: '{}',
+ propsMode: 'form',
+ properties: [
+ { key: 'TAGS', value: 'tag-a' },
+ { key: 'businessKey', value: 'b-1' },
+ ],
+ });
+
+ expect(isReservedMessageProperty('tags')).toBe(true);
+ expect(isReservedMessageProperty('businessKey')).toBe(false);
+ expect(preview.status).toBe('warning');
+ expect(preview.issues.find((issue) => issue.code ===
'RESERVED_PROPERTY_KEY')).toMatchObject({
+ severity: 'warning',
+ names: ['TAGS'],
+ });
+ });
+
+ it('summarizes empty property values and large property sets without
blocking send', () => {
+ const properties = Array.from({ length: 4 }, (_, index) => ({
+ key: `k${index}`,
+ value: index === 0 ? '' : 'v',
+ }));
+ const preview = analyzeMessagePayloadPreview(
+ {
+ topic: 'orders-topic',
+ body: '{}',
+ propsMode: 'form',
+ properties,
+ },
+ { maxProperties: 2, maxPropertyBytes: 6 },
+ );
+
+ expect(preview.status).toBe('warning');
+ expect(preview.blockingIssues).toEqual([]);
+ expect(preview.summary.propertyCount).toBe(4);
+ expect(preview.issues.map((issue) => issue.code)).toEqual([
+ 'EMPTY_PROPERTY_VALUE',
+ 'PROPERTY_COUNT_LIMIT',
+ 'PROPERTY_SIZE_LIMIT',
+ ]);
+ });
+
+ it('preserves prototype-like property names in the normalized property
object', () => {
+ const preview = analyzeMessagePayloadPreview({
+ topic: 'orders-topic',
+ body: '{}',
+ propsMode: 'text',
+ propsText: '__proto__=trace-prototype\nconstructor=trace-constructor',
+ });
+
+ expect(Object.keys(preview.properties)).toEqual(['__proto__',
'constructor']);
+ expect(preview.properties['__proto__']).toBe('trace-prototype');
+ expect(preview.properties['constructor']).toBe('trace-constructor');
+ expect(JSON.parse(JSON.stringify(preview.properties))).toEqual(
+ Object.fromEntries([
+ ['__proto__', 'trace-prototype'],
+ ['constructor', 'trace-constructor'],
+ ]),
+ );
+ });
+});
diff --git a/web/src/utils/messagePayloadPreview.ts
b/web/src/utils/messagePayloadPreview.ts
new file mode 100644
index 000000000..69113abf5
--- /dev/null
+++ b/web/src/utils/messagePayloadPreview.ts
@@ -0,0 +1,410 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { parseMessageProperties } from './messageProperties';
+
+export type MessagePropertyMode = 'form' | 'text';
+export type MessagePayloadPreviewStatus = 'ready' | 'warning' | 'error';
+export type MessagePayloadIssueSeverity = 'info' | 'warning' | 'error';
+export type MessageBodyFormat =
+ 'empty' | 'json-object' | 'json-array' | 'json-scalar' | 'plain-text';
+
+export type MessagePayloadIssueCode =
+ | 'EMPTY_BODY'
+ | 'BODY_SIZE_LIMIT'
+ | 'INVALID_PROPERTY_FORMAT'
+ | 'EMPTY_PROPERTY_KEY'
+ | 'DUPLICATE_PROPERTY_KEY'
+ | 'RESERVED_PROPERTY_KEY'
+ | 'EMPTY_PROPERTY_VALUE'
+ | 'PROPERTY_COUNT_LIMIT'
+ | 'PROPERTY_SIZE_LIMIT'
+ | 'TRIMMED_TAG'
+ | 'TRIMMED_KEY'
+ | 'PLAIN_TEXT_BODY'
+ | 'SCALAR_JSON_BODY';
+
+export interface MessagePropertyInput {
+ key?: string;
+ value?: string;
+}
+
+export interface MessagePayloadIssue {
+ code: MessagePayloadIssueCode;
+ severity: MessagePayloadIssueSeverity;
+ title: string;
+ description: string;
+ field?: 'body' | 'tag' | 'key' | 'properties';
+ names?: string[];
+}
+
+export interface MessagePayloadPreviewInput {
+ topic?: string;
+ tag?: string;
+ key?: string;
+ body?: string;
+ propsMode: MessagePropertyMode;
+ propsText?: string;
+ properties?: MessagePropertyInput[];
+}
+
+export interface MessagePayloadPreviewOptions {
+ maxBodyBytes?: number;
+ maxProperties?: number;
+ maxPropertyBytes?: number;
+}
+
+export interface MessagePropertyPreviewEntry {
+ key: string;
+ value: string;
+ keyBytes: number;
+ valueBytes: number;
+ reserved: boolean;
+}
+
+export interface MessagePayloadPreview {
+ status: MessagePayloadPreviewStatus;
+ issues: MessagePayloadIssue[];
+ blockingIssues: MessagePayloadIssue[];
+ normalized: {
+ topic: string;
+ tag?: string;
+ key?: string;
+ body: string;
+ };
+ properties: Record<string, string>;
+ propertyEntries: MessagePropertyPreviewEntry[];
+ summary: {
+ bodyBytes: number;
+ tagBytes: number;
+ keyBytes: number;
+ propertyCount: number;
+ propertyBytes: number;
+ bodyFormat: MessageBodyFormat;
+ maxBodyBytes: number;
+ maxProperties: number;
+ maxPropertyBytes: number;
+ };
+}
+
+export const DEFAULT_MAX_MESSAGE_BODY_BYTES = 4 * 1024 * 1024;
+export const DEFAULT_MAX_MESSAGE_PROPERTIES = 32;
+export const DEFAULT_MAX_MESSAGE_PROPERTY_BYTES = 16 * 1024;
+
+const RESERVED_PROPERTY_NAMES = new Set([
+ 'TAGS',
+ 'KEYS',
+ 'UNIQ_KEY',
+ 'WAIT',
+ 'DELAY',
+ 'RETRY_TOPIC',
+ 'REAL_TOPIC',
+ 'REAL_QID',
+ 'TRAN_MSG',
+ 'PGROUP',
+]);
+
+const textBytes = (value?: string): number => new TextEncoder().encode(value
?? '').length;
+
+const normalizeText = (value?: string): string => value?.trim() ?? '';
+
+const issue = (
+ code: MessagePayloadIssueCode,
+ severity: MessagePayloadIssueSeverity,
+ title: string,
+ description: string,
+ field?: MessagePayloadIssue['field'],
+ names?: string[],
+): MessagePayloadIssue => ({
+ code,
+ severity,
+ title,
+ description,
+ field,
+ names,
+});
+
+const bodyFormat = (body: string): MessageBodyFormat => {
+ const trimmed = body.trim();
+ if (!trimmed) return 'empty';
+
+ try {
+ const parsed = JSON.parse(trimmed) as unknown;
+ if (Array.isArray(parsed)) return 'json-array';
+ if (parsed !== null && typeof parsed === 'object') return 'json-object';
+ return 'json-scalar';
+ } catch {
+ return 'plain-text';
+ }
+};
+
+const fromEntries = (entries: MessagePropertyPreviewEntry[]): Record<string,
string> =>
+ Object.fromEntries(entries.map((entry) => [entry.key, entry.value]));
+
+export const isReservedMessageProperty = (key: string): boolean =>
+ RESERVED_PROPERTY_NAMES.has(key.trim().toUpperCase());
+
+export const buildMessagePropertiesFromRows = (
+ rows: MessagePropertyInput[] = [],
+): { entries: MessagePropertyPreviewEntry[]; issues: MessagePayloadIssue[] }
=> {
+ const entries: MessagePropertyPreviewEntry[] = [];
+ const issues: MessagePayloadIssue[] = [];
+ const seen = new Map<string, string>();
+ const duplicates = new Set<string>();
+
+ rows.forEach((row) => {
+ if (!row) return;
+ const rawKey = row.key ?? '';
+ const rawValue = row.value ?? '';
+ const key = rawKey.trim();
+ const value = rawValue.trim();
+
+ if (!key && !value) return;
+ if (!key) {
+ issues.push(
+ issue(
+ 'EMPTY_PROPERTY_KEY',
+ 'error',
+ '属性名不能为空',
+ `属性值“${value}”缺少对应属性名。`,
+ 'properties',
+ ),
+ );
+ return;
+ }
+
+ if (seen.has(key)) {
+ duplicates.add(key);
+ return;
+ }
+
+ seen.set(key, value);
+ entries.push({
+ key,
+ value,
+ keyBytes: textBytes(key),
+ valueBytes: textBytes(value),
+ reserved: isReservedMessageProperty(key),
+ });
+ });
+
+ if (duplicates.size > 0) {
+ issues.push(
+ issue(
+ 'DUPLICATE_PROPERTY_KEY',
+ 'error',
+ '属性名重复',
+ `重复属性会覆盖前面的值：${[...duplicates].sort().join(', ')}`,
+ 'properties',
+ [...duplicates].sort(),
+ ),
+ );
+ }
+
+ return { entries, issues };
+};
+
+const buildMessagePropertiesFromText = (
+ text: string,
+): { entries: MessagePropertyPreviewEntry[]; issues: MessagePayloadIssue[] }
=> {
+ const parsed = parseMessageProperties(text);
+ const entries = Object.entries(parsed.properties).map(([key, value]) => ({
+ key,
+ value,
+ keyBytes: textBytes(key),
+ valueBytes: textBytes(value),
+ reserved: isReservedMessageProperty(key),
+ }));
+
+ return {
+ entries,
+ issues: parsed.errors.map((errorText) =>
+ issue('INVALID_PROPERTY_FORMAT', 'error', '属性格式错误', errorText,
'properties'),
+ ),
+ };
+};
+
+export const analyzeMessagePayloadPreview = (
+ input: MessagePayloadPreviewInput,
+ options: MessagePayloadPreviewOptions = {},
+): MessagePayloadPreview => {
+ const maxBodyBytes = options.maxBodyBytes ?? DEFAULT_MAX_MESSAGE_BODY_BYTES;
+ const maxProperties = options.maxProperties ??
DEFAULT_MAX_MESSAGE_PROPERTIES;
+ const maxPropertyBytes = options.maxPropertyBytes ??
DEFAULT_MAX_MESSAGE_PROPERTY_BYTES;
+ const topic = normalizeText(input.topic);
+ const tag = normalizeText(input.tag);
+ const key = normalizeText(input.key);
+ const body = input.body ?? '';
+ const normalizedBody = body;
+ const issues: MessagePayloadIssue[] = [];
+
+ if (input.tag && input.tag !== tag) {
+ issues.push(
+ issue(
+ 'TRIMMED_TAG',
+ 'info',
+ 'Tag 会去除首尾空白',
+ '发送时会使用去除首尾空白后的 Tag。',
+ 'tag',
+ ),
+ );
+ }
+ if (input.key && input.key !== key) {
+ issues.push(
+ issue(
+ 'TRIMMED_KEY',
+ 'info',
+ 'Key 会去除首尾空白',
+ '发送时会使用去除首尾空白后的 Key。',
+ 'key',
+ ),
+ );
+ }
+
+ const format = bodyFormat(normalizedBody);
+ const bodyBytes = textBytes(normalizedBody);
+ if (format === 'empty') {
+ issues.push(issue('EMPTY_BODY', 'error', '消息体为空', '发送消息必须提供 Body。',
'body'));
+ } else if (bodyBytes > maxBodyBytes) {
+ issues.push(
+ issue(
+ 'BODY_SIZE_LIMIT',
+ 'error',
+ '消息体超过默认上限',
+ `当前 Body 为 ${bodyBytes} bytes，超过 ${maxBodyBytes} bytes。`,
+ 'body',
+ ),
+ );
+ } else if (format === 'plain-text') {
+ issues.push(
+ issue(
+ 'PLAIN_TEXT_BODY',
+ 'info',
+ 'Body 不是 JSON',
+ 'RocketMQ 支持文本消息，当前 Body 会按原始文本发送。',
+ 'body',
+ ),
+ );
+ } else if (format === 'json-scalar') {
+ issues.push(
+ issue(
+ 'SCALAR_JSON_BODY',
+ 'info',
+ 'Body 是 JSON 标量',
+ '当前 Body 是合法 JSON，但不是对象或数组。',
+ 'body',
+ ),
+ );
+ }
+
+ const propertyResult =
+ input.propsMode === 'text'
+ ? buildMessagePropertiesFromText(input.propsText ?? '')
+ : buildMessagePropertiesFromRows(input.properties);
+ issues.push(...propertyResult.issues);
+
+ const reservedNames = propertyResult.entries
+ .filter((entry) => entry.reserved)
+ .map((entry) => entry.key)
+ .sort();
+ if (reservedNames.length > 0) {
+ issues.push(
+ issue(
+ 'RESERVED_PROPERTY_KEY',
+ 'warning',
+ '属性名可能与系统属性冲突',
+ `建议改用业务属性名，避免覆盖或混淆系统属性：${reservedNames.join(', ')}`,
+ 'properties',
+ reservedNames,
+ ),
+ );
+ }
+
+ const emptyValueNames = propertyResult.entries
+ .filter((entry) => entry.value.length === 0)
+ .map((entry) => entry.key)
+ .sort();
+ if (emptyValueNames.length > 0) {
+ issues.push(
+ issue(
+ 'EMPTY_PROPERTY_VALUE',
+ 'info',
+ '存在空属性值',
+ `这些属性会以空字符串发送：${emptyValueNames.join(', ')}`,
+ 'properties',
+ emptyValueNames,
+ ),
+ );
+ }
+
+ const propertyBytes = propertyResult.entries.reduce(
+ (sum, entry) => sum + entry.keyBytes + entry.valueBytes,
+ 0,
+ );
+ if (propertyResult.entries.length > maxProperties) {
+ issues.push(
+ issue(
+ 'PROPERTY_COUNT_LIMIT',
+ 'warning',
+ '属性数量较多',
+ `当前 ${propertyResult.entries.length} 个属性，建议控制在 ${maxProperties} 个以内。`,
+ 'properties',
+ ),
+ );
+ }
+ if (propertyBytes > maxPropertyBytes) {
+ issues.push(
+ issue(
+ 'PROPERTY_SIZE_LIMIT',
+ 'warning',
+ '属性总大小较大',
+ `当前属性约 ${propertyBytes} bytes，建议控制在 ${maxPropertyBytes} bytes 以内。`,
+ 'properties',
+ ),
+ );
+ }
+
+ const blockingIssues = issues.filter((item) => item.severity === 'error');
+ const warningIssues = issues.filter((item) => item.severity === 'warning');
+ const status: MessagePayloadPreviewStatus =
+ blockingIssues.length > 0 ? 'error' : warningIssues.length > 0 ? 'warning'
: 'ready';
+
+ return {
+ status,
+ issues,
+ blockingIssues,
+ normalized: {
+ topic,
+ ...(tag ? { tag } : {}),
+ ...(key ? { key } : {}),
+ body: normalizedBody,
+ },
+ properties: fromEntries(propertyResult.entries),
+ propertyEntries: propertyResult.entries,
+ summary: {
+ bodyBytes,
+ tagBytes: textBytes(tag),
+ keyBytes: textBytes(key),
+ propertyCount: propertyResult.entries.length,
+ propertyBytes,
+ bodyFormat: format,
+ maxBodyBytes,
+ maxProperties,
+ maxPropertyBytes,
+ },
+ };
+};
