hrelayer 0.1.1

hrelayer是http协议转发器
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

use super::mailbox::{ErrorCode, WriteMailRequest, WriteMailResponse};
use crate::util::get_pool_and_meta;
use anyhow::anyhow;
use detcd::ServiceState;
use http_pool::net_pool::{Pool, Pools};

pub(super) async fn write_to_mailbox<P: Pool>(
    builder: &kmailbox::Builder,
    ns: Option<&String>,
    pools: Pools<P>,
    mut request: WriteMailRequest,
) -> (WriteMailResponse, Option<anyhow::Error>) {
    let mut response = WriteMailResponse::new();
    let mut err = None;

    loop {
        // 校验mb_req是否合法
        if request.payload.is_empty() {
            response.set_code(ErrorCode::EmptyPayload);
            break;
        }
        if request.service.is_empty() {
            response.set_code(ErrorCode::NoTarget);
            break;
        }

        // 获取服务类型
        let (pool, meta) = match get_pool_and_meta(&pools, &request.service) {
            Ok((p, m)) => (p, m),
            Err(e) => {
                err = Some(e);
                response.set_code(ErrorCode::NoTarget);
                break;
            }
        };

        // 获取分区
        let partition = match meta.state {
            Some(ServiceState::Fixed) => {
                // 校验service id是否合法
                if request.service_id.is_none() || request.service_id.unwrap() >= meta.instances {
                    response.set_code(ErrorCode::InvalidServiceId);
                    break;
                }
                request.service_id
            }
            Some(ServiceState::Stateful) => {
                // 校验key是否合法
                let key = request.key.as_ref().map_or("", |k| k);
                if key.is_empty() {
                    response.set_code(ErrorCode::EmptyKey);
                    break;
                }
                // 为了保证邮件的路由规则与消息包的路由规则一致, 这里要根据mb_req.key判断邮件去到哪个目标
                match pool.get_backend(key) {
                    Some(bs) => bs.id(),
                    None => {
                        response.set_code(ErrorCode::TargetNotReady);
                        break;
                    }
                }
            }
            _ => None,
        };

        // 写邮件
        request.service_id = partition;
        match write_mail(builder, ns, &request).await {
            Err(e) => {
                err = Some(anyhow!("{:?}", e));
                response.set_code(ErrorCode::ErrSystem);
                break;
            }
            _ => {}
        }

        break;
    }

    (response, err)
}

/// 发送到邮箱
async fn write_mail(
    builder: &kmailbox::Builder,
    ns: Option<&String>,
    request: &WriteMailRequest,
) -> Result<(), kmailbox::KafkaError> {
    let writer = builder.writer()?;

    // 构建邮箱结构
    let mut sm = kmailbox::SendMail::new()
        .key(request.key.as_ref())
        .payload(Some(&request.payload))
        .partition(request.service_id.map(|id| id as i32));

    // 填充头部
    for kv in request.headers.iter() {
        sm = sm.add_header(kv.0, Some(kv.1));
    }

    let name = ns.map_or(request.service.clone(), |ns| {
        if ns.is_empty() {
            request.service.clone()
        } else {
            ns.clone() + "." + &request.service
        }
    });

    writer.write(&name, sm).await
}