Table of Contents
- Choose or Create an ArmarX Package
- Create the Topic Interface
- Implement the Publisher
- Implement the Subscriber
- Test Publisher and Subscriber
- But There's More ...
- Next Up
Objective: Learn how to create the Ice interface of a topic, a publisher component sending messages on that topic and a subscriber component receiving messages of that topic in ArmarX.
Previous Tutorials: Understand Distributed Systems, Write a Server and Client Communicating via RPC (C++)
Next Tutorials: None
Reference Code: component_tutorials
In this tutorial, you will implement two components communicating in the publisher-subscriber paradigm (i.e. using a topic).
We will use the following names:
- Topic:
SimpleTopic - Publisher:
simple_publisher - Subscriber:
simple_subsriber
Choose or Create an ArmarX Package
To create a component, you need an ArmarX package. If you followed Create a "Hello World!" Component in ArmarX (C++), you can use the package created there:
If you want to create a new package, refer to Create a "Hello World!" Component in ArmarX (C++), or the ArmarX Package Tool for how to do that.
For this tutorial text, we will assume the ArmarX package is called component_tutorials.
Create the Topic Interface
Every topic needs an Ice interface, which defines what kind of messages can be sent over the topic. In the Server-Client Tutorial, we customized the server's Ice interface, which was implemented by the server and used by the client. However, a topic's interface does not belong to any publisher or subscriber in specific - it stands for its own. In other words, the topic's interface is used by both publishers and subscribers. In such cases - where we need a piece of software that is meant to be used by multiple other pieces of code - we need a library.
Create a Library
A software library is a collection of code (classes, functions, variables, ...), just like a real library is a collection of books. A library itself is not an executable program, as it does not contain a main() function. Instead, a library is a re-usable building block for your and others' applications.
Think about it like this:
- Your executable application represents a very specific use-case: yours. An application defines what code runs, which command line argument it takes, what it does and how it behaves.
- For example, maybe you implemented a face recognition program that reads images from a camera and detects faces in the image. This is very special use case - someone else might want to do things a bit differently. For instance, someone else, Alice, might want to use your face recognition, but read images from files, because they have no camera in their setup, but just want to run the face recognition offline on a bulk of images.
- If your face recognition code is only contained in a monolithic application (with a
main()function), there is no simple way for Alice to re-use your code performing the face recognition and just replacing the part reading the images. - In that case, it is better if the core of the face recognition code (containing the main business logic, e.g. taking an image and returning detected faces) is contained in a re-usable library that is used by both your application and Alice's.
- In a nutshell, think of a library as a toolbox full of building blocks that a developer (including yourself) can use to build their specific use case. The general goal is the following: If someone has a use-case that is differs from yours by 10%, they should only need to change 10% percent of the code. (Of course, it is difficult to measure the difference between use-cases, but that is the basic idea.)
To create a library, we can use armarx-package:
The output should be something like this:
As you can see, armarx-package created a number of files in component_tutorials/source/component_tutorials/simple_topic. This is the root directory of our new library.
To test whether this worked for CMake, run build the project:
Alright! As you can see, we got ourselves two new targets: simple_topic and simple_topicTest.
simple_topicis the actual C++ librarysimple_topciTestis an executable prepared where you can add unit tests
Have a look at the CMakeLists.txt located in component_tutorials/source/component_tutorials/simple_topic. It should look something like this:
Here, you can recognize these two targets: One is a library (declared with armarx_add_library()), and the other is a test (declared with armarx_add_test()).
For more information, see CMake/Libraries.
The library can contain ordinary C++ code (types, classes, functions, ...) which can be easily reused by other libraries, components or applications (or CMake targets in general). It will also be the host of our topic's Ice interface.
Add an Ice Interface Library
Similar to the interface of a component, an Ice topic requires an interface defined in Slice (the Specification Language for Ice, see Understand Distributed Systems for more details). That means we have to define the topic interface in a Slice (.ice) file, and Ice will generate C++ header and source files which we can include and use in our C++ code.
Therefore, Ice interfaces are defined in their own libraries ("Ice libraries" or "Slice interface libraries"). The CMake/Slice interface libraries explains how to declare it. In our case, we will associate the Ice library with our (standard C++) library simple_topic. It is customary to add a _ice suffix to the library name in that case. That is, the name of the Ice library will be simple_topic_ice.
So, we add this part to the CMakeLists.txt at the beginning of the simple_topic library:
For that to work, we need to create the file ice/SimpleTopic.ice. Similar to the _ice suffix, it is customary to put Slice files into their own directory called ice/, so you know what to expect when writing or reading the respective include.
Thus, create the directory simple_topic/ice/, and inside a file called SimpleTopic.ice (where SimpleTopic is the name of the topic we will define). Add these contents:
- We use modules to create the C++ namespace
component_tutorials::simple_topic, which is the C++ namespace of our library (matching the directory structure). - We define an interface
MyTopicInterface- the interface of our topic. - In the interface, we define a function
reportSequenceNumber(), which takes anint numberand returns nothing.
This should look familiar: For comparison, this is the interface of the server from the RPC tutorial:
Just looking at the Slice code, there is not much of a difference between the interface of a server (component) and a topic. There is one important difference, though: Functions of topic interfaces must always return void. That is because topics are a unidirectional way of communication: Messages flow from publishers to subscribers, but there is no response by the subscribers to the publishers. The publishers will call generatereRandomNumber(...) on the topic, passing the message content as arguments, but they will not get anything back.
Alright, we have got the Ice interface of our topic. Next, let us create a component publishing messages on that topic.
Implement the Publisher
Create the Publisher Component
First, we create and implement the publisher component. In the root directory of the ArmarXPackage, use armarx-package to create a new component called simple_publisher:
Try to build the fresh component: Navigate to the build directory of the package and run cmake and make.
Since a publisher publishes data on the topic (it "offers" the topic), it takes the "active" role in the sense that it initiates the communication. In contrast, the subscriber will "react" to incoming messages. Compared to server and client in RPC, the publisher is similar to the client (proactively publishing a message), and the subscriber is similar to the server (receiving and reacting to an incoming message).
Therefore, similar to an RPC client, the publisher does not need to have a special kind of Ice interface - this will only be necessary for the subscriber later on.
Implement the Publisher
Connect to the Topic for Publishing Messages - "Offer" the Topic
First, the publisher needs to inform Ice that it is going to publish messages on the Topic - and that there is a topic in the first place. In other words, the publisher offers the topic. To do that, requires two things:
- The topic's Ice interface type - this is what we defined in the previous step
- The topic's name - similar to the name of a server.
In contrast to RPC, where the name belongs to a running component, the topic name can be anything - the topic will be created by Ice if it does not exist yet. However, the subscriber needs to use the same topic name - otherwise it will not receive any messages from the publisher.
Let us tackle point 1. We can get the Ice interface type by including it:
simple_publisher/Component.h:
This is the header file that Slice generated from the SimpleTopic.ice file we wrote earlier on. Similar to a client RPC, the publisher's contact person communicating over Ice is a proxy. Using a proxy of the topic, the publisher can call the topic's interface functions, and Ice will take care of the actual communication to the subscribers behind the scenes.
As we need a proxy to the topic interface simple_topic::SimpleTopicInterface, the corresponding proxy type is simple_topic::SimpleTopicInterfacePrx (note the trailing Prx). So we just add a private member to the simple_publisher::Component class in the header file of simple_publisher such as this:
The fully qualified name of the topic interface is
::component_tutorials::simple_topic::SimpleTopicInterfacePrx. Because the publisher is also in the::component_tutorialsnamespace, we can qualify the topic interface type only assimple_topic::SimpleTopicInterfacePrx.
Try building now. You will probably get errors like this:
We encountered the same error in the RPC tutorial. If you look back there, the solution was linking to the correct library - in this case an Ice library (since there errors are related to the SimpleTopicPrx). Therefore, we need to modify the CMakeLists.txt of simple_publisher. When you open it, it should look like this (some out-commented parts removed):
Now, you might be prompted to add the library to the ICE_DEPENDENCIES - it is an Ice library after all, is it not? However, ICE_DEPENDENCIES refers to dependencies of your own Ice interface; in this case the Ice interface of the simple_publisher. But we did not add the SimpleTopicInterface to the simple_publisher's interface, so that can stay untouched.
Instead, the simple_publisher's implementation requires the definitions of the SimpleTopicPrx. The dependencies of that code are listed in DEPENDENCIES - not ICE_DEPENDENCIES.
The library we need is simple_topic_ice - that is the name we have chosen earlier on for the Ice interface containing the SimpleTopicInterface. So let us add it to DEPENDENCIES:
If you try to compile now, it should work again. However, take a closer look at the output of CMake. If you see the lines
then make sure that, in component_tutorials/source/component_tutorials/CMakeLists.txt, the libraries are listed before the components:
Now, the output of CMake should just contain
One thing is still missing: We have created the proxy of the topic, but we did not initialize it yet. In order to actually connect to the topic, we need to inform ArmarX about our intentions.
Head over to the source file of the simple_publisher component. Change the body of the function createPropertyDefinitions() to:
The line def->topic(topic, ...); does multiple things:
- Create a property for the name of the topic.
- A property is basically a parameter of a component or application. Its value can be specified via the command line or a scenario.
- The property's default value is set to
"MySimpleTopic" - The property's name and human-readable description are specified as well.
- By creating a property for the component's name, users are allowed to change it via configuration without re-compiling code.
- Ask ArmarX to set up the communication with the topic.
- Declare that the publisher offers the topic:
- The topic's name is specified by the property (
"MySimpleTopic"by default). - The topic's interface is specified by the type of
topic
(simple_topic::SimpleTopicInterfacePrx). - The publisher does not depend on another component: If no one is listening, the published messages will just vanish in the ether.
- The topic's name is specified by the property (
- Before the publisher's
onConnectComponent()is called, the variabletopicwill be initialized with a valid proxy to the topic.- Therefore, the earliest point where we are allowed to use proxies such as
topicis during theonConnectComponent(). - Or, in other words, we are not allowed to use proxies before
onConnectComponent()(e.g. in theonInitComponent()). - You can find more details about the component lifecycle [in the documentation](armarx::ManagedIceObject).
- Therefore, the earliest point where we are allowed to use proxies such as
- Declare that the publisher offers the topic:
In summary, the line defs->topic(topic, ...) makes sure that the variable topic is initialized and ready to use once the client reaches its onConnectComponent(), and allows the user to configure the topic's name.
You can also do these things "by hand", i.e. by writing the code yourself. You will see that a lot in older components which were written when the wrapper
defs->topic()was not available yet. For reference, the old pattern went like this:
- Define the property in
createPropertyDefinitions():Component::createPropertyDefinitions(){...def->defineOptionalProperty("tpc.pub.SimpleTopicName", "MySimpleTopic","Name of the simple topic.");...}- Declare that this component is using another component in the
onInitComponent():voidComponent::onInitComponent(){offeringTopic(getProperty<std::string>("tpc.pub.SimpleTopicName").getValue());}- Obtain the proxy to the other component in the
onConnectComponent():voidComponent::onConnectComponent(){server = getTopic<simple_topic::SimpleTopicInterfacePrx>(getProperty<std::string>("tpc.pub.SimpleTopicName").getValue());}As you see, that is a lot of boilerplate code, i.e. code that is almost always the same, for a very common use case. Therefore, we introduced the
defs->topic()pattern to support the standard use case, which should get you going in most cases with a single line.If you have more special requirements, though, you can always fall back to the "manual" way of doing things.
Actually, we can make the line even shorter. Have another look at our current code:
This still feels a bit repetitive, does it not? In fact, in a standard setup like this, defs->topic() is able to infer the default name and generate a suitable description just from the topic argument (or, more precisely, its type). Thus, we can reduce the call to just:
It does not get much more minimal than that. However, remember that you can always override the defaults if they do not fit your specific use case. For example, you can specify the default name as second argument, and still have the property name and description generated automatically:
Call the Topic Interface
Finally, we are ready to publish messages on the topic. At this point, that only takes a single line (and one line to print the result):
This publishes the number on the reportSequenceNumber() function of the topic, after logging this action using ARMARX_INFO.
Publish in a Running Task
Before we continue with the subscriber, let us make the publisher a bit more interesting by letting it publish a stream of increasing sequence numbers. As this is a longer undertaking, we should not do that in onConnectComponent(). Instead, we use a running tasks - i.e. a thread that we can start in onConnectComponent(), but which keeps running afterwards.
In the header file Component.h, include
and add a armarx::SimpleRunningTask<>::pointer_type to the component:
Also, add a function called publishSequenceNumbers() to your component:
In the source file Component.cpp, remove the current code from onConnectComponent(). Instead, initialize and start the running task:
Clean up the task in onDisconnectComponent():
Add the definition of publishSequenceNumbers() to Component.cpp and implement it like this:
For armarx::Metronome, you need to include an additional header in Component.cpp:
The function publishSequenceNumbers() will keep publishing a sequence of increasing numbers (starting at 1) until the task is stopped (which happens in onDisconnectComponent()). The armarx::Metronome makes sure that our loop runs at a fixed rate (instead of as fast as it can). Here, we chose 2 Hz, i.e. two messages per second.
- Note
- Do not use
task->isRunning()in the condition of the while loop.isRunning()indicates that the task is actually still running. In contrast,isStopped()indicates that the task has been stopped from outside, which is the information we need here.
Build your package. It is time for testing!
Test the Publisher
Now let us have a look at what our implementation does so far. Make sure the Ice service is running on your local system or in your network and that you can connect to it. This can be done via the following command:
If you get a message like the following,
you first need to start Ice using armarx start.
Next, start an ArmarX GUI via
In the ArmarX GUI, open the log viewer (LogViewer) if it is not already open. With the log viewer opened, run the server by executing the simple_publisher_run executable in a terminal window from the component_tutorials/build/bin directory:
and stop it by pressing Ctrl+C after a while.
The log should show an entry like this:
Or, in text form:
As you can see, the publisher works without a subscriber counterpart: It just starts publishing messages on the topic, each one 0.5 seconds apart.
Still, without someone listening to these messages, life is a bit boring. So now, let us go to the subscriber.
Implement the Subscriber
Create the Subscriber Component
First, we create and implement the subscriber component. In the root directory of the ArmarXPackage, use armarx-package to create a new component called simple_subscriber:
Try to build the fresh component: Navigate to the build directory of the package and run cmake and make.
Add the Ice Interface of the Topic to that of the Subscriber Component
As noted above, the subscriber is a bit similar to an RPC server as it needs to receive and react to incoming messages (although the subscriber does not send back a response). Therefore, it needs to offer an entry point where Ice can deliver incoming messages. It does that by implementing the topic's Ice interface.
As a component may only have one "final" Ice interface defined in Slice (this has to do with virtual inheritance and a diamond, long story), the subscriber needs to add the topic's Ice interface to its own Ice interface, which is defined in the Slice file ComponentInterface.ice. When we want to customize the component's interface, we need to edit that file.
Refer to ../102_distributed_systems/README.md "Understand Distributed Systems" to learn more about Slice and Ice interfaces.
Edit the Generated ComponentInterface.ice
The subscriber's generated Ice interface looks like this:
The ComponentInterface is the single Ice interface implemented by the simple_subscriber. But it can actually combine multiple other Ice interfaces by extending them.
To extend our SimpleTopicInterface, we first need to include it:
- Note
- In Slice, make sure to include the
.icefile, not the generated.hfile!
Then, let ComponentInterface extend the SimpleTopicInterface - do not forget about the simple_topic namespace, though:
If you compile your project now, you should get many new errors of the form
This is because we are using another Ice library, but do not depend on it. Again, this is CMake's business, so we need to inform it.
Open the simple_subsriber/CMakeLists.txt. As usual, it looks like this:
Now, what do we need to add? Just as with the publisher, you might want to add simple_topic_ice to DEPENDENCIES. However, here, the situation is different: The Ice interface of simple_subscriber (i.e. the Slice code) is already using simple_topic_ice, not just the subscriber's C++ code. Therefore, we need to add it to ICE_DEPENDENCIES:
Internally, the subscriber's implementation (simple_subscriber_cmp) already depends on its Ice interface (simple_subscriber_ice), so it gets simple_topic_ice transitively.
Implement the Topic Interface in the Subscriber
Build the Project and ... Get an Error?
Before we continue, try to build the project now.
You should get an error like this:
This is very similar to what we encountered when implementing the RPC server. There, the error was that the class Component inherited a pure virtual (aka "abstract") method, but does not implement it. This makes the whole class Component abstract, which in turn prohibits its direct instantiation (which causes new T() with T = Component to fail). We also are informed about the culprit: reportSequenceNumber(), the function of our topic's Ice interface.
Add the Topic's Ice Interface Function to the Component Class
So next, we have to implement the function reportSequenceNumber() defined in our topic interface. This will also tell the subscriber what to do when receiving a message.
First, open the header file Component.h. The important lines are:
This line imports the C++ declarations of the Ice interface. This includes the generated interface class component_tutorials::components::simple_subscriber::ComponentInterface.
Apart from the base component class armarx::Component, our class Component derives from the Ice interface class public component_tutorials::components::simple_subscriber::ComponentInterface that was generated from the ComponentInterface.ice file.
As said, because ComponentInterface contains the pure virtual function reportSequenceNumber() and Component does not override and implement it, Component is virtual as well. We can change that by adding a declaration and definition of reportSequenceNumber() in Component.
In QtCreator, you can use the same Alt+Enter trick as for the RPC server (cursor on class Component : > Alt+Enter > Insert Virtual Functions of Base Classes) to add a declaration of reportSequenceNumber() in Component:
Our work is not done: We still have to implement the function in Component.cpp. Use QtCreator and Alt+Ender on reportSequenceNumber() to generate the implementation:
This callback function is where we define what the subscriber should do with each incoming message. The message data is passed as argument. Just as the Ice interface function, the function returns nothing. As almost always, we can ignore the additional Ice::Current argument.
If you build now ... everything should compile! Since we do not need to return a response, we could leave the callback empty. But let us add something, so we can see what is going on:
Implement the Callback Function
For the moment, let us just log that we received a message and what its contents are:
Note that we removed the argument name of Ice::Current to avoid an unused argument warning.
Okay, looks like we are done with the subscriber!
...
Turns out, there is something we forgot: In contrast to an RPC server, which is ready to receive requests by clients automatically, a topic subscriber has to actively subscribe the topic before it can receive any messages.
Subscribe the Topic
Luckily, subscribing the topic is simple. As the publisher, all the subscriber needs is the topic's Ice interface and the topic's name. All we need to do is inform ArmarX about the intended subscription. We can do that easily by in createPropertyDefinitions():
Note that the subscriber specifies the vanilla topic interface here (SimpleTopicInterface), not the proxy. Since the subscriber receives data from Ice via its Ice interface, it does not need a proxy.
Here, we can again specify the default name of the topic: MySimpleTopic in our case. In order to receive messages, the subscriber must subscribe the same topic as the publisher is publishing to. This implies the same interface and the same name. By setting the default name to the same value in both the subscriber and the publisher, this should work right away. If you want to change the topic name, e.g. via properties in the scenario manager, you need to do that for both the publisher and subscriber.
And that's it!
- Note
- Your subscriber will start receiving topic messages after exiting
onConnectComponent(). As subscribers are reactive by nature and are expected to miss messages sent before they were started, this usually does not matter. However, keep in mind that you should not rely on receiving topic messages while still running in theonConnectComponent(). - For reference, here is how subscribe to a topic manually:
- Define the property in
createPropertyDefinitions():Component::createPropertyDefinitions(){...def->defineOptionalProperty("tpc.sub.SimpleTopicName", "MySimpleTopic","Name of the simple topic.");...} - Declare that this component is using another component in the
onInitComponent():voidComponent::onInitComponent(){usingTopic(getProperty<std::string>("tpc.sub.SimpleTopicName").getValue());} - No need to obtain a proxy in
onConnectComponent(), though.
Test Publisher and Subscriber
Now, you have two applications executing the publisher and subscriber separately in two processes.
Before we continue, add some more code to illustrate what happens:
- Change the publisher-side code in the while loop of
publishSequenceNumbers()to:while (task and not task->isStopped()){topic->reportSequenceNumber(number);number++;metronome.waitForNextTick();} - Change the subscriber-side code of
reportSequenceNumber()to:void Component::reportSequenceNumber(Ice::Int number, const Ice::Current& current){ARMARX_INFO << "[Subscriber] Received sequence number " << number << ".";armarx::Clock::WaitFor(armarx::Duration::MilliSeconds(1000));ARMARX_INFO << "[Subscriber] Finished processing number " << number << ".";}
For the subscriber, we added a armarx::Clock::WaitFor() to simulate a longer processing step. This requires an additional header in simple_subscriber/Component.cpp:
You are almost done. Woo! The (almost) only thing remaining now is having a look at what our code does.
Start the Components from the Terminal
As before, start an ArmarX GUI and open the log viewer (if not done already). Now, run the applications simple_subscriber_run and simple_publisher_run located in component_tutorials/build/bin/ in two terminals:
- Terminal 1: cd .../component_tutorials/build/bin/./simple_subscriber_run
- Terminal 2: cd .../component_tutorials/build/bin/./simple_publisher_run
Watch the log viewer while you are doing that. You should see the following:
or on the terminals:
and
The publisher does what it does before: It publishes one message after the other. What is new is that the subscriber is now receiving these messages. However, the publisher will never know about this ...
Have a closer look at the time stamps in the log (we left it in this time). The publisher is keeping its 2 Hz rate. In the combined log in the ArmarX GUI, you can see that
- the publisher is finished with sending right away - no need to wait for the subscriber to finish (or even receive the message)
- the subscriber is getting the message almost right away after it has been sent - although it may still be processing previous old message. This shows us the subscriber receives the messages in multiple threads which may run alongside each other. As a consequence, take care when accessing shared data (e.g. member variables of
Component) from topic callbacks: You should guard access to shared data with astd::mutexand astd::scoped_lockin such cases.
Try stopping and restarting the components while the other is running. What can you observe?
Start the Components in a Scenario
We will leave this as a small exercise to you. Have a look at the Create a "Hello World!" Component in ArmarX (C++) and the Write a Server and Client Communicating via RPC (C++) Do not skip this!
The resulting scenario could look like this:
Good job! You now know how to ...
- create a C++ library,
- create an Ice interface library,
- define a topic Ice interface,
- create a publisher component publishing to that topic, and
- create a subscriber component, add the topic to its Ice interface and implement the topic interface.
That is really great!
But There's More ...
However, there are still some things we did not cover:
- The Ice interface function
generateRandomNumber()currently takes one argument. What if we need to add more data to each message? How much work is it to update all subscribers implementing and publishers using the interface? Can we prepare our interface to reduce the required maintenance? - How can we add more parameters (aka properties) to a component that we can configure in the scenario?
So many questions! We will attend to them in future tutorials.
Next Up
Good job! You finished the basic component tutorials. You should now have a better understanding of Ice in ArmarX and what distributed systems feel like.
However, these are just the technical foundations! Feel free to roam the more advanced tutorials and discover the world of ArmarX.
