rocketmq_client_rust/consumer/store/
offset_serialize.rs

1// Copyright 2023 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 std::collections::HashMap;
16use std::sync::atomic::Ordering;
17
18use rocketmq_protocol::protocol::RemotingSerializable;
19use serde::Deserialize;
20use serde::Serialize;
21use tracing::warn;
22
23use crate::consumer::store::offset_serialize_wrapper::OffsetSerializeWrapper;
24
25#[derive(Serialize, Deserialize, Debug, Default)]
26#[serde(rename_all = "camelCase")]
27pub struct OffsetSerialize {
28    pub offset_table: HashMap<String, i64>,
29}
30
31impl From<OffsetSerializeWrapper> for OffsetSerialize {
32    fn from(wrapper: OffsetSerializeWrapper) -> Self {
33        let mut offset_table = HashMap::new();
34        for (k, v) in wrapper.offset_table {
35            match k.serialize_json() {
36                Ok(result) => {
37                    offset_table.insert(result, v.load(Ordering::Relaxed));
38                }
39                Err(err) => {
40                    warn!("skip offset entry because message queue serialization failed: {err}");
41                }
42            }
43        }
44        OffsetSerialize { offset_table }
45    }
46}
rocketmq_client_rust/consumer/store/offset_serialize.rs

rocketmq_client_rust/consumer/store/
offset_serialize.rs
