Struct Publisher 

pub struct Publisher { /* private fields */ }

A concurrently usable publisher. Dropping it releases its registration.

Implementations

impl Publisher

pub fn open(options: &Options) -> Result<Self>

Creates or joins a queue with the supplied identity and capacity.

Errors

Rejects invalid options, capacity mismatches, exhausted registrations, publisher limits (publishers only), and operating system failures.

Examples found in repository

examples/benchmark.rs (line 9)

fn main() {
    const ITERATIONS: usize = 1_000_000;
    for size in [3, 50, 1024] {
        let options = Options::new(format!("b{}x{size}", std::process::id()), 1 << 20);
        let publisher = Publisher::open(&options).unwrap();
        let subscriber = Subscriber::open(&options).unwrap();
        let message = vec![42; size];
        let mut received = vec![0; size];
        let mut samples = Vec::new();
        for round in 0..12 {
            let start = Instant::now();
            for _ in 0..ITERATIONS {
                publisher.try_send(black_box(&message)).unwrap();
                assert_eq!(
                    subscriber.try_recv_into(black_box(&mut received)).unwrap(),
                    Some(size)
                );
                black_box(&received);
            }
            if round >= 4 {
                samples.push(start.elapsed().as_nanos() as f64 / ITERATIONS as f64);
            }
        }
        let mean = samples.iter().sum::<f64>() / samples.len() as f64;
        let deviation = (samples.iter().map(|x| (x - mean).powi(2)).sum::<f64>()
            / (samples.len() - 1) as f64)
            .sqrt();
        println!(
            "{size} bytes: {mean:.2} ns/roundtrip, stddev {deviation:.2}, samples {samples:?}"
        );
    }
}

examples/interop.rs (line 28)

fn main() {
    let args = std::env::args().collect::<Vec<_>>();
    let options = Options::new(
        &args[2],
        std::env::var("INTEROP_CAPACITY")
            .map(|v| v.parse().unwrap())
            .unwrap_or(4096),
    )
    .with_path(&args[3]);
    let count = args[4].parse::<usize>().unwrap();
    if args[1].starts_with("hold-") {
        let _publisher;
        let _subscriber;
        if args[1] == "hold-publisher" {
            _publisher = Publisher::open(&options).unwrap();
        } else {
            _subscriber = Subscriber::open(&options).unwrap();
        }
        println!("READY");
        io::stdout().flush().unwrap();
        io::stdin().read_line(&mut String::new()).unwrap();
    } else if args[1] == "publish" {
        let publisher = Publisher::open(&options).unwrap();
        let start = args.get(5).map_or(0, |s| s.parse::<usize>().unwrap());
        if args.len() > 5 {
            println!("READY");
            io::stdout().flush().unwrap();
            io::stdin().read_line(&mut String::new()).unwrap();
        }
        for i in start..start + count {
            let data = message(i);
            while let Err(error) = publisher.try_send(&data) {
                assert!(error.is_full(), "{error}");
                std::thread::yield_now();
            }
        }
    } else {
        let subscriber = Subscriber::open(&options).unwrap();
        println!("READY");
        if args[1] == "collect" {
            loop {
                let data = subscriber
                    .recv_timeout(Duration::from_secs(30))
                    .unwrap()
                    .unwrap();
                if data.is_empty() {
                    break;
                }
                let id = u64::from_le_bytes(data[..8].try_into().unwrap()) as usize;
                assert_eq!(data, message(id));
                println!("{id}");
            }
            return;
        }
        for i in 0..count {
            assert_eq!(
                subscriber
                    .recv_timeout(Duration::from_secs(30))
                    .unwrap()
                    .unwrap(),
                message(i)
            );
        }
    }
}

pub fn try_send(&self, message: &[u8]) -> Result<()>

Returns Error::Full when there is insufficient space or recovery closes admission.

Examples found in repository

examples/benchmark.rs (line 17)

fn main() {
    const ITERATIONS: usize = 1_000_000;
    for size in [3, 50, 1024] {
        let options = Options::new(format!("b{}x{size}", std::process::id()), 1 << 20);
        let publisher = Publisher::open(&options).unwrap();
        let subscriber = Subscriber::open(&options).unwrap();
        let message = vec![42; size];
        let mut received = vec![0; size];
        let mut samples = Vec::new();
        for round in 0..12 {
            let start = Instant::now();
            for _ in 0..ITERATIONS {
                publisher.try_send(black_box(&message)).unwrap();
                assert_eq!(
                    subscriber.try_recv_into(black_box(&mut received)).unwrap(),
                    Some(size)
                );
                black_box(&received);
            }
            if round >= 4 {
                samples.push(start.elapsed().as_nanos() as f64 / ITERATIONS as f64);
            }
        }
        let mean = samples.iter().sum::<f64>() / samples.len() as f64;
        let deviation = (samples.iter().map(|x| (x - mean).powi(2)).sum::<f64>()
            / (samples.len() - 1) as f64)
            .sqrt();
        println!(
            "{size} bytes: {mean:.2} ns/roundtrip, stddev {deviation:.2}, samples {samples:?}"
        );
    }
}

examples/interop.rs (line 45)

fn main() {
    let args = std::env::args().collect::<Vec<_>>();
    let options = Options::new(
        &args[2],
        std::env::var("INTEROP_CAPACITY")
            .map(|v| v.parse().unwrap())
            .unwrap_or(4096),
    )
    .with_path(&args[3]);
    let count = args[4].parse::<usize>().unwrap();
    if args[1].starts_with("hold-") {
        let _publisher;
        let _subscriber;
        if args[1] == "hold-publisher" {
            _publisher = Publisher::open(&options).unwrap();
        } else {
            _subscriber = Subscriber::open(&options).unwrap();
        }
        println!("READY");
        io::stdout().flush().unwrap();
        io::stdin().read_line(&mut String::new()).unwrap();
    } else if args[1] == "publish" {
        let publisher = Publisher::open(&options).unwrap();
        let start = args.get(5).map_or(0, |s| s.parse::<usize>().unwrap());
        if args.len() > 5 {
            println!("READY");
            io::stdout().flush().unwrap();
            io::stdin().read_line(&mut String::new()).unwrap();
        }
        for i in start..start + count {
            let data = message(i);
            while let Err(error) = publisher.try_send(&data) {
                assert!(error.is_full(), "{error}");
                std::thread::yield_now();
            }
        }
    } else {
        let subscriber = Subscriber::open(&options).unwrap();
        println!("READY");
        if args[1] == "collect" {
            loop {
                let data = subscriber
                    .recv_timeout(Duration::from_secs(30))
                    .unwrap()
                    .unwrap();
                if data.is_empty() {
                    break;
                }
                let id = u64::from_le_bytes(data[..8].try_into().unwrap()) as usize;
                assert_eq!(data, message(id));
                println!("{id}");
            }
            return;
        }
        for i in 0..count {
            assert_eq!(
                subscriber
                    .recv_timeout(Duration::from_secs(30))
                    .unwrap()
                    .unwrap(),
                message(i)
            );
        }
    }
}

pub fn try_send_batch(&self, messages: &[&[u8]]) -> Result<usize>

Publishes an ordered prefix, amortizing publisher admission across a batch. Returns the committed prefix length. A short count (including zero) means full, recovery, or a mid-batch error; retry the unsent suffix to observe a persistent error. An error before any commit is returned immediately.

Trait Implementations

impl Debug for Publisher

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for Publisher

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.