rocketmq_model/common/message/

storage_metadata.rs

// Copyright 2023 The RocketMQ Rust Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::net::SocketAddr;

use bytes::BufMut;
use bytes::Bytes;
use bytes::BytesMut;
use cheetah_string::CheetahString;

/// Message storage metadata
///
/// Contains indexing and location information for messages in the Broker storage engine.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StorageMetadata {
    broker_name: CheetahString,
    queue_id: i32,
    queue_offset: i64,
    commit_log_offset: i64,
    store_timestamp: i64,
    store_host: SocketAddr,
    store_size: i32,
}

impl StorageMetadata {
    /// Creates new storage metadata
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        broker_name: CheetahString,
        queue_id: i32,
        queue_offset: i64,
        commit_log_offset: i64,
        store_timestamp: i64,
        store_host: SocketAddr,
        store_size: i32,
    ) -> Self {
        Self {
            broker_name,
            queue_id,
            queue_offset,
            commit_log_offset,
            store_timestamp,
            store_host,
            store_size,
        }
    }

    /// Gets the broker name
    #[inline]
    pub fn broker_name(&self) -> &str {
        self.broker_name.as_str()
    }

    /// Gets the queue ID
    #[inline]
    pub fn queue_id(&self) -> i32 {
        self.queue_id
    }

    /// Gets the queue logical offset
    #[inline]
    pub fn queue_offset(&self) -> i64 {
        self.queue_offset
    }

    /// Gets the CommitLog physical offset
    #[inline]
    pub fn commit_log_offset(&self) -> i64 {
        self.commit_log_offset
    }

    /// Gets the storage timestamp
    #[inline]
    pub fn store_timestamp(&self) -> i64 {
        self.store_timestamp
    }

    /// Gets the storage host address
    #[inline]
    pub fn store_host(&self) -> SocketAddr {
        self.store_host
    }

    /// Gets the storage size in bytes
    #[inline]
    pub fn store_size(&self) -> i32 {
        self.store_size
    }

    /// Serializes store_host to bytes
    pub fn store_host_bytes(&self) -> Bytes {
        Self::socket_addr_to_bytes(&self.store_host)
    }

    fn socket_addr_to_bytes(addr: &SocketAddr) -> Bytes {
        match addr {
            SocketAddr::V4(v4) => {
                let mut buf = BytesMut::with_capacity(8);
                buf.put_slice(&v4.ip().octets());
                buf.put_i32(v4.port() as i32);
                buf.freeze()
            }
            SocketAddr::V6(v6) => {
                let mut buf = BytesMut::with_capacity(20);
                buf.put_slice(&v6.ip().octets());
                buf.put_i32(v6.port() as i32);
                buf.freeze()
            }
        }
    }
}

impl Default for StorageMetadata {
    fn default() -> Self {
        Self {
            broker_name: CheetahString::new(),
            queue_id: 0,
            queue_offset: 0,
            commit_log_offset: 0,
            store_timestamp: 0,
            store_host: SocketAddr::from(([127, 0, 0, 1], 10911)),
            store_size: 0,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_storage_metadata_creation() {
        let addr: SocketAddr = "192.168.1.100:10911".parse().unwrap();
        let metadata = StorageMetadata::new(
            CheetahString::from_static_str("broker-a"),
            0,
            1000,
            50000,
            1234567890,
            addr,
            1024,
        );

        assert_eq!(metadata.broker_name(), "broker-a");
        assert_eq!(metadata.queue_id(), 0);
        assert_eq!(metadata.queue_offset(), 1000);
        assert_eq!(metadata.commit_log_offset(), 50000);
        assert_eq!(metadata.store_timestamp(), 1234567890);
        assert_eq!(metadata.store_host(), addr);
        assert_eq!(metadata.store_size(), 1024);
    }

    #[test]
    fn test_store_host_bytes_v4() {
        let addr: SocketAddr = "192.168.1.100:10911".parse().unwrap();
        let metadata = StorageMetadata::new(CheetahString::new(), 0, 0, 0, 0, addr, 0);

        let bytes = metadata.store_host_bytes();
        assert_eq!(bytes.len(), 8);
    }

    #[test]
    fn test_store_host_bytes_v6() {
        let addr: SocketAddr = "[::1]:10911".parse().unwrap();
        let metadata = StorageMetadata::new(CheetahString::new(), 0, 0, 0, 0, addr, 0);

        let bytes = metadata.store_host_bytes();
        assert_eq!(bytes.len(), 20);
    }

    #[test]
    fn test_default() {
        let metadata = StorageMetadata::default();
        assert_eq!(metadata.broker_name(), "");
        assert_eq!(metadata.queue_id(), 0);
        assert_eq!(metadata.queue_offset(), 0);
        assert_eq!(metadata.commit_log_offset(), 0);
        assert_eq!(metadata.store_size(), 0);
    }

    #[test]
    fn test_clone_and_eq_keep_broker_name_contract() {
        let addr: SocketAddr = "192.168.1.100:10911".parse().unwrap();
        let metadata = StorageMetadata::new(
            CheetahString::from_string(format!("broker-{}", "x".repeat(64))),
            1,
            100,
            5000,
            999999,
            addr,
            2048,
        );

        let cloned = metadata.clone();

        assert_eq!(cloned, metadata);
        assert_eq!(cloned.broker_name(), metadata.broker_name());
    }
}