RustPubSub

Rust Pub/Sub - Example 16

This example is a reimplementation of example 2 in Rust.

The Publisher (RustProtobufDataProductPublisher)

The Rust publisher looks somewhat similar to the C++ version:

let gn = GravityNode::new();
gn.init("ProtobufGravityComponentID");
  
gn.register_data_product(
    //identifies the data product to the service directory so others 
    //can subscribe to it
    "BasicCounterDataProduct",
    //Assign a transport type to the socket (almost always tcp, unless you are only
    //using the gravity data product between two processes on the same computer)
     GravityTransportType::TCP);    

let mut quit = false; 
let mut count = 1;
while !quit {   
    //create a data product to send across the network of type "BasicCounterDataProduct"
    let counter_data_product = GravityDataProduct::with_id("BasicCounterDataProduct"); 

    //Initialize our message
    let mut counter_data_pb = BasicCounterDataProductPB::new();
    counter_data_pb.set_count(count);
    
    //Put message into data product
    counter_data_product.set_data(&counter_data_pb);
    
    //Publish the data product
    gn.publish(&counter_data_product);

    //Increment count
    count += 1;
    if count > 50 { count = 1; }

    //Sleep for 1 second.
    std::thread::sleep(time::Duration::from_secs(1));
}

gn.wait_for_exit();  

The only big difference from C++ is how you create a gravity type, since there are no classes in Rust, only structs. The default constructor is called by ::new() and ones that take a parameter are called by ::with_().

The Subscriber (RustProtobufDataProductSubscriber)

As in the C++ version, we first need to define a subscription handler class: Similar to how we define a subscription handler class in C++, we need to define a struct that impements GravitySubscriber methods.

struct MySubscriber {}

impl GravitySubscriber for MySubscriber {
    
    fn subscription_filled(&mut self, data_products: Vec<GravityDataProduct>) {
        
        for data_product in data_products.iter() {
            //Get the protobuf object from the message
            let mut counter_data_pb = BasicCounterDataProductPB::new();
            data_product.populate_message(&mut counter_data_pb)

            //Process the message
            SpdLog::warn(format!("Current count: {}", counter_data_pb.count()));    
        }
    }
}

And then to use the subscriber:

    let mut gn = GravityNode::new();
    gn.init("ProtobufDataProductSubscriber");

    //This is just an example, you can have any other way to 
    //instantiate your own GravitySubscriber, as long as it impl GravitySubscriber trait
    // You must tokenize the subscriber with the GravityNode that will be subscribing with it
    // Gives the GravityNode ownership and control of the subscriber
    let subscriber = gn.tokenize_subscriber(MySubscriber {});

    //subscribe to the data product
    gn.subscribe("BasicCounterDataProduct", &subscriber);

    //Wait for us to exit (Ctrl-C or be killed)
    gn.wait_for_exit()

    gn.unsubscribe("BasicCounterDataProduct", &subscriber);

It is still pretty similar to the C++ version, but with structs and impl instead of classes and inheritance. The GravitySubscriber trait must be implemented for a struct when you pass it into the subscribe function.

A big difference is the tokenize part. Since something has to own the subscriber, it is best if the GravityNode holds it since it will be direclty accessing it. Hopefully soon shared ownership (with Arcs) will be implemented for this API.