Using TAO and OpenDDS with .NET, Part II

Using TAO and OpenDDS with .NET, Part II

by Charles Calkins, Senior Software Engineer

February 2009

Introduction

For several years, OCI has been engaged with a customer in the maintenance of a legacy data-acquisition application. Data is collected by remote sensing devices and stored in a database, and the sensing devices are managed, and the data viewed, by an application written for Microsoft Windows.

Although originally a single-user application referencing a local database, over time the application has evolved into one where multiple users can simultaneously connect to a single centralized database. If one user makes a change to the database, all other connected users must be made aware of the change so their local states can be updated.

A solution to this problem is to create a single process to manage access to the database, and to provide database change notifications to interested client applications. In part I of this article, we described the architecture of DataServer, an application written in a combination of C++ and C#, that manages access to a database, and interacts with client applications via the use of CORBA for control, such as adding, updating, and deleting database records, and DDS for client notification. The Object Computing, Inc. distributions of TAO and OpenDDS were used as the CORBA and DDS implementations, respectively.

As described in part I, TAO and OpenDDS were selected as the middleware to accomplish this task for two main reasons. The first is that it is easy to be up and running quickly — the prototype that was developed to illustrate the architecture was completed in under three days. The second is that, as both TAO and OpenDDS are Open Source, there are no license fees or other costs involved. The resulting application can be deployed widely without incurring a per-seat, or per-CPU, or even a development/SDK charge.

To review, the architecture is as follows, with components written in C# shown with box hatching and components written in C++ shown with angled hatching.

architecture tower

In this article, we will continue the development of DataServer by implementing the database notifications with OpenDDS. While this article will describe features of OpenDDS that are useful for our purpose, please see this article for more detailed information on OpenDDS itself.

As our objective is to notify all clients whenever any client performs a database operation, DataServer must be able to generate notifications as needed, and clients need to be able to receive them. The distributed publish and subscribe architecture of DDS is exactly what we need.


SIDEBAR
The code in this article was developed with Microsoft Visual Studio 2005. It was compiled against TAO version 1.6aOpenDDS version 1.3, and MPC version 3.7.2. Inline assembly was disabled to prevent the .NET-related compiler warning C4793, as the use of __asm forces native code generation. Wide character support was enabled, as .NET uses Unicode for string representation. The build settings for these features are as follows:

  1. // add to %ACE_ROOT%\ace\config.h
  2. #define ACE_LACKS_INLINE_ASSEMBLY 1
  3. #define ACE_USES_WCHAR 1
  4.  
  5. // add to %ACE_ROOT%\bin\MakeProjectCreator\config\default.features
  6. uses_wchar=1

DDS/OpenDDS Overview

We shall begin by reviewing the elements of DDS, and, where applicable, areas where OpenDDS provides more specialized functionality. OpenDDS supports the Data-Centric Publish and Subscribe (DCPS) layer of the DDS specification, so it is that layer of DDS that we will be concerned with in this article.

Processes that wish to exchange messages via DDS are part of one or more domains, and are known as domain participants. A domain is a global data space, identified by an integer. While a given process can join multiple domains, messages sent in one domain can only be received by other participants in that same domain.

publisher is responsible for distributing data in a domain. It may publish data of different types, with each type published by adata writer associated with that type. Data is received by a subscriber, with each data type received by a type-specific data readerQuality of service (QoS) policies exist at various levels to manage the data transfer process. Entities in a domain such as these may also have associated listeners where events can be received asynchronously when states are changed, data is received, or the like.

The data that is transferred between publishers and subscribers is represented by a topic. A topic has a name which is unique in the domain, a data type that is expressed in IDL, and quality of service policies associated with the data. A sample is a particular data element of a topic, and a sample may have one or more fields defined as a key. An instance is a set of samples which have the same key.

OpenDDS allows samples to be transmitted using various protocols, including: TCP (SimpleTcp), UDP (SimpleUdp), unreliable multicast (SimpleMcast), and reliable multicast (ReliableMulticast). This is known as a transport.

In this article, we will develop a database notification topic, add the ability for DataServer to publish samples of that topic, and for the client application to subscribe to samples of that topic. We will build upon the code developed for part I, focusing on OpenDDS integration, so please review the earlier article for the details of using TAO in DataServer. Also, while relevant code segments will be highlighted, please see the code archive that accompanies this article for the full details.

Database Notification

We wish to notify all clients of all operations that are performed on the database by other clients — reads, writes, updates and deletes of records. Although, for purposes of demonstration, DataServer manages only one table named Items, we will include the table name in the notification for completeness, in addition to the ID of the row that was acted upon.

For use by DDS, the database notification must be expressed in IDL. We create a directory, DatabaseNotification off of the root, and add a file, DatabaseNotification.idl, with the following contents:

  1. // DatabaseNotification.idl
  2. enum NotificationType
  3. {
  4. NOTIFICATION_CREATED,
  5. NOTIFICATION_READ,
  6. NOTIFICATION_UPDATED,
  7. NOTIFICATION_DELETED
  8. };
  9.  
  10. #pragma DCPS_DATA_TYPE "DatabaseNotification"
  11. #pragma DCPS_DATA_KEY "DatabaseNotification id"
  12. #pragma DCPS_DATA_KEY "DatabaseNotification table_name"
  13. struct DatabaseNotification
  14. {
  15. long long id;
  16. string table_name;
  17. NotificationType notification_type;
  18. };

The DatabaseNotification structure represents the notification itself, containing both a 64-bit table row ID, the table name with which the ID is associated, and referencing an enumeration which specifies the various types of actions that can occur. Additional information could also be provided, such as a unique identifier representing the client that performed the action, but, for now, the above will suffice.

In order for a structure expressed in IDL to be used as an OpenDDS type, the structure must be identified by the DCPS_DATA_TYPE pragma. A key can be defined by the DCPS_DATA_KEY pragma, where both the containing structure, and the field name, are provided. Here, the key is composed of two fields: id and table_name. A key is not required, but in this example, we use keys to identify each unique combination of table_name and id as a separate instance.

To compile the IDL definition, we next create an MPC file in the same directory named DatabaseNotification.mpc

  1. // DatabaseNotification.mpc
  2. project : dcps, CPPBase {
  3. sharedname = DatabaseNotificationCommon
  4. dynamicflags = DATABASENOTIFICATIONCOMMON_BUILD_DLL
  5. requires += tao_orbscvs
  6. includes += $(TAO_ROOT)/orbsvcs
  7. idlflags += -I$(TAO_ROOT)/orbsvcs
  8. idlflags += -Wb,export_macro=DatabaseNotificationCommon_Export
  9. idlflags += \
  10. -Wb,export_include=DatabaseNotificationCommon_Export.h
  11. dcps_ts_flags += --export=DatabaseNotificationCommon_Export
  12.  
  13. TypeSupport_Files {
  14. DatabaseNotification.idl >> \
  15. DatabaseNotificationTypeSupport.idl \
  16. DatabaseNotificationTypeSupportImpl.h \
  17. DatabaseNotificationTypeSupportImpl.cpp
  18. }
  19.  
  20. IDL_Files {
  21. DatabaseNotificationTypeSupport.idl
  22. DatabaseNotification.idl
  23. }
  24.  
  25. Header_Files {
  26. DatabaseNotificationTypeSupportImpl.h
  27. }
  28.  
  29. Source_Files {
  30. DatabaseNotificationTypeSupportImpl.cpp
  31. }
  32. }

This project inherits from the dcps base project, which provides many definitions needed for compiling with OpenDDS. The result of compiling the IDL is a C++ library, DatabaseNotificationCommon. The project also inherits from CPPBase, as defined in part I, to set output directories, include paths, and other properties.

OpenDDS uses elements of TAO, so a dependence on tao_orbscvs is needed. Macros must be defined to manage the declaration and exporting of generated symbols, with the primary macro set to the name of the project, DatabaseNotificationCommon, suffixed with _Export. The definition of this and other macros is in a header file with the same name and .h extension. This command is used to create that file (typed on one line):

generate_export_file.pl DatabaseNotificationCommon > DatabaseNotificationCommon_Export.h

The remaining sections of the MPC file describe the files that will be compiled into the library. Note that the >> indicates that the result of processing one file (here, running the tao_idl compiler on DatabaseNotification.idl) results in additional files being generated which themselves must be processed to produce the library.

Now that the MPC file is complete, it must be added to the workspace so it can be compiled as part of the project. After adding DatabaseNotification to the DataServer.mwc file, the result is as follows:

  1. // DataServer.mwc
  2. workspace {
  3. specific {
  4. cmdline += -language csharp
  5. DataLib
  6. Client
  7. }
  8. DataServer
  9. DataServerConnectorLib
  10. IDL
  11. DatabaseNotification
  12. }

Publisher

With the definition of the database notification complete, we can add the ability to DataServer to publish notifications when database operations are performed. We will update DataServer to be a DDS publisher by adding code to create the relevant DDS entities, and to publish samples when database operations are successfully executed by extending the servant that had been created in part I.

We start with defining a few constants, for convenience, to represent the domain, as well as the notification type and topic names. In DataServer.cpp, we add the following before ACE_TMAIN():

  1. DDS::DomainId_t DATABASENOTIFICATION_DOMAIN_ID = 1066;
  2. const char* DATABASENOTIFICATION_TYPE =
  3. "DatabaseNotification Type";
  4. const char* DATABASENOTIFICATION_TOPIC =
  5. "DatabaseNotification Topic";

At the start of ACE_TMAIN(), we define two variables that represent fundamental DDS elements:

  1. int ACE_TMAIN(int argc, ACE_TCHAR *argv[]) {
  2. DDS::DomainParticipantFactory_var dpf =
  3. DDS::DomainParticipantFactory::_nil();
  4. DDS::DomainParticipant_var participant =
  5. DDS::DomainParticipant::_nil();

The DomainParticipantFactory allows creation of DomainParticipant objects, while the DomainParticipant acts as a container of, and a factory for, DDS entities within a specific domain.

When mixing TAO and OpenDDS in the same application, it is necessary to create a separate argc/argv set for OpenDDS initialization, for two reasons. The first is that the initialization of OpenDDS (or TAO) will cause command-line arguments to be consumed, so necessary but already consumed arguments will be unavailable for the initialization of the other. The second is that it is important to ensure that arguments used by TAO and OpenDDS do not conflict. For instance, DataServer would typically be started with an argument such as the following on the command line to specify a listening port:

-ORBListenEndpoints iiop://:12346

Passing this same argument to both TAO and OpenDDS results in an error as both would attempt to listen on the same port, so a CORBA BAD_PARAM exception will be thrown. As we want that option to be used only for the initialization of TAO, and not of OpenDDS, we must create a new argc/argv set to be used for OpenDDS which does not contain -ORBListenEndpoints.

  1. try {
  2. // must duplicate argc/argv as each ORB absorbs parameters
  3. int argcDDS = 0;
  4. ACE_TCHAR **argvDDS=new ACE_TCHAR *[argc];
  5. for (int i=0; i<argc; i++)
  6. // do not pass -ORBListenEndpoints and its argument
  7. if (!ACE_OS::strcmp(
  8. ACE_TEXT("-ORBListenEndpoints"),
  9. argv[i]))
  10. i++;
  11. else
  12. argvDDS[argcDDS++] = ACE_OS::strdup(argv[i]);

We can now initialize the DomainParticipantFactory with the new argc/argv set.

  1. dpf = TheParticipantFactoryWithArgs(argcDDS, argvDDS);

TheParticipantFactoryWithArgs() is a macro that expands to:

TheServiceParticipant->get_domain_participant_factory(argc, argv)

where TheServiceParticipant is a singleton which allows applications to configure OpenDDS, as well as providing other management features. For instance, calling TheServiceParticipant->shutdown() will terminate OpenDDS processing.

Returning to DataServer, we can create the DomainParticipant for our domain. DataServer does not need to participate in more than one domain, so only one DomainParticipant for the DATABASENOTIFICATION_DOMAIN_ID domain is needed. The default quality of service policy is sufficient for the DomainParticipant, so the PARTICIPANT_QOS_DEFAULT constant is specified. For a list of what QoS policies are applied as a default, please refer to Appendix A in this article.

DataServer does not need to monitor any events associated with the DomainParticipant, so no listener is needed.

  1. participant = dpf->create_participant(
  2. DATABASENOTIFICATION_DOMAIN_ID,
  3. PARTICIPANT_QOS_DEFAULT,
  4. DDS::DomainParticipantListener::_nil());
  5. if (CORBA::is_nil(participant.in()))
  6. throw std::exception("create_participant failed");

We must now initialize the transport. Please refer to Appendix B for a discussion on various ways that this can be done. For flexibility, we will choose to have the transport specified via configuration files, so we add the following to DataServer.cpp:

  1. const OpenDDS::DCPS::TransportIdType TRANSPORT_IMPL_ID = 1;
  2.  
  3. OpenDDS::DCPS::TransportImpl_rch trans_impl =
  4. TheTransportFactory->create_transport_impl(
  5. TRANSPORT_IMPL_ID,
  6. OpenDDS::DCPS::AUTO_CONFIG);

We now create a Publisher, using the DomainParticipant as a factory. As before, we do not need a listener, and the default QoS is sufficient.

  1. DDS::Publisher_var pub = participant->create_publisher(
  2. PUBLISHER_QOS_DEFAULT,
  3. DDS::PublisherListener::_nil());
  4. if (CORBA::is_nil(pub.in()))
  5. throw std::exception("create_publisher failed");

With the transport defined, we can attach the Publisher to the transport.

 
  1. OpenDDS::DCPS::AttachStatus status =
  2. trans_impl->attach(pub.in());
  3. if (status != OpenDDS::DCPS::ATTACH_OK) {
  4. std::string msg("Cannot attach to the transport: ");
  5. switch (status) {
  6. case OpenDDS::DCPS::ATTACH_BAD_TRANSPORT:
  7. throw std::exception(
  8. (msg+"ATTACH_BAD_TRANSPORT").c_str());
  9. case OpenDDS::DCPS::ATTACH_ERROR:
  10. throw std::exception(
  11. (msg+"ATTACH_ERROR").c_str());
  12. case OpenDDS::DCPS::ATTACH_INCOMPATIBLE_QOS:
  13. throw std::exception(
  14. (msg+"ATTACH_INCOMPATIBLE_QOS").c_str());
  15. default:
  16. throw std::exception(
  17. (msg+"unknown status").c_str());
  18. }
  19. }

Next, we register the type. This informs the middleware that the database notification type exists, and allows it to manage data of that type.

  1. DatabaseNotificationTypeSupport_var
  2. databaseNotification_servant
  3. = new DatabaseNotificationTypeSupportImpl();
  4. if (DDS::RETCODE_OK !=
  5. databaseNotification_servant->register_type(
  6. participant.in(),
  7. DATABASENOTIFICATION_TYPE))
  8. throw std::exception("register_type failed");

As the data type itself has been registered, we now inform the middleware of the topic. A topic is comprised of a data type, a topic name, QoS, and optionally, a listener. A listener is still not needed, and once again the default QoS is sufficient. Although the constant TOPIC_QOS_DEFAULT could be used, for demonstration purposes we choose to retrieve the default QoS into a structure via get_default_topic_qos() and apply that structure as the QoS.

  1. DDS::TopicQos default_topic_qos;
  2. participant->get_default_topic_qos(default_topic_qos);
  3.  
  4. DDS::Topic_var databaseNotification_topic =
  5. participant->create_topic(
  6. DATABASENOTIFICATION_TOPIC,
  7. DATABASENOTIFICATION_TYPE,
  8. default_topic_qos,
  9. DDS::TopicListener::_nil());
  10. if (CORBA::is_nil(databaseNotification_topic.in()))
  11. throw std::exception("create_topic failed");

Next, we create a data writer for the database notification. At this point, we still do not require more than the default QoS, and no listener is needed.

  1. DDS::DataWriterQos dw_default_qos;
  2. pub->get_default_datawriter_qos(dw_default_qos);
  3.  
  4. DDS::DataWriter_var databaseNotification_base_dw =
  5. pub->create_datawriter(databaseNotification_topic.in(),
  6. dw_default_qos,
  7. DDS::DataWriterListener::_nil());
  8. if (CORBA::is_nil(databaseNotification_base_dw.in()))
  9. throw std::exception("create_datawriter failed");
  10.  
  11. DatabaseNotificationDataWriter_var databaseNotification_dw
  12. = DatabaseNotificationDataWriter::_narrow(
  13. databaseNotification_base_dw.in());
  14. if (CORBA::is_nil(databaseNotification_dw.in()))
  15. throw std::exception(
  16. "DatabaseNotificationDataWriter could not be narrowed");

To make use of the data writer, we pass it to the servant which implements the CORBA Database interface that we developed in part I.

  1. Database_i servant(%database, databaseNotification_dw.in());

Finally, we must clean up the various DDS components that were allocated. delete_contained_entities() is called to free all entities that are associated with the participant, and the participant itself is freed with a call to delete_participant(). If any other DomainParticipants for other domains were allocated, they must also be freed at this point. When no others remain, the TransportFactory is freed and DDS processing is terminated by calling shutdown() on TheServiceParticipant.

  1. try {
  2. if (!CORBA::is_nil(participant.in()))
  3. participant->delete_contained_entities();
  4. if (!CORBA::is_nil(dpf.in()))
  5. dpf->delete_participant(participant.in());
  6. } catch (CORBA::Exception& e) {
  7. std::cerr << "Exception during cleanup: " << e << std::endl;
  8. }
  9. TheTransportFactory->release();
  10. TheServiceParticipant->shutdown();

This completes the changes to DataServer.cpp. We must now update the servant to make use of the data writer. We first add a member variable to store a pointer to the data writer, and update the constructor to set the variable.

  1. // Database_i.h
  2. class Database_i
  3. : public virtual POA_Database
  4. {
  5. gcroot<DataLib::Database^> database_;
  6. DatabaseNotificationDataWriter_ptr dataWriter_;
  7. ...
  8. public:
  9. Database_i(gcroot<DataLib::Database^> database,
  10. DatabaseNotificationDataWriter_ptr dataWriter);
  1. // Database_i.cpp
  2. Database_i::Database_i(gcroot<DataLib::Database^> database,
  3. DatabaseNotificationDataWriter_ptr dataWriter) :
  4. database_(database), dataWriter_(dataWriter)
  5. {
  6. }

Next, we add a helper method, WriteNotification(), which is called to publish the database notification itself. The DatabaseNotification structure is populated based on the method parameters, and then passed to write() of the DataWriter to be published as a DDS sample. For simplicity, a sample instance handle is not used, and the constant DDS::HANDLE_NILM is passed instead.

  1. // Database_i.cpp
  2. void Database_i::WriteNotification(::CORBA::LongLong id,
  3. ::CORBA::Char *tableName, NotificationType notificationType) {
  4. DatabaseNotification notification;
  5. notification.id = id;
  6. notification.table_name = ::CORBA::string_dup(tableName);
  7. notification.notification_type = notificationType;
  8.  
  9. int ret = dataWriter_->write(notification, DDS::HANDLE_NIL);
  10. if (ret != DDS::RETCODE_OK)
  11. ACE_ERROR ((LM_ERROR,
  12. ACE_TEXT("(%P|%t)ERROR: write returned %d.\n"), ret));
  13. }

Each database method is now updated to call WriteNotification(). For instance, we modify CreateItem() as follows — the other functions are updated in a similar way.

  1. ::CORBA::Boolean Database_i::CreateItem (
  2. const ::CORBA::WChar * description,
  3. ::CORBA::LongLong_out id)
  4. {
  5. System::String^ netDescription =
  6. gcnew System::String(description);
  7. ::CORBA::Boolean result =
  8. database_->CreateItem(netDescription, id);
  9. delete netDescription;
  10. if (result)
  11. WriteNotification(id, "Items", NOTIFICATION_CREATED);
  12. return result;
  13. }

To complete the changes to DataServer, we must update DataServer.mpc. Only two changes are necessary — the project must inherit from dcpsexe as this application uses elements of OpenDDS, and a dependency upon DatabaseNotification is needed, as DataServer must link to the DatabaseNotificationCommon library. The resulting file is as follows:

  1. // DataServer.mpc
  2. project : taoserver, dcpsexe, CPPBase, iortable {
  3. exename = DataServer
  4. after += IDL
  5. after += DataLib
  6. after += DatabaseNotification
  7.  
  8. includes += ../IDL
  9. Source_Files {
  10. Database_i.cpp
  11. DataServer.cpp
  12. ../IDL/DatabaseC.cpp
  13. ../IDL/DatabaseS.cpp
  14. }
  15. IDL_Files {
  16. }
  17.  
  18. managed = 1
  19. }

Subscriber

With the publisher complete, we can begin the subscriber. As we mixed DDS publisher code with the TAO server in DataServer.cpp, we mix DDS subscriber code with the TAO client inDataServerConnectorLib.cpp. As with the publisher, we again define constants representing the domain, notification type, and topic, and create variables for the DomainParticipantFactory and the DomainParticipant.

  1. DDS::DomainId_t DATABASENOTIFICATION_DOMAIN_ID = 1066;
  2. const char* DATABASENOTIFICATION_TYPE =
  3. "DatabaseNotification Type";
  4. const char* DATABASENOTIFICATION_TOPIC =
  5. "DatabaseNotification Topic";
  6.  
  7. void DataServerConnector::Run() {
  8. DDS::DomainParticipantFactory_var dpf =
  9. DDS::DomainParticipantFactory::_nil();
  10. DDS::DomainParticipant_var participant =
  11. DDS::DomainParticipant::_nil();

We must create separate argc/argv sets for OpenDDS and TAO initialization, but as this is the .NET world, it is done slightly differently. Command-line arguments in .NET are available as an array of .NET strings — these must be converted to standard argc/argv.

  1. int argc = 0, argc2 = 0;
  2. wchar_t **argv = NULL, **argv2 = NULL;
  3.  
  4. try {
  5. array<String^>^ arguments = Environment::GetCommandLineArgs();
  6. argc = arguments->Length;
  7. argc2 = argc;
  8. argv = new wchar_t *[argc];
  9. argv2 = new wchar_t *[argc2];
  10. for (int i=0; i<argc; i++) {
  11. pin_ptr<const wchar_t> arg = PtrToStringChars(arguments[i]);
  12. argv[i] = _wcsdup(arg);
  13. argv2[i] = _wcsdup(arg);
  14. }

The initialization of the participant factory, the participant, and the transport are identical to that of the publisher.

  1. dpf = TheParticipantFactoryWithArgs(argc, argv);
  2.  
  3. participant = dpf->create_participant(
  4. DATABASENOTIFICATION_DOMAIN_ID,
  5. PARTICIPANT_QOS_DEFAULT,
  6. DDS::DomainParticipantListener::_nil());
  7. if (CORBA::is_nil(participant.in()))
  8. throw std::exception("create_participant failed");
  9.  
  10. const OpenDDS::DCPS::TransportIdType TRANSPORT_IMPL_ID = 1;
  11. OpenDDS::DCPS::TransportImpl_rch trans_impl =
  12. TheTransportFactory->create_transport_impl(
  13. TRANSPORT_IMPL_ID,
  14. OpenDDS::DCPS::AUTO_CONFIG);

Next, instead of creating a publisher, we create a subscriber. No listener is needed and the default QoS is sufficient.

  1. DDS::Subscriber_var sub =
  2. participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT,
  3. DDS::SubscriberListener::_nil());
  4. if (CORBA::is_nil(sub.in()))
  5. throw std::exception("create_subscriber failed");

We attach the subscriber to the transport in the same way that the publisher was attached to the transport.

  1. OpenDDS::DCPS::AttachStatus status =
  2. trans_impl->attach(sub.in());
  3. if (status != OpenDDS::DCPS::ATTACH_OK) {
  4. std::string msg("Cannot attach to the transport: ");
  5. switch (status) {
  6. case OpenDDS::DCPS::ATTACH_BAD_TRANSPORT:
  7. throw std::exception(
  8. (msg+"ATTACH_BAD_TRANSPORT").c_str());
  9. case OpenDDS::DCPS::ATTACH_ERROR:
  10. throw std::exception(
  11. (msg+"ATTACH_ERROR").c_str());
  12. case OpenDDS::DCPS::ATTACH_INCOMPATIBLE_QOS:
  13. throw std::exception(
  14. (msg+"ATTACH_INCOMPATIBLE_QOS").c_str());
  15. default:
  16. throw std::exception(
  17. (msg+"unknown status").c_str());
  18. }
  19. }

We also register the type and topic in the same manner as the publisher.

  1. DatabaseNotificationTypeSupport_var
  2. databaseNotification_servant =
  3. new DatabaseNotificationTypeSupportImpl();
  4. if (DDS::RETCODE_OK !=
  5. databaseNotification_servant->register_type(
  6. participant.in(),
  7. DATABASENOTIFICATION_TYPE))
  8. throw std::exception("register_type failed");
  9.  
  10. DDS::TopicQos default_topic_qos;
  11. participant->get_default_topic_qos(default_topic_qos);
  12.  
  13. DDS::Topic_var databaseNotification_topic =
  14. participant->create_topic(
  15. DATABASENOTIFICATION_TOPIC,
  16. DATABASENOTIFICATION_TYPE,
  17. default_topic_qos, DDS::TopicListener::_nil());
  18. if (CORBA::is_nil(databaseNotification_topic.in()))
  19. throw std::exception("create_topic failed");

We do need a listener for the data reader, so we create an instance of its implementation, and we pass this as a parameter so the listener can interact with this class. The purpose of the listener is to notify the application when samples arrive indicating database changes have occurred. The next section will describe the listener further.

  1. DDS::DataReaderListener_var
  2. databaseNotification_listener(
  3. new DatabaseNotificationDataReaderListenerImpl(this));
  4. if (CORBA::is_nil(databaseNotification_listener.in()))
  5. throw std::exception("cannot create the listener");

We create a data reader in a manner similar to how we had created the data writer, although this time we pass a listener reference to create_datareader().

  1. DDS::DataReaderQos dr_default_qos;
  2. sub->get_default_datareader_qos(dr_default_qos);
  3.  
  4. DDS::DataReader_var databaseNotification_dr =
  5. sub->create_datareader(databaseNotification_topic.in(),
  6. dr_default_qos,
  7. databaseNotification_listener.in());
  8. if (CORBA::is_nil(databaseNotification_dr.in()))
  9. throw std::exception("create_datareader failed");

At the end of DataConnector::Run(), we add the same cleanup code as we did at the end of ACE_TMAIN(), though we format the exception as a DataConnectorException for processing by the client.

  1. try {
  2. if (!CORBA::is_nil(participant.in()))
  3. participant->delete_contained_entities();
  4. if (!CORBA::is_nil(dpf.in()))
  5. dpf->delete_participant(participant.in());
  6. } catch (CORBA::Exception& e) {
  7. std::stringstream ss;
  8. ss << "Exception during cleanup: " << e;
  9. throw gcnew
  10. DataConnectorException(gcnew String(ss.str().c_str()));
  11. }
  12. TheTransportFactory->release();
  13. TheServiceParticipant->shutdown();

DatabaseNotification Listener

As samples must be processed when they arrive in order to pass the database notification to the user of DataServerConnector, a listener for the DataReader is needed. In the DataServerConnector project, create a file, DatabaseNotificationDataReaderListener.h, to implement a subclass of DDS::DataReaderListener. We store a reference to the DataServerConnector as a member variable for later use. As DataServerConnector is a .NET type and DatabaseNotificationDataReaderListenerImpl is not, gcroot<> is needed.

  1. // DatabaseNotificationDataReaderListener.h
  2. class DatabaseNotificationDataReaderListenerImpl
  3. : public virtual OpenDDS::DCPS::LocalObject<DDS::DataReaderListener>
  4. {
  5. gcroot<DataServerConnector^> dataConnector_;
  6. public:
  7. DatabaseNotificationDataReaderListenerImpl(
  8. gcroot<DataServerConnector^> dataConnector) :
  9. dataConnector_(dataConnector) {}

Only one listener method needs to be implemented, as only the arrival of data samples is of interest. We declare on_data_available() to be implemented in the CPP file, but provide empty bodies for the other methods.

  1. virtual ~DatabaseNotificationDataReaderListenerImpl() {}
  2.  
  3. virtual void on_requested_deadline_missed(
  4. DDS::DataReader_ptr /*reader*/,
  5. const DDS::RequestedDeadlineMissedStatus & /*status*/)
  6. throw (CORBA::SystemException) {}
  7.  
  8. virtual void on_requested_incompatible_qos(
  9. DDS::DataReader_ptr /*reader*/,
  10. const DDS::RequestedIncompatibleQosStatus & /*status*/)
  11. throw (CORBA::SystemException) {}
  12.  
  13. virtual void on_liveliness_changed(
  14. DDS::DataReader_ptr /*reader*/,
  15. const DDS::LivelinessChangedStatus & /*status*/)
  16. throw (CORBA::SystemException) {}
  17.  
  18. virtual void on_subscription_match(
  19. DDS::DataReader_ptr /*reader*/,
  20. const DDS::SubscriptionMatchStatus & /*status*/)
  21. throw (CORBA::SystemException) {}
  22.  
  23. virtual void on_sample_rejected(
  24. DDS::DataReader_ptr /*reader*/,
  25. const DDS::SampleRejectedStatus & /*status*/)
  26. throw (CORBA::SystemException) {}
  27.  
  28. virtual void on_data_available(
  29. DDS::DataReader_ptr reader)
  30. throw (CORBA::SystemException);
  31.  
  32. virtual void on_sample_lost(
  33. DDS::DataReader_ptr /*reader*/,
  34. const DDS::SampleLostStatus & /*status*/)
  35. throw (CORBA::SystemException) {}
  36. };

We next create a CPP file for the implementaton of on_data_available() named DatabaseNotificationDataReaderListener.cpp. A DataReader is passed into the method as a parameter, and must be narrowed to the specific type it represents — here, a DatabaseNotificationDataReader.

  1. // DatabaseNotificationDataReaderListener.cpp
  2. void DatabaseNotificationDataReaderListenerImpl::
  3. on_data_available(DDS::DataReader_ptr reader)
  4. throw (CORBA::SystemException)
  5. {
  6. try {
  7. DatabaseNotificationDataReader_var databaseNotification_dr
  8. = DatabaseNotificationDataReader::_narrow(reader);
  9. if (CORBA::is_nil(databaseNotification_dr.in()))
  10. throw std::exception(
  11. "DatabaseNotificationDataReaderListenerImpl::
  12. "on_data_available: _narrow failed.");

We must now call take_next_sample() until no more data remains. A DatabaseNotification structure is populated with the data from the sample, and information on the sample is returned in the SampleInfo structure.

  1. while (true) {
  2. DatabaseNotification notification;
  3. DDS::SampleInfo si;
  4. DDS::ReturnCode_t status =
  5. databaseNotification_dr->
  6. take_next_sample(notification, si);

If the status returned from take_next_sample() is DDS::RETCODE_OK, then further processing is needed, though if the return is DDS::RETCODE_NO_DATA, then the while loop can be exited as no more samples are currently available.

If data is available, the instance_state of the SampleInfo must be examined. If the instance_state is either ALIVE (there are live DataWriters writing the instance and the samples have not been disposed) or NOT_ALIVE_NO_WRITERS (there are no live DataWriters but the samples have not been disposed), then the sample is valid and should be processed. The fields of the sample are then converted to .NET types after extracting them from the DatabaseNotification structure, and the method ProcessNotification() is called on the DataServerConnector.

If the instance_state is NOT_ALIVE_DISPOSED, then the samples have been deleted, and should not be consumed.

  1. if (status == DDS::RETCODE_OK) {
  2. if ((si.instance_state == DDS::ALIVE_INSTANCE_STATE) ||
  3. (si.instance_state == DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE))
  4. dataConnector_->ProcessNotification(
  5. dataConnector_,
  6. gcnew ProcessNotificationEventArgs(
  7. notification.id,
  8. gcnew String(notification.table_name.in()),
  9. (Notification)notification.notification_type));
  10. }
  11. else if (status == DDS::RETCODE_NO_DATA)
  12. break;
  13. }

We conclude the method by handling exceptions, and re-throwing them as .NET ones.

  1. }
  2. catch (CORBA::Exception& ex) {
  3. std::stringstream ss;
  4. ss << "Exception in read: " << ex;
  5. throw gcnew DataConnectorException(
  6. gcnew String(ss.str().c_str()));
  7. }
  8. catch (std::exception& ex) {
  9. std::stringstream ss;
  10. ss << "Exception in read: " << ex.what();
  11. throw gcnew DataConnectorException(
  12. gcnew String(ss.str().c_str()));
  13. }
  14. }

The listener is now complete, and we can return to DataServerConnectorDataServerConnector must provide an interface to allow users of the library to receive the data received from the DDS sample. This is accomplished by a .NET event.

To create the event, we first construct a delegate. A delegate is essentially a type-safe function pointer, though it may reference more than one function at a time. For a discussion of delegates, see references [1] and [2].

We return to DataServerConnectorLib.h, and declare a delegate in the DataServerConnector class, and a class member variable of that type.

  1. // DataServerConnectorLib.h, class DataServerConnector
  2. delegate void ProcessNotificationEventHandler(Object^ sender, ProcessNotificationEventArgs ^args);
  3. ProcessNotificationEventHandler^ pProcessNotification_;

The parameters of the delegate match the pattern used by events, as explained in reference [3]. The first parameter is a reference to a System::Object, a fundamental .NET type. The second argument, ProcessNotificationEventArgs, is a custom class which is a subclass of the .NET EventArgs class. This class is no more than a container for the values from the DDS sample, with accessor properties to easily retrieve the stored values. Also in DataServerConnectorLib.h, we declare the class as follows:

  1. // DataServerConnectorLib.h
  2. public ref class ProcessNotificationEventArgs : public EventArgs {
  3. Int64 id_;
  4. String^ tableName_;
  5. Notification notificationType_;
  6. public:
  7. ProcessNotificationEventArgs(Int64 id, String^ tableName,
  8. Notification notificationType) : id_(id),
  9. tableName_(tableName), notificationType_(notificationType) {}
  10. property Int64 ID { Int64 get() { return id_; } }
  11. property String^ TableName { String^ get() { return tableName_; } }
  12. property Notification NotificationType {
  13. Notification get() { return notificationType_; }
  14. }
  15. };

We now create the event itself. In class DataServerConnector, declare the event as follows:

  1. // DataServerConnectorLib.h, class DataServerConnector
  2. event ProcessNotificationEventHandler^ ProcessNotification
  3. {
  4. void add(ProcessNotificationEventHandler^ p)
  5. { pProcessNotification_ +=p; }
  6. void remove(ProcessNotificationEventHandler^ p)
  7. { pProcessNotification_ -=p; }
  8. void raise(Object^ obj, ProcessNotificationEventArgs^ args) {
  9. if (pProcessNotification_!=nullptr)
  10. pProcessNotification_(obj, args);
  11. }
  12. }

The add() method of the event is called whenever an object wishes to become an observer of the event. The reference to the object is added to the delegate variable, and as many observers as desired are allowed. Conversely, remove() is called whenever an object no longer wishes to be an observer of the event. The raise() method is called whenever an event is fired, and all observers are notified of the event. Note that before invoking the delegate, a test for the existence of observers must first be performed (a comparison against nullptr) — invoking a null delegate is an error.

Invoking the event as a function, as the listener does in on_data_available():

  1. // DatabaseNotificationDataReaderListener.cpp
  2. dataConnector_->ProcessNotification(dataConnector_,
  3. gcnew ProcessNotificationEventArgs(
  4. notification.id,
  5. gcnew String(notification.table_name.in()),
  6. (Notification)notification.notification_type));

implicitly calls raise(), which then propagates the sample data to all observers.

Now that DDS elements have been added to DataServerConnectorLib, we must update the MPC file to reflect the changes. The project must now inherit from dcps, link with DatabaseNotificationCommon, reference TAO's orbsvcs, and include DatabaseNotificationDataReaderListenerImpl.cpp in its compilation.

  1. // DataServerConnectorLib.mpc
  2. project : taoexe, dcps, CPPBase {
  3. sharedname = DataServerConnectorLib
  4.  
  5. after += IDL
  6. includes += ../IDL
  7. includes += $(TAO_ROOT)/orbsvcs
  8.  
  9. after += DatabaseNotification
  10.  
  11. Source_Files {
  12. DataServerConnectorLib.cpp
  13. ../IDL/DatabaseC.cpp
  14. DatabaseNotificationDataReaderListenerImpl.cpp
  15. }
  16. IDL_Files {
  17. }
  18.  
  19. managed = 1
  20. }

Adding an observer of the event in the Client is simple. We add a new method to class Client in Client.cs to handle the event — in this case, log that the event arrived:

  1. // Client.cs
  2. void OnNotification(object obj, ProcessNotificationEventArgs args)
  3. {
  4. Log("EVENT: Table '" + args.TableName + "', item " +
  5. args.ID + " " + args.NotificationType);
  6. }

In the constructor of class Client, we register the OnNotification() function with the event, to register the object as an event observer.

  1. dataConnector_.ProcessNotification +=
  2. new DataServerConnector.
  3. ProcessNotificationEventHandler(OnNotification);

In C#, the += operation invokes the event's add() method to perform the observer registration. This completes the database notification subscriber.

Conclusion

The following screen shots demonstrate the system. We start two Clients, as well as the DataServer and the DCPSInfoRepo which are not shown. For this run, the DCPSInfoRepo was run on the machine oci1373 and started with the following command (on a single line):

DCPSInfoRepo -ORBSvcConf lib_tcp.conf -ORBListenEndpoints iiop://:12345 
    -ORBDottedDecimalAddresses 0

The server was started, also on oci1373, with this command (on a single line):

DataServer -ORBSvcConf lib_tcp.conf -ORBListenEndpoints iiop://:12346 
    -DCPSInfoRepo corbaloc::localhost:12345/DCPSInfoRepo 
    -DCPSConfigFile tcp_conf.ini

Each of the Client instances were started with this command (on a single line):

Client -ORBSvcConf lib_tcp.conf -ORBDottedDecimalAddresses 0 
    -ORBInitRef DataServer=corbaloc:iiop:localhost:12346/DataServer 
    -DCPSInfoRepo corbaloc::localhost:12345/DCPSInfoRepo 
    -DCPSConfigFile tcp_conf.ini

Note that, when running on multiple hosts, addresses will need to be updated to use host names or IP addresses as appropriate, instead of localhost.

We enter "My First Item" into the TextBox associated with the Create button on the first Client.

Client one

Pressing the Create button creates the database item, and the generated ID of 1 is reflected in the ListView. An event indicating that item 1 in the Items table was created is also shown in both clients.

Client event

On the second Client, we enter the ID of 1 into the TextBox associated with the Read button.

second client read table

Pressing the Read button displays "My First Item" as the item description, demonstrating that the second Client has referenced the same database as the first Client. An event indicating that item 1 was read is also shown in both clients.

client1 reference client 2 database

This article has described how to mix TAO and OpenDDS in a .NET application to implement a client-server system. The next article in this series will illustrate various refinements to DataServer.

References

[1] An Introduction to Delegates
http://msdn.microsoft.com/en-us/magazine/cc301810.aspx

[2] Delegates, Part 2
http://msdn.microsoft.com/en-us/magazine/cc301816.aspx

[3] Implementation of Events with Delegates
http://msdn.microsoft.com/en-us/magazine/cc301822.aspx

Appendix A

As of OpenDDS 1.3, the default quality of service policies correspond to the following table.

DomainParticipant

USER_DATA value (not set)
ENTITY_FACTORY autoenable_created_entities 1

Topic

TOPIC_DATA value (not set)
DURABILITY kind
service_cleanup_delay.sec
service_cleanup_delay.nanosec
VOLATILE_DURABILITY_QOS
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
DURABILITY_SERVICE service_cleanup_delay.sec
service_cleanup_delay.nanosec
history_kind
history_depth
max_samples
max_instances
max_samples_per_instance
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
KEEP_LAST_HISTORY_QOS
1
LENGTH_UNLIMITED
LENGTH_UNLIMITED
LENGTH_UNLIMITED
DEADLINE period.sec
period.nanosec
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
LATENCY_BUDGET duration.sec
duration.nanosec
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
LIVELINESS kind
lease_duration.sec
lease_duration.nanosec
AUTOMATIC_LIVELINESS_QOS
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
RELIABILITY kind
max_blocking_time.sec
max_blocking_time.nanosec
BEST_EFFORT_RELIABILITY_QOS
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
DESTINATION_ORDER kind BY_RECEPTION_TIMESTAMP_DESTINATIONORDER_QOS
HISTORY kind
depth
KEEP_LAST_HISTORY_QOS
1
RESOURCE_LIMITS max_samples
max_instances
max_samples_per_instance
LENGTH_UNLIMITED
LENGTH_UNLIMITED
LENGTH_UNLIMITED
TRANSPORT_PRIORITY value 0
LIFESPAN duration.sec
duration.nanosec
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
OWNERSHIP kind SHARED_OWNERSHIP_QOS

Publisher

PRESENTATION access_scope
coherent_access
ordered_access
INSTANCE_PRESENTATION_QOS
0
0
PARTITION name (empty sequence)
GROUP_DATA value (not set)
ENTITY_FACTORY autoenable_created_entities 1

Subscriber

PRESENTATION access_scope
coherent_access
ordered_access
INSTANCE_PRESENTATION_QOS
0
0
PARTITION name (empty sequence)
GROUP_DATA value (not set)
ENTITY_FACTORY autoenable_created_entities 1

DataWriter

DURABILITY kind
service_cleanup_delay.sec
service_cleanup_delay.nanosec
VOLATILE_DURABILITY_QOS
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
DURABILITY_SERVICE service_cleanup_delay.sec
service_cleanup_delay.nanosec
history_kind
history_depth
max_samples
max_instances
max_samples_per_instance
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
KEEP_LAST_HISTORY_QOS
1
LENGTH_UNLIMITED
LENGTH_UNLIMITED
LENGTH_UNLIMITED
DEADLINE period.sec
period.nanosec
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
LATENCY_BUDGET duration.sec
duration.nanosec
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
LIVELINESS kind
lease_duration.sec
lease_duration.nanosec
AUTOMATIC_LIVELINESS_QOS
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
RELIABILITY kind
max_blocking_time.sec
max_blocking_time.nanosec
BEST_EFFORT_RELIABILITY_QOS
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
DESTINATION_ORDER kind BY_RECEPTION_TIMESTAMP_DESTINATIONORDER_QOS
HISTORY kind
depth
KEEP_LAST_HISTORY_QOS
1
RESOURCE_LIMITS max_samples
max_instances
max_samples_per_instance
LENGTH_UNLIMITED
LENGTH_UNLIMITED
LENGTH_UNLIMITED
TRANSPORT_PRIORITY value 0
LIFESPAN duration.sec
duration.nanosec
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
USER_DATA value (not set)
OWNERSHIP kind SHARED_OWNERSHIP_QOS
OWNERSHIP_STRENGTH value 0
WRITER_DATA_LIFECYCLE autodispose_unregistered_instances 1

DataReader

DURABILITY kind
service_cleanup_delay.sec
service_cleanup_delay.nanosec
VOLATILE_DURABILITY_QOS
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
DEADLINE period.sec
period.nanosec
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
LATENCY_BUDGET duration.sec
duration.nanosec
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
LIVELINESS kind
lease_duration.sec
lease_duration.nanosec
AUTOMATIC_LIVELINESS_QOS
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
RELIABILITY kind
max_blocking_time.sec
max_blocking_time.nanosec
BEST_EFFORT_RELIABILITY_QOS
DURATION_INFINITY_SEC
DURATION_INFINITY_NSEC
DESTINATION_ORDER kind BY_RECEPTION_TIMESTAMP_DESTINATIONORDER_QOS
HISTORY kind
depth
KEEP_LAST_HISTORY_QOS
1
RESOURCE_LIMITS max_samples
max_instances
max_samples_per_instance
LENGTH_UNLIMITED
LENGTH_UNLIMITED
LENGTH_UNLIMITED
USER_DATA value (not set)
TIME_BASED_FILTER minimum_separation.sec
minimum_separation.nanosec
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
READER_DATA_LIFECYCLE autopurge_nowriter_samples_delay.sec
autopurge_nowriter_samples_delay.nanosec
DURATION_ZERO_SEC
DURATION_ZERO_NSEC
OWNERSHIP kind SHARED_OWNERSHIP_QOS

Appendix B

A transport can be selected in several different ways. One way is to defer the selection of the transport to the user at runtime. For example, the following process allows the SimpleTcp transport to be used.

Method 1

For this method, create two files in the filesystem. Name the first file lib_tcp.conf, and add the following to it (as one line):

  1. // lib_tcp.conf
  2. dynamic DCPS_SimpleTcpLoader Service_Object *
  3. SimpleTcp:_make_DCPS_SimpleTcpLoader() "-type SimpleTcp"

Name the second file tcp_conf.ini, and include a section to configure SimpleTcp transport parameters, as follows

  1. // tcp_conf.ini
  2. [transport_impl_1]
  3. transport_type=SimpleTcp

In DataServer.cpp, define a constant equal to the same value as the number of the section in tcp_conf.ini — here, the value 1, to match transport_impl_1 — and then use create_transport_impl() to create the transport:

  1. // DataServer.cpp (method 1)
  2. const OpenDDS::DCPS::TransportIdType TRANSPORT_IMPL_ID = 1;
  3.  
  4. OpenDDS::DCPS::TransportImpl_rch trans_impl =
  5. TheTransportFactory->create_transport_impl(
  6. TRANSPORT_IMPL_ID,
  7. OpenDDS::DCPS::AUTO_CONFIG);

When DataServer is run with these command-line arguments:

-ORBSvcConf lib_tcp.conf -DCPSConfigFile tcp_conf.ini

the SimpleTcp transport will be loaded by lib_tcp.conf, referenced as the transport to use in tcp_conf.ini, and instantiated by create_transport_impl().

Method 2

A second way to load a transport is entirely within code, by name, so external configuration files are not required. A call to get_or_create_configuration() will load a specific transport at the specified ID for use by create_transport_impl():

  1. // DataServer.cpp (method 2)
  2. const OpenDDS::DCPS::TransportIdType TRANSPORT_IMPL_ID = 1;
  3.  
  4. TheTransportFactory->get_or_create_configuration(
  5. TRANSPORT_IMPL_ID, ACE_TEXT("SimpleTcp"));
  6.  
  7. OpenDDS::DCPS::TransportImpl_rch trans_impl =
  8. TheTransportFactory->create_transport_impl(
  9. TRANSPORT_IMPL_ID,
  10. OpenDDS::DCPS::AUTO_CONFIG);

Method 3

A third way is to use constants for transport IDs that have been predefined for the standard transports, rather than defining an ID of our own:

  1. // DataServer.cpp (method 3)
  2. OpenDDS::DCPS::TransportImpl_rch trans_impl =
  3. TheTransportFactory->create_transport_impl(
  4. ::OpenDDS::DCPS::DEFAULT_SIMPLE_TCP_ID,
  5. OpenDDS::DCPS::AUTO_CONFIG);

The Middleware News Brief is a periodic newsletter. The purpose and intent of this publication is to advance and inform on features, news, and technical information about Open Source, middleware technologies (e.g., TAO, OpenDDS, JacORB), including case studies and other technical content.

© Copyright Object Computing, Inc. 1993, 2016. All rights reserved

Subscribe

secret