feat(rocketmq): replace Java agent with native Go implementation

This commit is contained in:
t8y2 committed 2026-08-12 23:15:20 +08:00
1 parent baec487a21
commit 9fb70bcfc9
44 files changed
+4872 -5414

No files matched your search

+2
View File
@@ -14,7 +14,9 @@ mq/
├── service.rs - 服务层函数
└── adapters/
├── pulsar.rs - Pulsar 实现
├── kafka.rs - Kafka 实现 (Java agent)
├── rabbitmq.rs - RabbitMQ 实现 (Go native agent)
├── rocketmq.rs - RocketMQ 实现 (Go native agent)
└── pulsar_version.rs - 版本探测
```
+8 -9
View File
@@ -1,10 +1,9 @@
//! Apache RocketMQ admin adapter. Communicates with a Java agent process
//! (`RocketMqAgent.java`) via JSON-RPC over stdin/stdout. The Java agent uses
//! `DefaultMQAdminExt` for admin operations and `DefaultMQProducer` for
//! message production.
//! Apache RocketMQ admin adapter. Communicates with the Go native agent via
//! JSON-RPC over stdin/stdout. The agent uses `rocketmq-admin-go` plus direct
//! remoting calls for compatibility gaps in the upstream admin API.
//!
//! This adapter follows the same pattern as the ZooKeeper/Etcd agents:
//! 1. Spawn a Java agent process via `AgentDriverClient`
//! This adapter follows the same process-agent pattern as the other drivers:
//! 1. Spawn the native agent process via `AgentDriverClient`
//! 2. Perform JSON-RPC handshake + connect
//! 3. Delegate all `MessageQueueAdmin` trait methods to JSON-RPC calls
@@ -97,10 +96,10 @@ fn cluster_info_from_agent_result(result: &serde_json::Value) -> MqClusterInfo {
}
impl RocketMqAdmin {
/// Spawn the RocketMQ Java agent, perform handshake, and connect.
/// Spawn the RocketMQ native agent, perform handshake, and connect.
///
/// Callers must probe NameServer reachability before invoking this so the
/// connect-timeout wall covers only JVM spawn + handshake + connect.
/// connect-timeout wall covers only process spawn + handshake + connect.
pub async fn new(cfg: MqAdminConfig, launch: AgentLaunchSpec) -> Result<Self, String> {
let mut client = AgentDriverClient::spawn(launch).await?;
@@ -814,7 +813,7 @@ fn rocketmq_topic_admin_params(topic: &TopicRef, partitions: Option<u32>) -> ser
params
}
/// Build the connection params JSON from MqAdminConfig for the Java agent.
/// Build the connection params JSON from MqAdminConfig for the native agent.
fn build_connection_params(cfg: &MqAdminConfig) -> serde_json::Value {
let extra = &cfg.extra;
let access_key = extra_str(extra, "accessKey")