1 
use std::thread::sleep;
2 
use std::time::Duration;
3 

            

        

            

                4
            

            

                
            

            

                
use bonsaidb::core::pubsub::{PubSub, Subscriber};

            

        

            

                5
            

            

                
            

            

                
use bonsaidb::local::config::{Builder, StorageConfiguration};

            

        

            

                6
            

            

                
            

            

                
use bonsaidb::local::Database;

            

        

            

                7
            

            

                
            

            

                

            

        

            

                8
            

            

                1
            

            

                
fn main() -> Result<(), bonsaidb::local::Error> {

            

        

            

                9
            

            

                
            

            

                
    // This example is using a database with no collections, because PubSub is a

            

        

            

                10
            

            

                
            

            

                
    // system independent of the data stored in the database.

            

        

            

                11
            

            

                1
            

            

                
    let db = Database::open::<()>(StorageConfiguration::new("pubsub.bonsaidb"))?;

            

        

            

                12
            

            

                
            

            

                

            

        

            

                13
            

            

                1
            

            

                
    let subscriber = db.create_subscriber()?;

            

        

            

                14
            

            

                
            

            

                
    // Subscribe for messages sent to the topic "pong"

            

        

            

                15
            

            

                1
            

            

                
    subscriber.subscribe_to(&"pong")?;

            

        

            

                16
            

            

                
            

            

                

            

        

            

                17
            

            

                
            

            

                
    // Launch a thread that sends out "ping" messages.

            

        

            

                18
            

            

                1
            

            

                
    let thread_db = db.clone();

            

        

            

                19
            

            

                1
            

            

                
    std::thread::spawn(move || pinger(thread_db));

            

        

            

                20
            

            

                1
            

            

                
    // Launch a thread that receives "ping" messages and sends "pong" responses.

            

        

            

                21
            

            

                1
            

            

                
    std::thread::spawn(move || ponger(db));

            

        

            

                22
            

            

                
            

            

                

            

        

            

                23
            

            

                
            

            

                
    // Loop until a we receive a message letting us know when the ponger() has

            

        

            

                24
            

            

                
            

            

                
    // no pings remaining.

            

        

            

                25
            

            

                
            

            

                
    loop {

            

        

            

                26
            

            

                5
            

            

                
        let message = subscriber.receiver().receive()?;

            

        

            

                27
            

            

                5
            

            

                
        let pings_remaining = message.payload::<usize>()?;

            

        

            

                28
            

            

                5
            

            

                
        println!(

            

        

            

                29
            

            

                5
            

            

                
            "<-- Received {}, pings remaining: {}",

            

        

            

                30
            

            

                5
            

            

                
            message.topic::<String>()?,

            

        

            

                31
            

            

                
            

            

                
            pings_remaining

            

        

            

                32
            

            

                
            

            

                
        );

            

        

            

                33
            

            

                5
            

            

                
        if pings_remaining == 0 {

            

        

            

                34
            

            

                1
            

            

                
            break;

            

        

            

                35
            

            

                4
            

            

                
        }

            

        

            

                36
            

            

                
            

            

                
    }

            

        

            

                37
            

            

                
            

            

                

            

        

            

                38
            

            

                1
            

            

                
    println!("Received all pongs.");

            

        

            

                39
            

            

                1
            

            

                

            

        

            

                40
            

            

                1
            

            

                
    Ok(())

            

        

            

                41
            

            

                1
            

            

                
}

            

        

            

                42
            

            

                
            

            

                

            

        

            

                43
            

            

                1
            

            

                
fn pinger<P: PubSub>(pubsub: P) -> Result<(), bonsaidb::local::Error> {

            

        

            

                44
            

            

                1
            

            

                
    let mut ping_count = 0u32;

            

        

            

                45
            

            

                
            

            

                
    loop {

            

        

            

                46
            

            

                10
            

            

                
        ping_count += 1;

            

        

            

                47
            

            

                10
            

            

                
        println!("-> Sending ping {ping_count}");

            

        

            

                48
            

            

                10
            

            

                
        pubsub.publish(&"ping", &ping_count)?;

            

        

            

                49
            

            

                9
            

            

                
        sleep(Duration::from_millis(250));

            

        

            

                50
            

            

                
            

            

                
    }

            

        

            

                51
            

            

                1
            

            

                
}

            

        

            

                52
            

            

                
            

            

                

            

        

            

                53
            

            

                1
            

            

                
fn ponger<P: PubSub>(pubsub: P) -> Result<(), bonsaidb::local::Error> {

            

        

            

                54
            

            

                
            

            

                
    const NUMBER_OF_PONGS: usize = 5;

            

        

            

                55
            

            

                1
            

            

                
    let subscriber = pubsub.create_subscriber()?;

            

        

            

                56
            

            

                1
            

            

                
    subscriber.subscribe_to(&"ping")?;

            

        

            

                57
            

            

                1
            

            

                
    let mut pings_remaining = NUMBER_OF_PONGS;

            

        

            

                58
            

            

                1
            

            

                

            

        

            

                59
            

            

                1
            

            

                
    println!("Ponger started, waiting to respond to {pings_remaining} pings");

            

        

            

                60
            

            

                
            

            

                

            

        

            

                61
            

            

                6
            

            

                
    while pings_remaining > 0 {

            

        

            

                62
            

            

                5
            

            

                
        let message = subscriber.receiver().receive()?;

            

        

            

                63
            

            

                5
            

            

                
        println!(

            

        

            

                64
            

            

                5
            

            

                
            "<- Received {}, id {}",

            

        

            

                65
            

            

                5
            

            

                
            message.topic::<&str>().unwrap(),

            

        

            

                66
            

            

                5
            

            

                
            message.payload::<u32>().unwrap()

            

        

            

                67
            

            

                5
            

            

                
        );

            

        

            

                68
            

            

                5
            

            

                
        pings_remaining -= 1;

            

        

            

                69
            

            

                5
            

            

                
        pubsub.publish(&"pong", &pings_remaining)?;

            

        

            

                70
            

            

                
            

            

                
    }

            

        

            

                71
            

            

                
            

            

                

            

        

            

                72
            

            

                1
            

            

                
    println!("Ponger finished.");

            

        

            

                73
            

            

                1
            

            

                

            

        

            

                74
            

            

                1
            

            

                
    Ok(())

            

        

            

                75
            

            

                1
            

            

                
}

            

        

            

                76
            

            

                
            

            

                

            

        

            

                77
            

            

                1
            

            

                
#[test]

            

        

            

                78
            

            

                1
            

            

                
fn runs() {

            

        

            

                79
            

            

                1
            

            

                
    main().unwrap()

            

        

            

                80
            

            

                1
            

            

                
}