MBS Enqueue (消息推送)

概览

字段	值
Action 名称	MBS Enqueue (消息推送)
Connector Type	MBS
Description	向 MBS 指定 Broker 的 Queue 推送消息

Input Parameters

Name	Label	Data Type	Control Type	必填	说明
`brokerName`	Broker Name	string	text	是	Broker 名称，如 `builtin`
`queueName`	Queue Name	string	text	是	队列名称
`body`	Message Body	string	text	是	消息体内容（JSON 或纯文本）

参数详解

brokerName

MBS 中注册的 Broker 名称。常见值：

builtin - MBS 内置的 PostgreSQL 队列后端
其他由管理员在 MBS 管理后台创建的 Broker 名称（如 kafka-prod、rabbitmq-dev）

queueName

目标队列名称。消息会被投递到该 Broker 下的指定队列中。

body

消息体内容，支持任意格式：

JSON 格式：{"event": "order_created", "orderId": "ORD-2024-001"}
纯文本：2024-01-15 10:30:00 ERROR Application crashed
XML 格式：<message><type>alert</type></message>

备注

MBS 对消息体格式不做限制，以 application/octet-stream 方式原样存储和投递。消费者需要自行解析消息格式。

Output Parameters

Name	Label	Data Type	说明
`messageId`	Message ID	string	MBS 返回的消息唯一标识

messageId 是由 MBS 生成的 UUID，可用于后续追踪消息状态。

Action Script

def pathStr = 'api/broker/' + inputs.brokerName + '/queue/' + inputs.queueName + '/enqueue'

// inputs.body 可能被框架解析为 Map，需要转回字符串
def bodyContent = inputs.body
if (!(bodyContent instanceof String)) {
    bodyContent = new groovy.json.JsonBuilder(bodyContent).toString()
}

def result = context.http {
    path pathStr
    method "POST"
    headers (["Content-Type": "application/octet-stream"])
    body (bodyContent.getBytes("UTF-8"))
}

if (result?.statusCode != null && result.statusCode >= 400) {
    throw new RuntimeException("MBS failed [" + result.statusCode + "]: " + result.body)
}
return result

脚本逻辑说明：

拼接 API 路径 - api/broker/{brokerName}/queue/{queueName}/enqueue
处理消息体 - 如果 inputs.body 被框架自动解析为了 Map/Object（当消息体是合法 JSON 时可能发生），则用 JsonBuilder 转回字符串
发送请求 - 使用 application/octet-stream Content-Type，将消息体以 UTF-8 编码的字节数组发送
错误处理 - 当 HTTP 状态码 >= 400 时抛出异常，异常信息包含状态码和响应体

API 对应关系

此 Action 对应 MBS 的如下 REST API：

POST /api/broker/{broker_name}/queue/{queue_name}/enqueue

请求头：

Header	值	说明
`Authorization`	`Bearer {token}`	由 Connector 自动附加
`Content-Type`	`application/octet-stream`	Action Script 中设置

响应体示例：

{
  "messageId": "550e8400-e29b-41d4-a716-446655440000"
}

使用示例

示例 1：发送 JSON 业务消息

参数	值
Broker Name	`builtin`
Queue Name	`order-events`
Message Body	`{"event": "order_created", "orderId": "ORD-2024-001", "amount": 299.9, "currency": "CNY"}`

示例 2：发送纯文本日志

参数	值
Broker Name	`kafka-prod`
Queue Name	`app-logs`
Message Body	`2024-01-15 10:30:00 ERROR Application crashed`

示例 3：在 Flow 中使用变量

Message Body 中可以使用 Flow 变量动态拼接消息内容：

{"event": "user_registered", "userId": "${userId}", "email": "${email}", "timestamp": "${now}"}

错误处理

HTTP 状态码	错误码	原因
401	`AUTHENTICATION_ERROR`	Token 无效或已过期
403	`AUTHORIZATION_ERROR`	Token 没有该 Queue 的 write 权限
404	`INSTANCE_NOT_FOUND`	Broker 或 Queue 不存在
502	`BAD_GATEWAY`	后端 MQ 不可用

提示

如果 Enqueue 失败，请先确认：

Connector 中的 Token 是否有效
Token 是否有目标 {brokerName}/{queueName} 的 write 权限
目标 Broker 和 Queue 是否已在 MBS 管理后台创建
MBS 服务是否正常运行（访问 /health 端点检查）

概览​

Input Parameters​

参数详解​

brokerName​

queueName​

body​

Output Parameters​

Action Script​

API 对应关系​

使用示例​

示例 1：发送 JSON 业务消息​

示例 2：发送纯文本日志​

示例 3：在 Flow 中使用变量​

错误处理​

概览