sz-orm-queue 1.2.2

//! RocketMQ 真实队列 provider 实现（基于 reqwest 调用 RocketMQ 5.x Proxy HTTP API）
//!
//! 由于无成熟 Rust 原生客户端（`rocketmq-client-rust` 采用率极低、需独立 broker 才能运行与测试），
//! 按任务约定回退到 HTTP 代理模式，通过 HTTP 调用 RocketMQ Proxy 的消息接口：
//! - 发送消息：`POST   {base}/message`            body: `{ topic, group, payload }`
//! - 消费消息：`GET    {base}/message?topic=&group=` → `{ id, topic, payload }`
//! - 确认消息：`DELETE {base}/message/{id}?group=`
//!
//! 依赖：RocketMQ 5.x Proxy 需开启 HTTP 接入（或部署兼容上述接口的 Dashboard/网关）。
//! `payload` 以 JSON 数组（u8 序列）传输，保证任意二进制可往返。

use crate::error::MqError;
use crate::queue::{Message, MessageQueue};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// 发送消息的请求体
#[derive(Serialize)]
struct SendRequest<'a> {
    topic: &'a str,
    group: &'a str,
    /// 以 JSON 数组（u8 序列）传输，保证任意二进制可往返
    payload: &'a [u8],
}

/// 消费消息的响应体
#[derive(Deserialize)]
struct ConsumeResponse {
    /// 服务端返回的消息 ID（用于 ack）
    id: String,
    topic: String,
    /// 以 JSON 数组（u8 序列）传输，反序列化为字节
    payload: Vec<u8>,
}

/// RocketMQ HTTP 队列 provider
pub struct RocketMqProvider {
    /// HTTP 基地址（如 `http://127.0.0.1:8081`），末尾 `/` 会被去除
    base_url: String,
    /// 默认 topic（信息字段，实际 topic 由 publish/consume 参数指定）
    topic: String,
    /// 消费者组
    consumer_group: String,
    client: reqwest::Client,
}

impl RocketMqProvider {
    /// 便捷构造：连接 RocketMQ Proxy HTTP 端点（使用默认 topic/group）
    ///
    /// 等价于 `connect(name_server, "default-topic", "default-group")`。
    /// 如需自定义 topic/group，请使用 [`connect`](Self::connect)。
    pub async fn new(name_server: impl Into<String>) -> Result<Self, MqError> {
        Self::connect(name_server, "default-topic", "default-group").await
    }

    /// 创建并连接 RocketMQ Proxy HTTP 端点
    ///
    /// - `base_url`：HTTP 基地址（如 `http://127.0.0.1:8081`）
    /// - `topic`：默认 topic
    /// - `consumer_group`：消费者组
    pub async fn connect(
        base_url: impl Into<String>,
        topic: impl Into<String>,
        consumer_group: impl Into<String>,
    ) -> Result<Self, MqError> {
        let client = reqwest::Client::builder()
            .build()
            .map_err(|e| MqError::Connection(format!("reqwest client build failed: {e}")))?;
        Ok(Self {
            base_url: base_url.into().trim_end_matches('/').to_string(),
            topic: topic.into(),
            consumer_group: consumer_group.into(),
            client,
        })
    }

    /// 返回 HTTP 基地址（已去除末尾 `/`）
    pub fn base_url(&self) -> &str {
        &self.base_url
    }

    /// 返回默认 topic
    pub fn topic(&self) -> &str {
        &self.topic
    }

    /// 返回消费者组
    pub fn consumer_group(&self) -> &str {
        &self.consumer_group
    }
}
#[async_trait]
impl MessageQueue for RocketMqProvider {
    /// 发送消息：`POST /message`，body 为 `{ topic, group, payload }`
    async fn publish(&self, topic: &str, message: &[u8]) -> Result<(), MqError> {
        let req = SendRequest {
            topic,
            group: &self.consumer_group,
            payload: message,
        };
        let url = format!("{}/message", self.base_url);
        let resp = self
            .client
            .post(&url)
            .json(&req)
            .send()
            .await
            .map_err(|e| MqError::Publish(format!("RocketMQ HTTP send failed: {e}")))?;
        if !resp.status().is_success() {
            return Err(MqError::Publish(format!(
                "RocketMQ HTTP send status {}",
                resp.status()
            )));
        }
        Ok(())
    }

    /// 消费消息：`GET /message?topic=&group=`，返回 `{ id, topic, payload }`
    ///
    /// 服务端无消息时返回 204 No Content → `None`。
    async fn consume(&self, topic: &str) -> Result<Option<Message>, MqError> {
        let url = format!("{}/message", self.base_url);
        let resp = self
            .client
            .get(&url)
            .query(&[("topic", topic), ("group", self.consumer_group.as_str())])
            .send()
            .await
            .map_err(|e| MqError::Connection(format!("RocketMQ HTTP consume failed: {e}")))?;
        if resp.status() == reqwest::StatusCode::NO_CONTENT {
            return Ok(None);
        }
        if !resp.status().is_success() {
            return Err(MqError::Connection(format!(
                "RocketMQ HTTP consume status {}",
                resp.status()
            )));
        }
        let body: ConsumeResponse = resp
            .json()
            .await
            .map_err(|e| MqError::Connection(format!("RocketMQ HTTP decode failed: {e}")))?;
        Ok(Some(Message {
            topic: body.topic,
            payload: body.payload,
            key: None,
            timestamp: current_timestamp_millis(),
            headers: HashMap::new(),
            id: body.id,
            retry_count: 0,
        }))
    }

    /// 确认消息：`DELETE /message/{id}?group=...`
    async fn ack(&self, message_id: &str) -> Result<(), MqError> {
        let url = format!(
            "{}/message/{}?group={}",
            self.base_url, message_id, self.consumer_group
        );
        let resp = self
            .client
            .delete(&url)
            .send()
            .await
            .map_err(|e| MqError::Publish(format!("RocketMQ HTTP ack failed: {e}")))?;
        if !resp.status().is_success() {
            return Err(MqError::Publish(format!(
                "RocketMQ HTTP ack status {}",
                resp.status()
            )));
        }
        Ok(())
    }

    /// 订阅：RocketMQ 通过消费者组自动消费，无需显式订阅；此处为 no-op
    async fn subscribe(&self, _topic: &str) -> Result<(), MqError> {
        Ok(())
    }
}

/// 当前时间戳（毫秒）
fn current_timestamp_millis() -> i64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as i64
}
#[cfg(test)]
mod tests {
    use super::*;

    /// 编译验证：发送请求体序列化（payload 以 JSON 数组传输）
    #[test]
    fn test_rocketmq_send_request_serialize() {
        let req = SendRequest {
            topic: "t",
            group: "g",
            payload: &[1, 2, 3],
        };
        let json = serde_json::to_value(&req).unwrap();
        assert_eq!(json["topic"], "t");
        assert_eq!(json["group"], "g");
        assert_eq!(json["payload"], serde_json::json!([1, 2, 3]));
    }

    /// 编译验证：消费响应体反序列化（JSON 数组 → Vec<u8>）
    #[test]
    fn test_rocketmq_consume_response_deserialize() {
        let json = serde_json::json!({ "id": "abc", "topic": "t", "payload": [104, 105] });
        let resp: ConsumeResponse = serde_json::from_value(json).unwrap();
        assert_eq!(resp.id, "abc");
        assert_eq!(resp.topic, "t");
        assert_eq!(resp.payload, b"hi");
    }

    /// 编译验证：connect 构造客户端、base_url 去尾斜杠、subscribe 为 no-op
    #[tokio::test]
    async fn test_rocketmq_connect_builds_client() {
        let q = RocketMqProvider::connect("http://127.0.0.1:8081/", "t", "g")
            .await
            .unwrap();
        assert_eq!(q.base_url, "http://127.0.0.1:8081");
        assert_eq!(q.topic(), "t");
        assert_eq!(q.consumer_group(), "g");
        q.subscribe("t").await.unwrap();
    }

    /// 编译验证：new 便捷构造使用默认 topic/group
    #[tokio::test]
    async fn test_rocketmq_new_uses_defaults() {
        let q = RocketMqProvider::new("http://127.0.0.1:8081").await.unwrap();
        assert_eq!(q.base_url, "http://127.0.0.1:8081");
        assert_eq!(q.topic(), "default-topic");
        assert_eq!(q.consumer_group(), "default-group");
    }

    // ----------------------------------------------------------------------
    // 真实 RocketMQ 集成测试（需 RocketMQ 5.x Proxy HTTP 端点，默认 #[ignore]）
    // ----------------------------------------------------------------------

    #[tokio::test]
    #[ignore = "需真实 RocketMQ 5.x Proxy HTTP 端点"]
    async fn test_rocketmq_publish_consume_ack() {
        let q = RocketMqProvider::connect("http://127.0.0.1:8081", "test-topic", "test-group")
            .await
            .unwrap();
        q.publish("test-topic", b"hello-rocket").await.unwrap();
        let msg = q
            .consume("test-topic")
            .await
            .unwrap()
            .expect("应有消息");
        assert_eq!(msg.payload, b"hello-rocket");
        q.ack(&msg.id).await.unwrap();
    }
}