netgauze_bmp_service/
lib.rs

1// Copyright (C) 2022-present The NetGauze 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
12// implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16use netgauze_bmp_pkt::BmpMessage;
17use netgauze_bmp_pkt::codec::BmpCodecDecoderError;
18use serde::{Deserialize, Serialize};
19use std::fmt::{Display, Formatter};
20use std::io;
21use std::net::SocketAddr;
22use std::sync::Arc;
23
24pub mod actor;
25pub mod handle;
26pub mod server;
27pub mod supervisor;
28pub mod transport;
29
30pub type ActorId = u32;
31pub type SubscriberId = u32;
32pub type BmpRequest = (AddrInfo, BmpMessage);
33
34pub type BmpSender = async_channel::Sender<Arc<BmpRequest>>;
35pub type BmpReceiver = async_channel::Receiver<Arc<BmpRequest>>;
36
37pub fn create_bmp_channel(buffer_size: usize) -> (BmpSender, BmpReceiver) {
38    async_channel::bounded(buffer_size)
39}
40
41#[derive(Debug, Clone)]
42pub struct Subscription {
43    actor_id: ActorId,
44    id: SubscriberId,
45}
46
47impl Subscription {
48    pub const fn new(actor_id: ActorId, id: SubscriberId) -> Self {
49        Self { actor_id, id }
50    }
51
52    pub const fn actor_id(&self) -> ActorId {
53        self.actor_id
54    }
55
56    pub const fn id(&self) -> SubscriberId {
57        self.id
58    }
59}
60impl Display for Subscription {
61    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
62        write!(
63            f,
64            "Subscription {{ actor_id: {}, id: {} }}",
65            self.actor_id, self.id
66        )
67    }
68}
69
70/// Capture the address of both sides of a socket
71#[derive(Copy, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Debug, Serialize, Deserialize)]
72pub struct AddrInfo {
73    local_socket: SocketAddr,
74    remote_socket: SocketAddr,
75}
76
77impl AddrInfo {
78    pub const fn new(local_socket: SocketAddr, remote_socket: SocketAddr) -> Self {
79        Self {
80            local_socket,
81            remote_socket,
82        }
83    }
84
85    pub const fn local_socket(&self) -> SocketAddr {
86        self.local_socket
87    }
88
89    pub const fn remote_socket(&self) -> SocketAddr {
90        self.remote_socket
91    }
92}
93
94/// Associate a value with a tag
95#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
96pub struct TaggedData<T, V> {
97    tag: T,
98    value: V,
99}
100
101impl<T: Copy, V> TaggedData<T, V> {
102    pub const fn new(tag: T, value: V) -> Self {
103        Self { tag, value }
104    }
105
106    pub const fn tag(&self) -> T {
107        self.tag
108    }
109
110    pub const fn value(&self) -> &V {
111        &self.value
112    }
113}
114
115impl Display for TaggedData<AddrInfo, BmpCodecDecoderError> {
116    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
117        write!(f, "{self:?}")
118    }
119}
120
121impl std::error::Error for TaggedData<AddrInfo, BmpCodecDecoderError> {}
122
123/// Enable socket reuse and bind to a device or a VRF on selected platforms for
124/// TCP. Binding to device or VRF is supported on: MacOS and Linux.
125pub fn new_tcp_reuse_port(
126    local_addr: SocketAddr,
127    device: Option<String>,
128    backlog: i32,
129) -> io::Result<tokio::net::TcpListener> {
130    let tcp_sock = socket2::Socket::new(
131        if local_addr.is_ipv4() {
132            socket2::Domain::IPV4
133        } else {
134            socket2::Domain::IPV6
135        },
136        socket2::Type::STREAM,
137        None,
138    )?;
139    tcp_sock.set_reuse_address(true)?;
140    #[cfg(all(unix, not(any(target_os = "solaris", target_os = "illumos"))))]
141    tcp_sock.set_reuse_port(true)?;
142    #[cfg(unix)]
143    tcp_sock.set_cloexec(true)?;
144    tcp_sock.set_nonblocking(true)?;
145
146    #[cfg(any(
147        target_os = "ios",
148        target_os = "macos",
149        target_os = "tvos",
150        target_os = "watchos",
151        target_os = "android",
152        target_os = "fuchsia",
153        target_os = "linux"
154    ))]
155    if let Some(name) = device {
156        #[cfg(any(
157            target_os = "ios",
158            target_os = "macos",
159            target_os = "tvos",
160            target_os = "watchos",
161        ))]
162        {
163            let c_str = std::ffi::CString::new(name)?;
164            let c_index = unsafe { libc::if_nametoindex(c_str.as_ptr() as *const libc::c_char) };
165            let index = std::num::NonZeroU32::new(c_index as u32);
166            if local_addr.is_ipv4() {
167                tcp_sock.bind_device_by_index_v4(index)?;
168            } else {
169                tcp_sock.bind_device_by_index_v6(index)?;
170            }
171        }
172        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
173        tcp_sock.bind_device(Some(name.as_bytes()))?
174    }
175
176    tcp_sock.bind(&socket2::SockAddr::from(local_addr))?;
177    tcp_sock.listen(backlog)?;
178    let tcp_sock: std::net::TcpListener = tcp_sock.into();
179    tokio::net::TcpListener::from_std(tcp_sock)
180}
netgauze_bmp_service/lib.rs

netgauze_bmp_service/
lib.rs
