Publish/Subscribe

Table of Contents

• Publishers
• Subscribers
• Access Control

RCF provides a built in publish/subscribe implementation.

Publish/subscribe is a messaging paradigm where publishers send out messsages to groups of subscribers. Rather than publishers being aware of individual subscribers and sending messages directly to them, publishers instead categorize their messages into topics, and subscribers choose which topics to receive messages from. When a publisher publishes a message on a particular topic, all subscribers subscribing to that topic will receive the message.

Publish/subscribe can be used in the presence of firewalls and NAT's. The only network topology requirement is that the subscriber must be able to initiate a network connection to the publisher. Publishers will never attempt to establish network connections to their subscribers.

Publishers

To create a publisher, use RCF::RcfServer::createPublisher<>():

    RCF::RcfServer pubServer( RCF::TcpEndpoint(50001) );
    pubServer.start();

    typedef RCF::Publisher<I_PrintService> PrintServicePublisher;
    typedef std::shared_ptr< PrintServicePublisher > PrintServicePublisherPtr;
    PrintServicePublisherPtr publisherPtr = pubServer.createPublisher<I_PrintService>();

This will create a publisher with a default topic name. The default topic name is the runtime name of the RCF interface passed to RCF::RcfServer::createPublisher<>() (in this case "I_PrintService").

The topic name can also be set explicitly. For example, to create two publishers, with different topic names, using the same RCF interface:

    RCF::PublisherParms pubParms;
    pubParms.setTopicName("PrintService_Topic_1");
    PrintServicePublisherPtr publisher1Ptr = pubServer.createPublisher<I_PrintService>(pubParms);

    pubParms.setTopicName("PrintService_Topic_2");
    PrintServicePublisherPtr publisher2Ptr = pubServer.createPublisher<I_PrintService>(pubParms);

The RCF::Publisher<> object returned by RCF::RcfServer::createPublisher<>() is used to publish remote calls. Published remote calls always have one-way semantics, and are received by all subscribers currently subscribing to that publishing topic. To publish a call, use RCF::Publisher<>::publish():

    publisherPtr->publish().Print("First published message.");
    publisherPtr->publish().Print("Second published message.");

The publishing topic is closed, and all of its subscribers disconnected, when the RCF::Publisher<> object is destroyed, or when RCF::Publisher<>::close() is called.

    // Close the publishing topic. All subscribers will be disconnected.
    publisherPtr->close();

Subscribers

To subscribe to a publisher, use RCF::RcfServer::createSubscription<>():

    RCF::RcfServer subServer( RCF::TcpEndpoint(-1) );
    subServer.start();

    PrintService printService;
    RCF::SubscriptionParms subParms;
    subParms.setPublisherEndpoint( RCF::TcpEndpoint(50001) );

    RCF::SubscriptionPtr subscriptionPtr = subServer.createSubscription<I_PrintService>(
        printService, 
        subParms);

As for publishers, the topic name defaults to the runtime name of the RCF interface (in this case "I_PrintService"). The topic name can also be specified explicitly:

        PrintService printService;

        RCF::SubscriptionParms subParms;
        subParms.setPublisherEndpoint( RCF::TcpEndpoint(50001) );
        subParms.setTopicName("PrintService_Topic1");

        RCF::SubscriptionPtr subscription1Ptr = subServer.createSubscription<I_PrintService>(
            printService, 
            subParms);

        subParms.setTopicName("PrintService_Topic2");

        RCF::SubscriptionPtr subscription2Ptr = subServer.createSubscription<I_PrintService>(
            printService,
            subParms);

The first parameter passed to RCF::RcfServer::createSubscription<>() is the object which will receive the published messages. It is the applications responsibility to make sure this object is not destroyed while the subscription is still connected.

A disconnect callback can be provided, which will be called if the subscriber is disconnected:

void onSubscriptionDisconnected(RCF::RcfSession & session)
{
    // Handle subscription disconnection here.
    // ...
}

        PrintService printService;

        RCF::SubscriptionParms subParms;
        subParms.setPublisherEndpoint( RCF::TcpEndpoint(50001) );
        subParms.setOnSubscriptionDisconnect(&onSubscriptionDisconnected);

        RCF::SubscriptionPtr subscriptionPtr = subServer.createSubscription<I_PrintService>(
            printService, 
            subParms);

To terminate a subscription, destroy the Subscription object, or call Subscription::close():

        subscriptionPtr->close();

Access Control

Access controls can be applied to publishers, in the form of an access control callback which will be called on the publishing server, for each subscriber attempting to set up a subscription. Similarly to servant binding access controls, publisher access controls can be used to inspect the RCF::RcfSession of the subscriber connections for any relevant authentication information:

bool onSubscriberConnect(RCF::RcfSession & session, const std::string & topicName)
{
    // Return true to allow access, false otherwise.
    // ...

    return true;
}

void onSubscriberDisconnect(RCF::RcfSession & session, const std::string & topicName)
{
    // ...
}

        RCF::PublisherParms pubParms;
        pubParms.setOnSubscriberConnect(onSubscriberConnect);
        pubParms.setOnSubscriberDisconnect(onSubscriberDisconnect);
        PrintServicePublisherPtr publisher1Ptr = pubServer.createPublisher<I_PrintService>(pubParms);