mcrx_core/
subscription.rs

1use crate::config::SubscriptionConfig;
2use crate::error::McrxError;
3#[cfg(feature = "metrics")]
4use crate::metrics::SubscriptionMetricsSnapshot;
5use crate::packet::{Packet, PacketWithMetadata};
6use crate::platform::{ReceiveSocket, recv_packet, recv_packet_with_metadata, socket_local_addr};
7use socket2::Socket;
8use std::net::SocketAddr;
9#[cfg(feature = "metrics")]
10use std::sync::Mutex;
11#[cfg(feature = "metrics")]
12use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
13#[cfg(feature = "metrics")]
14use std::time::SystemTime;
15
16/// Identifies a subscription within a context.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
18pub struct SubscriptionId(pub u64);
19
20/// Represents the lifecycle state of a subscription.
21///
22/// A subscription is always associated with a bound socket, but may or may not
23/// currently be joined to a multicast group.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum SubscriptionState {
26    /// Socket is bound but multicast group is not joined.
27    Bound,
28    /// Multicast group is joined.
29    Joined,
30}
31
32/// Owned subscription state that can be extracted from a context and moved into
33/// another event loop or runtime.
34#[derive(Debug)]
35pub struct SubscriptionParts {
36    /// The subscription's ID inside the originating context.
37    pub id: SubscriptionId,
38    /// The multicast configuration associated with the socket.
39    pub config: SubscriptionConfig,
40    /// The owned socket for external integration.
41    pub socket: Socket,
42    /// The lifecycle state at the time of extraction.
43    pub state: SubscriptionState,
44}
45
46#[cfg(feature = "metrics")]
47#[derive(Debug)]
48struct LastPacketMetrics {
49    source: Option<SocketAddr>,
50    receive_at: Option<SystemTime>,
51}
52
53#[cfg(feature = "metrics")]
54#[derive(Debug)]
55struct SubscriptionMetricsInner {
56    packets_received: AtomicU64,
57    bytes_received: AtomicU64,
58    would_block_count: AtomicU64,
59    receive_errors: AtomicU64,
60    join_count: AtomicU64,
61    leave_count: AtomicU64,
62    last_payload_len: AtomicUsize,
63    last_packet: Mutex<LastPacketMetrics>,
64}
65
66#[cfg(feature = "metrics")]
67impl Default for SubscriptionMetricsInner {
68    fn default() -> Self {
69        Self {
70            packets_received: AtomicU64::new(0),
71            bytes_received: AtomicU64::new(0),
72            would_block_count: AtomicU64::new(0),
73            receive_errors: AtomicU64::new(0),
74            join_count: AtomicU64::new(0),
75            leave_count: AtomicU64::new(0),
76            last_payload_len: AtomicUsize::new(usize::MAX),
77            last_packet: Mutex::new(LastPacketMetrics {
78                source: None,
79                receive_at: None,
80            }),
81        }
82    }
83}
84
85/// Represents a registered subscription stored inside a context.
86#[derive(Debug)]
87pub struct Subscription {
88    id: SubscriptionId,
89    config: SubscriptionConfig,
90    socket: ReceiveSocket,
91    state: SubscriptionState,
92    #[cfg(feature = "metrics")]
93    metrics: SubscriptionMetricsInner,
94}
95
96impl Subscription {
97    #[cfg(feature = "metrics")]
98    fn lock_unpoisoned<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
99        match mutex.lock() {
100            Ok(guard) => guard,
101            Err(poisoned) => poisoned.into_inner(),
102        }
103    }
104
105    fn with_socket(id: SubscriptionId, config: SubscriptionConfig, socket: ReceiveSocket) -> Self {
106        Self {
107            id,
108            config,
109            socket,
110            state: SubscriptionState::Bound,
111            #[cfg(feature = "metrics")]
112            metrics: SubscriptionMetricsInner::default(),
113        }
114    }
115
116    #[cfg(feature = "metrics")]
117    fn record_received_packet(&self, packet: &Packet) {
118        self.metrics
119            .packets_received
120            .fetch_add(1, Ordering::Relaxed);
121        self.metrics
122            .bytes_received
123            .fetch_add(packet.payload.len() as u64, Ordering::Relaxed);
124        self.metrics
125            .last_payload_len
126            .store(packet.payload.len(), Ordering::Relaxed);
127        let mut last_packet = Self::lock_unpoisoned(&self.metrics.last_packet);
128        last_packet.source = Some(packet.source);
129        last_packet.receive_at = Some(SystemTime::now());
130    }
131
132    #[cfg(feature = "metrics")]
133    fn record_would_block(&self) {
134        self.metrics
135            .would_block_count
136            .fetch_add(1, Ordering::Relaxed);
137    }
138
139    #[cfg(feature = "metrics")]
140    fn record_receive_error(&self) {
141        self.metrics.receive_errors.fetch_add(1, Ordering::Relaxed);
142    }
143
144    /// Creates a new subscription from an ID, configuration, and socket.
145    ///
146    /// This is a low-level constructor. Callers are responsible for providing a
147    /// socket that is compatible with the subscription configuration. For the
148    /// checked convenience path, prefer `Context::add_subscription_with_socket()`.
149    /// The socket is switched to non-blocking mode and destination metadata is
150    /// enabled; a setup failure is returned by the first receive operation.
151    pub fn new(id: SubscriptionId, config: SubscriptionConfig, socket: Socket) -> Self {
152        let receive_socket = ReceiveSocket::adopt(socket, &config);
153        Self::with_socket(id, config, receive_socket)
154    }
155
156    pub(crate) fn from_receive_socket(
157        id: SubscriptionId,
158        config: SubscriptionConfig,
159        socket: ReceiveSocket,
160    ) -> Self {
161        Self::with_socket(id, config, socket)
162    }
163
164    /// Returns the subscription's ID.
165    pub fn id(&self) -> SubscriptionId {
166        self.id
167    }
168
169    /// Returns a read-only reference to the subscription's configuration.
170    pub fn config(&self) -> &SubscriptionConfig {
171        &self.config
172    }
173
174    /// Returns a read-only reference to the subscription's socket.
175    pub fn socket(&self) -> &Socket {
176        self.socket.socket()
177    }
178
179    /// Returns a mutable reference to the subscription's socket.
180    ///
181    /// This is useful when an external event loop or registry needs mutable
182    /// socket access during registration.
183    pub fn socket_mut(&mut self) -> &mut Socket {
184        self.socket.socket_mut()
185    }
186
187    /// Attempts to receive a single packet without blocking.
188    ///
189    /// Returns:
190    /// - `Ok(Some(packet))` if a packet was received,
191    /// - `Ok(None)` if no packet is currently available,
192    /// - `Err(...)` on an actual receive failure.
193    pub fn try_recv(&self) -> Result<Option<Packet>, McrxError> {
194        if !self.is_joined() {
195            return Err(McrxError::SubscriptionNotJoined);
196        }
197        match recv_packet(&self.socket, self.id, &self.config) {
198            Ok(Some(packet)) => {
199                #[cfg(feature = "metrics")]
200                self.record_received_packet(&packet);
201
202                Ok(Some(packet))
203            }
204            Ok(None) => {
205                #[cfg(feature = "metrics")]
206                self.record_would_block();
207
208                Ok(None)
209            }
210            Err(err) => {
211                #[cfg(feature = "metrics")]
212                self.record_receive_error();
213
214                Err(err)
215            }
216        }
217    }
218
219    /// Attempts to receive a single packet together with richer receive metadata
220    /// without blocking.
221    pub fn try_recv_with_metadata(&self) -> Result<Option<PacketWithMetadata>, McrxError> {
222        if !self.is_joined() {
223            return Err(McrxError::SubscriptionNotJoined);
224        }
225
226        match recv_packet_with_metadata(&self.socket, self.id, &self.config) {
227            Ok(Some(packet)) => {
228                #[cfg(feature = "metrics")]
229                self.record_received_packet(&packet.packet);
230
231                Ok(Some(packet))
232            }
233            Ok(None) => {
234                #[cfg(feature = "metrics")]
235                self.record_would_block();
236
237                Ok(None)
238            }
239            Err(err) => {
240                #[cfg(feature = "metrics")]
241                self.record_receive_error();
242
243                Err(err)
244            }
245        }
246    }
247
248    /// Returns the raw Unix file descriptor of the underlying socket.
249    ///
250    /// This is useful for integrating subscriptions into external event loops.
251    #[cfg(unix)]
252    pub fn as_raw_fd(&self) -> std::os::fd::RawFd {
253        use std::os::fd::AsRawFd;
254        self.socket().as_raw_fd()
255    }
256
257    /// Returns the raw Windows socket handle of the underlying socket.
258    ///
259    /// This is useful for integrating subscriptions into external event loops.
260    #[cfg(windows)]
261    pub fn as_raw_socket(&self) -> std::os::windows::io::RawSocket {
262        use std::os::windows::io::AsRawSocket;
263        self.socket().as_raw_socket()
264    }
265
266    /// Returns the local socket address currently bound by this subscription.
267    pub fn local_addr(&self) -> Result<SocketAddr, McrxError> {
268        self.socket
269            .local_addr()
270            .or_else(|_| socket_local_addr(self.socket()))
271    }
272
273    /// Consumes the subscription and returns its owned socket.
274    ///
275    /// This is useful when handing a joined or bound socket off to an external
276    /// event loop or async runtime.
277    pub fn into_socket(self) -> Socket {
278        self.socket.into_socket()
279    }
280
281    /// Consumes the subscription and returns all owned parts.
282    pub fn into_parts(self) -> SubscriptionParts {
283        SubscriptionParts {
284            id: self.id,
285            config: self.config,
286            socket: self.socket.into_socket(),
287            state: self.state,
288        }
289    }
290
291    /// Returns the current lifecycle state of the subscription.
292    ///
293    /// This can be used by callers to inspect whether the subscription is
294    /// currently joined to its multicast group or only bound.
295    pub fn state(&self) -> SubscriptionState {
296        self.state
297    }
298
299    /// Returns a snapshot of the subscription's current metrics.
300    ///
301    /// Counter values in the returned snapshot are cumulative and can be
302    /// compared against a later snapshot using `delta_since()`.
303    #[cfg(feature = "metrics")]
304    pub fn metrics_snapshot(&self) -> SubscriptionMetricsSnapshot {
305        let last_payload_len = match self.metrics.last_payload_len.load(Ordering::Relaxed) {
306            usize::MAX => None,
307            payload_len => Some(payload_len),
308        };
309        let last_packet = Self::lock_unpoisoned(&self.metrics.last_packet);
310
311        SubscriptionMetricsSnapshot {
312            packets_received: self.metrics.packets_received.load(Ordering::Relaxed),
313            bytes_received: self.metrics.bytes_received.load(Ordering::Relaxed),
314            would_block_count: self.metrics.would_block_count.load(Ordering::Relaxed),
315            receive_errors: self.metrics.receive_errors.load(Ordering::Relaxed),
316            join_count: self.metrics.join_count.load(Ordering::Relaxed),
317            leave_count: self.metrics.leave_count.load(Ordering::Relaxed),
318            last_payload_len,
319            last_source: last_packet.source,
320            last_receive_at: last_packet.receive_at,
321            captured_at: SystemTime::now(),
322        }
323    }
324
325    /// Returns `true` if the subscription is currently joined to its multicast group.
326    ///
327    /// This is a convenience helper for checking whether the subscription is in
328    /// the `SubscriptionState::Joined` state.
329    pub fn is_joined(&self) -> bool {
330        matches!(self.state, SubscriptionState::Joined)
331    }
332
333    /// Marks the subscription as joined.
334    ///
335    /// This should be called after a successful multicast join operation on the
336    /// underlying socket.
337    ///
338    /// Returns an error if the subscription is already in the joined state.
339    pub fn mark_joined(&mut self) -> Result<(), McrxError> {
340        if self.state == SubscriptionState::Joined {
341            return Err(McrxError::SubscriptionAlreadyJoined);
342        }
343
344        self.state = SubscriptionState::Joined;
345
346        #[cfg(feature = "metrics")]
347        self.metrics.join_count.fetch_add(1, Ordering::Relaxed);
348
349        Ok(())
350    }
351
352    /// Marks the subscription as bound (not joined).
353    ///
354    /// This should be called after leaving a multicast group, while keeping the
355    /// underlying socket open and bound.
356    ///
357    /// Returns an error if the subscription is already in the bound state.
358    pub fn mark_bound(&mut self) -> Result<(), McrxError> {
359        if self.state == SubscriptionState::Bound {
360            return Err(McrxError::SubscriptionNotJoined);
361        }
362
363        self.state = SubscriptionState::Bound;
364
365        #[cfg(feature = "metrics")]
366        self.metrics.leave_count.fetch_add(1, Ordering::Relaxed);
367
368        Ok(())
369    }
370}
371
372#[cfg(unix)]
373impl std::os::fd::AsFd for Subscription {
374    fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> {
375        self.socket().as_fd()
376    }
377}
378
379#[cfg(unix)]
380impl std::os::fd::AsRawFd for Subscription {
381    fn as_raw_fd(&self) -> std::os::fd::RawFd {
382        self.socket().as_raw_fd()
383    }
384}
385
386#[cfg(windows)]
387impl std::os::windows::io::AsSocket for Subscription {
388    fn as_socket(&self) -> std::os::windows::io::BorrowedSocket<'_> {
389        self.socket().as_socket()
390    }
391}
392
393#[cfg(windows)]
394impl std::os::windows::io::AsRawSocket for Subscription {
395    fn as_raw_socket(&self) -> std::os::windows::io::RawSocket {
396        self.socket().as_raw_socket()
397    }
398}
399
400#[cfg(test)]
401mod tests {
402    use super::*;
403    use crate::config::{SourceFilter, SubscriptionConfig};
404    use crate::platform;
405    use crate::test_support::{
406        ipv6_group_socket_addr, ipv6_multicast_loopback_available, make_multicast_sender,
407        make_multicast_sender_v6, make_multicast_sender_v6_for_source,
408        sample_config_on_unused_port, sample_config_v6_on_unused_port,
409        sample_ssm_receive_config_v6_on_unused_port, unused_udp_port_v4,
410    };
411    use socket2::SockRef;
412    use std::io::ErrorKind;
413    use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, UdpSocket};
414    use std::time::{Duration, Instant};
415
416    fn test_ssm_config(port: u16, interface: Ipv4Addr) -> SubscriptionConfig {
417        SubscriptionConfig {
418            group: IpAddr::V4(Ipv4Addr::new(232, 1, 2, 3)),
419            source: SourceFilter::Source(IpAddr::V4(interface)),
420            dst_port: port,
421            interface: Some(IpAddr::V4(interface)),
422            interface_index: None,
423        }
424    }
425
426    fn ipv4_group(config: &SubscriptionConfig) -> Ipv4Addr {
427        config.ipv4_membership().unwrap().group
428    }
429
430    fn ipv4_group_socket_addr(config: &SubscriptionConfig) -> SocketAddrV4 {
431        SocketAddrV4::new(ipv4_group(config), config.dst_port)
432    }
433
434    fn primary_ipv4() -> Ipv4Addr {
435        let probe = UdpSocket::bind(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0)).unwrap();
436        probe
437            .connect(SocketAddrV4::new(Ipv4Addr::new(8, 8, 8, 8), 9))
438            .unwrap();
439
440        match probe.local_addr().unwrap() {
441            SocketAddr::V4(addr) => *addr.ip(),
442            SocketAddr::V6(_) => panic!("expected an IPv4 local address for SSM test"),
443        }
444    }
445
446    fn recv_next_subscription_packet(subscription: &Subscription, deadline: Instant) -> Packet {
447        loop {
448            match subscription.try_recv().unwrap() {
449                Some(packet) => return packet,
450                None if Instant::now() < deadline => {
451                    std::thread::sleep(Duration::from_millis(10));
452                }
453                None => panic!("timed out waiting for packet"),
454            }
455        }
456    }
457
458    fn assert_pktinfo_metadata(packet: &PacketWithMetadata, expected_destination: IpAddr) {
459        #[cfg(any(
460            target_os = "linux",
461            target_os = "android",
462            windows,
463            target_vendor = "apple",
464            target_os = "freebsd",
465            target_os = "dragonfly",
466            target_os = "netbsd",
467            target_os = "openbsd"
468        ))]
469        {
470            assert_eq!(
471                packet.metadata.destination_local_ip,
472                Some(expected_destination)
473            );
474            assert!(packet.metadata.ingress_interface_index.is_some());
475        }
476
477        #[cfg(not(any(
478            target_os = "linux",
479            target_os = "android",
480            windows,
481            target_vendor = "apple",
482            target_os = "freebsd",
483            target_os = "dragonfly",
484            target_os = "netbsd",
485            target_os = "openbsd"
486        )))]
487        {
488            let _ = expected_destination;
489            assert_eq!(packet.metadata.destination_local_ip, None);
490            assert_eq!(packet.metadata.ingress_interface_index, None);
491        }
492    }
493
494    #[test]
495    fn try_recv_returns_none_when_no_packet_is_available() {
496        let config = sample_config_on_unused_port();
497        let socket = platform::open_bound_socket(&config).unwrap();
498        let mut subscription = Subscription::from_receive_socket(SubscriptionId(1), config, socket);
499        platform::join_multicast_group(subscription.socket(), subscription.config()).unwrap();
500        subscription.mark_joined().unwrap();
501
502        let result = subscription.try_recv().unwrap();
503
504        assert!(result.is_none());
505    }
506
507    #[test]
508    fn low_level_constructor_switches_socket_to_nonblocking() {
509        let config = sample_config_on_unused_port();
510        let socket = Socket::from(
511            UdpSocket::bind(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, config.dst_port)).unwrap(),
512        );
513        let mut subscription = Subscription::new(SubscriptionId(1), config, socket);
514        subscription.mark_joined().unwrap();
515
516        let started = Instant::now();
517        assert!(subscription.try_recv().unwrap().is_none());
518        assert!(started.elapsed() < Duration::from_millis(100));
519    }
520
521    #[test]
522    fn try_recv_rejects_unicast_sent_to_the_bound_port() {
523        let config = sample_config_on_unused_port();
524        let socket = platform::open_bound_socket(&config).unwrap();
525        let mut subscription =
526            Subscription::from_receive_socket(SubscriptionId(1), config.clone(), socket);
527        platform::join_multicast_group(subscription.socket(), subscription.config()).unwrap();
528        subscription.mark_joined().unwrap();
529
530        let sender = UdpSocket::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0)).unwrap();
531        let payload = b"hello multicast core";
532
533        sender
534            .send_to(
535                payload,
536                SocketAddrV4::new(Ipv4Addr::LOCALHOST, config.dst_port),
537            )
538            .unwrap();
539
540        let deadline = Instant::now() + Duration::from_millis(200);
541        while Instant::now() < deadline {
542            assert!(subscription.try_recv().unwrap().is_none());
543            std::thread::sleep(Duration::from_millis(5));
544        }
545
546        let multicast_sender = make_multicast_sender();
547        multicast_sender
548            .send_to(payload, ipv4_group_socket_addr(&config))
549            .unwrap();
550
551        let packet =
552            recv_next_subscription_packet(&subscription, Instant::now() + Duration::from_secs(1));
553        assert_eq!(packet.group, config.group);
554        assert_eq!(&packet.payload[..], payload);
555    }
556
557    #[test]
558    fn try_recv_with_metadata_exposes_current_socket_context() {
559        let config = sample_config_on_unused_port();
560        let socket = platform::open_bound_socket(&config).unwrap();
561        let mut subscription =
562            Subscription::from_receive_socket(SubscriptionId(1), config.clone(), socket);
563        platform::join_multicast_group(subscription.socket(), subscription.config()).unwrap();
564        subscription.mark_joined().unwrap();
565
566        let sender = make_multicast_sender();
567        let payload = b"hello detailed receive";
568
569        sender
570            .send_to(payload, ipv4_group_socket_addr(&config))
571            .unwrap();
572
573        let deadline = Instant::now() + Duration::from_secs(1);
574        let packet = loop {
575            match subscription.try_recv_with_metadata().unwrap() {
576                Some(packet) => break packet,
577                None if Instant::now() < deadline => {
578                    std::thread::sleep(Duration::from_millis(10));
579                }
580                None => panic!("timed out waiting for packet with metadata"),
581            }
582        };
583
584        assert_eq!(packet.packet.subscription_id, SubscriptionId(1));
585        assert_eq!(packet.packet.group, IpAddr::V4(ipv4_group(&config)));
586        assert_eq!(packet.packet.dst_port, config.dst_port);
587        assert_eq!(&packet.packet.payload[..], payload);
588        assert_pktinfo_metadata(&packet, IpAddr::V4(ipv4_group(&config)));
589        assert_eq!(
590            packet.metadata.socket_local_addr,
591            Some(SocketAddr::V4(SocketAddrV4::new(
592                Ipv4Addr::UNSPECIFIED,
593                config.dst_port,
594            )))
595        );
596        assert_eq!(packet.metadata.configured_interface, config.interface);
597    }
598
599    #[test]
600    fn try_recv_with_metadata_exposes_current_ipv6_socket_context() {
601        if !ipv6_multicast_loopback_available() {
602            return;
603        }
604
605        let config = sample_config_v6_on_unused_port();
606        let socket = platform::open_bound_socket(&config).unwrap();
607        let mut subscription =
608            Subscription::from_receive_socket(SubscriptionId(1), config.clone(), socket);
609        platform::join_multicast_group(subscription.socket(), subscription.config()).unwrap();
610        subscription.mark_joined().unwrap();
611
612        let sender = make_multicast_sender_v6(Ipv6Addr::LOCALHOST);
613        let payload = b"hello detailed receive ipv6";
614
615        sender
616            .send_to(payload, ipv6_group_socket_addr(&config))
617            .unwrap();
618
619        let deadline = Instant::now() + Duration::from_secs(1);
620        let packet = loop {
621            match subscription.try_recv_with_metadata().unwrap() {
622                Some(packet) => break packet,
623                None if Instant::now() < deadline => {
624                    std::thread::sleep(Duration::from_millis(10));
625                }
626                None => panic!("timed out waiting for IPv6 packet with metadata"),
627            }
628        };
629
630        assert_eq!(packet.packet.subscription_id, SubscriptionId(1));
631        assert_eq!(packet.packet.group, config.group);
632        assert_eq!(packet.packet.dst_port, config.dst_port);
633        assert_eq!(&packet.packet.payload[..], payload);
634        assert_pktinfo_metadata(&packet, config.group);
635        assert_eq!(
636            packet.metadata.socket_local_addr,
637            Some(SocketAddr::V6(SocketAddrV6::new(
638                Ipv6Addr::UNSPECIFIED,
639                config.dst_port,
640                0,
641                0,
642            )))
643        );
644        assert_eq!(packet.metadata.configured_interface, config.interface);
645        assert_eq!(
646            packet.metadata.configured_interface_index,
647            config.interface_index
648        );
649    }
650
651    #[test]
652    fn try_recv_receives_multicast_packet_from_joined_group() {
653        let config = sample_config_on_unused_port();
654        let socket = platform::open_bound_socket(&config).unwrap();
655        let mut subscription =
656            Subscription::from_receive_socket(SubscriptionId(1), config.clone(), socket);
657        platform::join_multicast_group(subscription.socket(), subscription.config()).unwrap();
658        subscription.mark_joined().unwrap();
659
660        let sender = make_multicast_sender();
661
662        let sender_port = sender.local_addr().unwrap().port();
663        let payload = b"hello real asm multicast";
664
665        sender
666            .send_to(payload, ipv4_group_socket_addr(&config))
667            .unwrap();
668
669        let deadline = Instant::now() + Duration::from_secs(1);
670        let packet = recv_next_subscription_packet(&subscription, deadline);
671
672        assert_eq!(packet.subscription_id, SubscriptionId(1));
673        assert_eq!(packet.group, IpAddr::V4(ipv4_group(&config)));
674        assert_eq!(packet.dst_port, config.dst_port);
675        assert_eq!(&packet.payload[..], payload);
676        assert_eq!(packet.source.port(), sender_port);
677    }
678
679    #[test]
680    fn try_recv_receives_ssm_packet_from_allowed_source() {
681        let interface = primary_ipv4();
682        let config = test_ssm_config(unused_udp_port_v4(), interface);
683        let socket = platform::open_bound_socket(&config).unwrap();
684        let mut subscription =
685            Subscription::from_receive_socket(SubscriptionId(1), config.clone(), socket);
686        platform::join_multicast_group(subscription.socket(), subscription.config()).unwrap();
687        subscription.mark_joined().unwrap();
688
689        let sender = UdpSocket::bind(SocketAddrV4::new(interface, 0)).unwrap();
690        sender.set_multicast_loop_v4(true).unwrap();
691        sender.set_multicast_ttl_v4(1).unwrap();
692        SockRef::from(&sender)
693            .set_multicast_if_v4(&interface)
694            .unwrap();
695
696        let sender_port = sender.local_addr().unwrap().port();
697        let payload = b"hello real ssm multicast";
698
699        match sender.send_to(payload, ipv4_group_socket_addr(&config)) {
700            Ok(_) => {}
701            Err(err)
702                if matches!(
703                    err.kind(),
704                    ErrorKind::HostUnreachable | ErrorKind::NetworkUnreachable
705                ) =>
706            {
707                // Hosted runners can lack a route for physical-interface SSM.
708                return;
709            }
710            Err(err) => panic!("failed to send IPv4 SSM test packet: {err}"),
711        }
712
713        let deadline = Instant::now() + Duration::from_secs(1);
714        let packet = recv_next_subscription_packet(&subscription, deadline);
715
716        assert_eq!(packet.subscription_id, SubscriptionId(1));
717        assert_eq!(packet.group, IpAddr::V4(ipv4_group(&config)));
718        assert_eq!(packet.dst_port, config.dst_port);
719        assert_eq!(&packet.payload[..], payload);
720        assert_eq!(packet.source.port(), sender_port);
721        assert_eq!(packet.source.ip(), IpAddr::V4(interface));
722    }
723
724    #[test]
725    fn try_recv_receives_ipv6_ssm_packet_from_allowed_source() {
726        let Some(config) = sample_ssm_receive_config_v6_on_unused_port() else {
727            return;
728        };
729        let interface = match config.source_addr().unwrap() {
730            IpAddr::V6(source) => source,
731            IpAddr::V4(_) => panic!("expected an IPv6 source for IPv6 SSM test"),
732        };
733        let socket = platform::open_bound_socket(&config).unwrap();
734        let mut subscription =
735            Subscription::from_receive_socket(SubscriptionId(1), config.clone(), socket);
736        platform::join_multicast_group(subscription.socket(), subscription.config()).unwrap();
737        subscription.mark_joined().unwrap();
738
739        let sender = make_multicast_sender_v6_for_source(interface);
740        let sender_port = sender.local_addr().unwrap().port();
741        let payload = b"hello real ipv6 ssm multicast";
742
743        sender
744            .send_to(payload, ipv6_group_socket_addr(&config))
745            .unwrap();
746
747        let deadline = Instant::now() + Duration::from_secs(1);
748        let packet = recv_next_subscription_packet(&subscription, deadline);
749
750        assert_eq!(packet.subscription_id, SubscriptionId(1));
751        assert_eq!(packet.group, config.group);
752        assert_eq!(packet.dst_port, config.dst_port);
753        assert_eq!(&packet.payload[..], payload);
754        assert_eq!(packet.source.port(), sender_port);
755        assert_eq!(packet.source.ip(), IpAddr::V6(interface));
756    }
757
758    #[test]
759    fn mark_joined_transitions_bound_to_joined_state() {
760        let config = sample_config_on_unused_port();
761        let socket = platform::open_bound_socket(&config).unwrap();
762        let mut subscription = Subscription::from_receive_socket(SubscriptionId(1), config, socket);
763
764        subscription.mark_joined().unwrap();
765
766        assert_eq!(subscription.state(), SubscriptionState::Joined);
767    }
768
769    #[test]
770    fn mark_joined_rejects_already_joined_subscription() {
771        let config = sample_config_on_unused_port();
772        let socket = platform::open_bound_socket(&config).unwrap();
773        let mut subscription = Subscription::from_receive_socket(SubscriptionId(1), config, socket);
774
775        subscription.mark_joined().unwrap();
776        let result = subscription.mark_joined();
777
778        assert!(matches!(result, Err(McrxError::SubscriptionAlreadyJoined)));
779    }
780
781    #[test]
782    fn mark_bound_transitions_joined_to_bound_state() {
783        let config = sample_config_on_unused_port();
784        let socket = platform::open_bound_socket(&config).unwrap();
785        let mut subscription = Subscription::from_receive_socket(SubscriptionId(1), config, socket);
786
787        subscription.mark_joined().unwrap();
788        subscription.mark_bound().unwrap();
789
790        assert_eq!(subscription.state(), SubscriptionState::Bound);
791    }
792
793    #[test]
794    fn mark_bound_rejects_already_bound_subscription() {
795        let config = sample_config_on_unused_port();
796        let socket = platform::open_bound_socket(&config).unwrap();
797        let mut subscription = Subscription::from_receive_socket(SubscriptionId(1), config, socket);
798
799        let result = subscription.mark_bound();
800
801        assert!(matches!(result, Err(McrxError::SubscriptionNotJoined)));
802    }
803
804    #[test]
805    fn local_addr_returns_bound_socket_address() {
806        let config = sample_config_on_unused_port();
807        let socket = platform::open_bound_socket(&config).unwrap();
808        let subscription =
809            Subscription::from_receive_socket(SubscriptionId(1), config.clone(), socket);
810
811        let local_addr = subscription.local_addr().unwrap();
812
813        assert_eq!(
814            local_addr,
815            SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, config.dst_port))
816        );
817    }
818}
mcrx_core/subscription.rs

mcrx_core/
subscription.rs
