FlowController Use Cases
[Programming How-To's]

Working with flow controllers. More...

Creating a flow controller

Set up participant

Create a flow controller

    DDS_ReturnCode_t retcode;
    DDSFlowController *controller = NULL;
    DDS_FlowControllerProperty_t property;
    
    retcode = participant->get_default_flowcontroller_property(property);
    
    if (retcode != DDS_RETCODE_OK) {
        printf("***Error: failed to get default flow controller property\n");
    }
    
    // optionally modify flow controller property values
    
    controller = participant->create_flowcontroller(
        "my flow controller name", property);
    
    if (controller == NULL) {
        printf("***Error: failed to create flow controller\n");
    }

Flow controlling a data writer

Set up participant

Create flow controller

Create an asynchronous data writer, FooDataWriter, of user data type Foo:

    DDS_DataWriterQos writer_qos;
    DDS_ReturnCode_t retcode;

    // MyWriterListener is user defined and 
    // extends DDSDataWriterListener
    MyWriterListener* writer_listener = new MyWriterListener(); // or = NULL

    retcode = publisher->get_default_datawriter_qos(writer_qos);

    if (retcode != DDS_RETCODE_OK) {
        // ... error 
    }

    /* Change the writer QoS to publish asnychronously */
    writer_qos.publish_mode.kind = DDS_ASYNCHRONOUS_PUBLISH_MODE_QOS;

    /* Setup to use the previously created flow controller */
    writer_qos.publish_mode.flow_controller_name = 
        DDS_String_dup("my flow controller name");

    /* Samples queued for asynchronous write are subject to the History Qos policy */
    writer_qos.history.kind = DDS_KEEP_ALL_HISTORY_QOS;

    FooDataWriter* writer = publisher->create_datawriter(topic,
                                                     writer_qos, 
                                                     writer_listener,
                             DDS_STATUS_MASK_ALL);
    if (writer == NULL) {
        // ... error 
    };

    /* Send data asynchronously... */
    
    /* Wait for asynchronous send completes, if desired */
    retcode = writer->wait_for_asynchronous_publishing(timout);

    if (retcode != DDS_RETCODE_OK) {
        printf("***Error: failed to wait for asynchronous publishing\n");
    }

Using the built-in flow controllers

RTI Data Distribution Service provides several built-in flow controllers.

The DDS_DEFAULT_FLOW_CONTROLLER_NAME built-in flow controller provides the basic asynchronous writer behavior. When calling FooDataWriter::write, the call signals the DDSPublisher asynchronous publishing thread (DDS_PublisherQos::asynchronous_publisher) to send the actual data. As with any DDS_ASYNCHRONOUS_PUBLISH_MODE_QOS DDSDataWriter, the FooDataWriter::write call returns immediately afterwards. The data is sent immediately in the context of the DDSPublisher asynchronous publishing thread.

When using the DDS_FIXED_RATE_FLOW_CONTROLLER_NAME flow controller, data is also sent in the context of the DDSPublisher asynchronous publishing thread, but at a regular fixed interval. The thread accumulates samples from different DDSDataWriter instances and generates data on the wire only once per DDS_FlowControllerTokenBucketProperty_t::period.

In contrast, the DDS_ON_DEMAND_FLOW_CONTROLLER_NAME flow controller permits flow only when DDSFlowController::trigger_flow is called. The data is still sent in the context of the DDSPublisher asynchronous publishing thread. The thread accumulates samples from different DDSDataWriter instances (across any DDSPublisher) and sends all data since the previous trigger.

The properties of the built-in DDSFlowController instances can be adjusted.

Set up participant

Lookup built-in flow controller

    DDSFlowController *controller = NULL;

    controller = participant->lookup_flowcontroller(
        DDS_DEFAULT_FLOW_CONTROLLER_NAME);

    /* This should never happen, built-in flow controllers are always created */
    if (controller == NULL) {
        printf("***Error: failed to lookup flow controller\n");
    }

Change property of built-in flow controller, if desired

    DDS_ReturnCode_t retcode;
    DDS_FlowControllerProperty_t property;

    /* Get the property of the flow controller */
    retcode = controller->get_property(property);

    if (retcode != DDS_RETCODE_OK) {
        printf("***Error: failed to get flow controller property\n");
    }

    /* Change the property value as desired */
    property.token_bucket.period.sec = 2;
    property.token_bucket.period.nanosec = 0;

    /* Update the flow controller property */
    retcode = controller->set_property(property);

    if (retcode != DDS_RETCODE_OK) {
        printf("***Error: failed to set flow controller property\n");
    }

Create a data writer using the correct flow controller name

Shaping the network traffic for a particular transport

Set up participant

Create the transports

Create a separate flow controller for each transport

Configure DDSDataWriter instances to only use a single transport

Associate all data writers using the same transport to the corresponding flow controller

For each transport, the corresponding flow controller limits the network traffic based on the token bucket properties

Coalescing multiple samples in a single network packet

Set up participant

Create a flow controller with a desired token bucket period

Associate the data writer with the flow controller