rocketmq_protocol/protocol/body/
proxy_drain.rs

1// Copyright 2026 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use serde::Deserialize;
16use serde::Serialize;
17
18pub const PROXY_DRAIN_SCHEMA_VERSION: &str = "rocketmq.proxy-drain.v1";
19
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21#[serde(rename_all = "camelCase", deny_unknown_fields)]
22pub struct ProxyDrainOperationRequestBody {
23    pub schema_version: String,
24    pub operation_id: String,
25}
26
27#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
28#[serde(rename_all = "camelCase")]
29pub struct ProxyDrainPendingBody {
30    pub active_connections: usize,
31    pub sessions: usize,
32    pub receipt_handles: usize,
33    pub prepared_transactions: usize,
34    pub telemetry_links: usize,
35    pub remoting_channels: usize,
36    pub telemetry_commands: usize,
37    pub rpc_in_flight: usize,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41#[serde(rename_all = "camelCase")]
42pub struct ProxyDrainStateResponseBody {
43    pub schema_version: String,
44    pub phase: String,
45    pub operation_id: Option<String>,
46    pub admission_open: bool,
47    pub routing_open: bool,
48    pub readiness_published: bool,
49    pub zero_pending: bool,
50    pub pending: ProxyDrainPendingBody,
51}
52
53#[cfg(test)]
54mod tests {
55    use crate::protocol::RemotingDeserializable;
56    use crate::protocol::RemotingSerializable;
57
58    use super::*;
59
60    #[test]
61    fn operation_request_rejects_unknown_fields() {
62        let invalid = br#"{
63            "schemaVersion":"rocketmq.proxy-drain.v1",
64            "operationId":"restart-1",
65            "unexpected":true
66        }"#;
67        assert!(ProxyDrainOperationRequestBody::decode(invalid).is_err());
68    }
69
70    #[test]
71    fn state_response_round_trips_with_all_pending_dimensions() {
72        let expected = ProxyDrainStateResponseBody {
73            schema_version: PROXY_DRAIN_SCHEMA_VERSION.to_owned(),
74            phase: "draining".to_owned(),
75            operation_id: Some("restart-1".to_owned()),
76            admission_open: false,
77            routing_open: false,
78            readiness_published: false,
79            zero_pending: false,
80            pending: ProxyDrainPendingBody {
81                active_connections: 2,
82                sessions: 1,
83                receipt_handles: 3,
84                prepared_transactions: 4,
85                telemetry_links: 1,
86                remoting_channels: 1,
87                telemetry_commands: 5,
88                rpc_in_flight: 6,
89            },
90        };
91        let encoded = expected.encode().unwrap();
92        let decoded = ProxyDrainStateResponseBody::decode(encoded.as_slice()).unwrap();
93        assert_eq!(decoded, expected);
94    }
95}
rocketmq_protocol/protocol/body/proxy_drain.rs

rocketmq_protocol/protocol/body/
proxy_drain.rs
