mcrx-core

mcrx-core is a runtime-agnostic and portable IPv4 multicast receiver library for ASM ((*, G)) and SSM ((S, G)).

It is built for applications and integrations that want a small multicast receive core with explicit lifecycle and socket ownership control.

Highlights

• IPv4 ASM and SSM receive support
• Non-blocking receive API
• Explicit subscription lifecycle: add, join, leave, remove
• Multiple concurrent subscriptions with fair receive across them
• Caller-provided socket support
• Event-loop friendly socket borrowing and extraction APIs
• Optional Tokio adapter via the tokio feature
• Optional receive metadata on platforms that expose it
• Optional metrics via the metrics feature

Install

cargo add mcrx-core

With the optional Tokio adapter:

cargo add mcrx-core --features tokio

With optional metrics:

cargo add mcrx-core --features metrics

Quick Start

use mcrx_core::{Context, SubscriptionConfig};
use std::net::Ipv4Addr;

let mut ctx = Context::new();

let config = SubscriptionConfig::asm(Ipv4Addr::new(239, 1, 2, 3), 5000);
let id = ctx.add_subscription(config) ?;
ctx.join_subscription(id) ?;

if let Some(packet) = ctx.try_recv_any() ? {
println ! ("received {} bytes", packet.payload.len());
}

Existing Sockets

Use add_subscription_with_socket() when you need to create or bind the socket yourself:

use mcrx_core::{Context, SubscriptionConfig};
use socket2::{Domain, Protocol, SockAddr, Socket, Type};
use std::net::{Ipv4Addr, SocketAddrV4};

let mut ctx = Context::new();
let config = SubscriptionConfig::asm(Ipv4Addr::new(239, 1, 2, 3), 5000);

let socket = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP)) ?;
socket.set_reuse_address(true) ?;
socket.bind( & SockAddr::from(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 5000))) ?;

let id = ctx.add_subscription_with_socket(config, socket) ?;
ctx.join_subscription(id) ?;

Event Loop Integration

Borrow the live socket from a subscription:

let subscription = ctx.get_subscription(id).unwrap();
let socket = subscription.socket();

#[cfg(unix)]
let raw = subscription.as_raw_fd();

Or extract the subscription and move it into another loop or runtime:

let subscription = ctx.take_subscription(id).unwrap();
let parts = subscription.into_parts();
let socket = parts.socket;

Tokio Integration

With the tokio feature enabled, you can wrap an extracted subscription and await packets asynchronously:

use mcrx_core::TokioSubscription;

let subscription = ctx.take_subscription(id).unwrap();
let subscription = TokioSubscription::new(subscription) ?;
let packet = subscription.recv().await?;

Run the Tokio example with:

cargo run --features tokio --bin mcrx_tokio_recv -- 239.1.2.3 5000

Optional Receive Metadata

If you need more delivery context than source, group, port, and payload, use the metadata-aware receive APIs:

let subscription = ctx.get_subscription(id).unwrap();
if let Some(packet) = subscription.try_recv_with_metadata() ? {
println ! ("socket addr: {:?}", packet.metadata.socket_local_addr);
println ! ("destination ip: {:?}", packet.metadata.destination_local_ip);
}

Demo Binaries

Basic receiver:

cargo run --bin mcrx_recv -- 239.1.2.3 5000

Sender:

cargo run --bin mcrx_send -- 239.1.2.3 5000 hello

Tokio receiver:

cargo run --features tokio --bin mcrx_tokio_recv -- 239.1.2.3 5000

Metadata inspection receiver:

cargo run --bin mcrx_recv_meta -- 239.1.2.3 5000

Documentation

• Usage Guide
• Architecture
• Demo Binaries
• Metrics
• Design Decisions

Platform Support

Compatibility

ASM cross-platform compatibility:

SSM cross-platform compatibility:

Notes

• macOS may temporarily emit IGMPv2 reports in some SSM setups
• that can break SSM behavior on the network until the host state recovers

License

BSD 2-Clause