7.2.3. Protobuf: Person#

In the last section you learned how to send strings to an eCAL Topic. Using strings is great for simple data that has a textual representation. Quite often however your data will be more complex, so you need some kind of protocol that defines how your data is structured.

Our recommended way is to use Google protobuf to do that, because:

It solves the problem of how to serialize and de-serialize data for you
You get downward compatibility out of the box (if you follow the guidelines)
It is maintained by Google and the API is stable
The eCAL Monitor can display a reflection view of the data

Important

It is important to remember, that all your applications must agree on the data format. As protobuf messages are defined in .proto files, all of your applications should share the same files.

eCAL supports protobuf serialization natively for C++, C# and Python.

The usage of protobuf for data exchange in eCAL is very simple. You know already from the “String: Hello World” how to send and receive simple string data. The basic setup will be the same, but instead of using the string publisher, we will use the protobuf publisher and subscriber.

7.2.3.1. Person Protobuf#

As the sender and receiver need the same .proto files, we place them in a separate directory next to the source directories for the sender and the receiver.

 Person Protobuf File
├─  person.proto
│
├─  animal.proto
│
└─  house.proto

Let’s start with the protobuf/person.proto file!

syntax = "proto3";

// import message definitions from different proto files
import "animal.proto";
import "house.proto";

// assign a package name, that will appear as a namespace
package pb.People;

// create the message "Person" with
//   - base types like int32 and string
//   - individual defined enum SType
//   - imported message types
message Person
{
  enum SType
  {
    MALE   = 0;
    FEMALE = 1;
  }

  int32  id               = 1;
  string name             = 2;
  SType  stype            = 3;
  string email            = 4;

  Animal.Dog        dog   = 5;
  Environment.House house = 6;
}

As you can see, the person.proto file imports also other message definitions: animal.proto and house.proto. So we need them as well. The definitions are straight forward. For more information about protobuf, please refer to the detailed official documentation.

syntax = "proto3";

package pb.Animal;

message Dog
{
  string name   = 1;
  string colour = 2;
}

syntax = "proto3";

package pb.Environment;

message House
{
  int32 rooms = 1;
}

7.2.3.2. Person Publisher#

The main differences to the string publisher are:

we need to include / import the protobuf specific publishers
also we need to include / import the compiled protobuf message definitions for the specific programming language
we need to utilize the protobuf message class Person instead of the string class

// Including the eCAL convenience header
#include <ecal/ecal.h>
// In addition we include the msg protobuf publisher
#include <ecal/msg/protobuf/publisher.h>

#include <iostream>

// Here we include the compiled protobuf header for the message "Person"
#include "person.pb.h"

int main()
{
  std::cout << "--------------------" << std::endl;
  std::cout << " C++: PERSON SENDER"  << std::endl;
  std::cout << "--------------------" << std::endl;

  /*
    Initialize eCAL. You always have to initialize eCAL before using its API.
    The name of our eCAL Process will be "person send". 
    This name will be visible in the eCAL Monitor, once the process is running.
  */
  eCAL::Initialize("person send");

  /*
    Print some eCAL version information.
  */
  std::cout << "eCAL " << eCAL::GetVersionString() << " (" << eCAL::GetVersionDateString() << ")" << "\n";

  /*
    Set the state for the program.
    You can vary between different states like healthy, warning, critical ...
    This can be used to communicate the application state to applications like eCAL Monitor/Sys.
  */
  eCAL::Process::SetState(eCAL::Process::eSeverity::healthy, eCAL::Process::eSeverityLevel::level1, "I feel good!");

  /*
    Now we create a new publisher that will publish the topic "person".
    The data type is "pb.People.Person", generated from the protobuf definition.    
  */
  eCAL::protobuf::CPublisher<pb::People::Person> publisher("person");

  /*
    Construct a message. The message is a protobuf struct that will be sent to the subscribers.
  */
  pb::People::Person person;
  person.set_id(0);
  person.set_name("Max");
  person.set_stype(pb::People::Person_SType_MALE);
  person.set_email("max@mail.net");
  person.mutable_dog()->set_name("Brandy");
  person.mutable_house()->set_rooms(4);

  /*
    Creating an infinite publish-loop.
    eCAL Supports a stop signal; when an eCAL Process is stopped, eCAL_Ok() will return false.
  */
  auto loop_count = 0;
  while(eCAL::Ok())
  {
    /*
      Change in each loop the content of the message to see a difference per message.
    */
    person.set_id(loop_count++);

    /*
      Send the message. The message is sent to all subscribers that are currently connected to the topic "person".
    */
    if (publisher.Send(person)) 
    {
      std::cout << "----------------------------------"        << "\n";
      std::cout << "Sent protobuf message in C++: "            << "\n";
      std::cout << "----------------------------------"        << "\n";
      std::cout << "person id    : " << person.id()            << "\n";
      std::cout << "person name  : " << person.name()          << "\n";
      std::cout << "person stype : " << person.stype()         << "\n";
      std::cout << "person email : " << person.email()         << "\n";
      std::cout << "dog.name     : " << person.dog().name()    << "\n";
      std::cout << "house.rooms  : " << person.house().rooms() << "\n";
      std::cout << "----------------------------------"        << "\n";
      std::cout                                                << "\n";
    }
    else
    {
      std::cout << "Failed to send Protobuf message in C++!"   << "\n";
    }

    /*
      Sleep for 500ms to send in a frequency of 2 hz.
    */
    eCAL::Process::SleepMS(500);
  }

  /*
    Finalize eCAL. This will stop all eCAL processes and free all resources.
    You should always finalize eCAL when you are done using it.
  */
  eCAL::Finalize();

  return(0);
}

using System;
// Include the eCAL API namespace
using Eclipse.eCAL.Core;

public class PersonSend
{
  static void Main()
  {
    Console.WriteLine("-------------------");
    Console.WriteLine(" C#: PERSON SENDER");
    Console.WriteLine("-------------------");

    /*
      Initialize eCAL. You always have to initialize eCAL before using its API.
      The name of our eCAL Process will be "person send c#". 
      This name will be visible in the eCAL Monitor, once the process is running.
    */
    Core.Initialize("person send c#");

    /*
      Print version info.
    */
    Console.WriteLine(String.Format("eCAL {0} ({1})\n", Core.GetVersionString(), Core.GetVersionDateString()));

    /*
      Set the state for the program.
      You can vary between different states like healthy, warning, critical ...
      This can be used to communicate the application state to applications like eCAL Monitor/Sys.
    */
    Process.SetState(eProcessSeverity.Healthy, eProcessSeverityLevel.Level1, "I feel good!");

    /*
      Now we create a new publisher that will publish the topic "person".
      The data type is "Pb.People.Person", generated from the protobuf definition.    
    */
    var publisher = new ProtobufPublisher<Pb.People.Person>("person");

    /*
      Construct a message. The message is a protobuf struct that will be sent to the subscribers.
    */
    var person = new Pb.People.Person
    {
      Id = 0,
      Email = "max@online.de",
      Name = "Max",
      Stype = Pb.People.Person.Types.SType.Female,
      Dog = new Pb.Animal.Dog { Name = "Brandy" },
      House = new Pb.Environment.House { Rooms = 4 }
    };

    /*
      Creating an infinite publish-loop.
      eCAL Supports a stop signal; when an eCAL Process is stopped, eCAL_Ok() will return false.
    */
    int loop_count = 0;
    while (Core.Ok())
    {
      /*
        Change in each loop the content of the message to see a difference per message.
      */
      person.Id = loop_count++;

      /*
        Send the message. The message is sent to all subscribers that are currently connected to the topic "person".
      */
      if (publisher.Send(person))
      {
        Console.WriteLine("----------------------------------");
        Console.WriteLine("Sent protobuf message in C#: ");
        Console.WriteLine("----------------------------------");
        Console.WriteLine("person id    : {0}", person.Id);
        Console.WriteLine("person name  : {0}", person.Name);
        Console.WriteLine("person stype : {0}", person.Stype);
        Console.WriteLine("person email : {0}", person.Email);
        Console.WriteLine("dog.name     : {0}", person.Dog.Name);
        Console.WriteLine("house.rooms  : {0}", person.House.Rooms);
        Console.WriteLine("----------------------------------");
        Console.WriteLine("");
      }
      else
      {
        Console.WriteLine("Sending protobuf message in C# failed!");
      }

      /*
        Sleep for 500ms to send in a frequency of 2 hz.
      */
      System.Threading.Thread.Sleep(500);
    }

    /*
      Cleanup. Dispose the publisher to free resources.
    */
    publisher.Dispose();

    /*
      Terminate eCAL. This will stop all eCAL processes and free all resources.
      You should always terminate eCAL when you are done using it.
    */
    Core.Terminate();
  }
}

import os
import sys
import time
# import the eCAL core API
import ecal.nanobind_core as ecal_core
# import the eCAL publisher AP
from ecal.msg.proto.core import Publisher as ProtobufPublisher

import google.protobuf
from packaging import version
proto_version = version.parse(google.protobuf.__version__)
sys.path.insert(1, os.path.join(sys.path[0], f"./_protobuf/v{proto_version.major}"))
# import the compiled protobuf message classes
import person_pb2

def main():
  print("-------------------------------")
  print(" Python: PERSON SENDER         ")
  print("-------------------------------")
  
  # Initialize eCAL. You always have to initialize eCAL before using it.
  # The name of our eCAL Process will be "person send python".
  # This name will be visible in the eCAL Monitor, once the process is running.
  ecal_core.initialize("person send python")

  # Print used eCAL version and date
  print("eCAL {} ({})\n".format(ecal_core.get_version_string(), ecal_core.get_version_date_string()))
 
  # Set the state for the program.
  # You can vary between different states like healthy, warning, critical ...
  # This can be used to communicate the application state to applications like eCAL Monitor/Sys.
  ecal_core.process.set_state(ecal_core.process.Severity.HEALTHY, ecal_core.process.SeverityLevel.LEVEL1, "I feel good!")
  
  # Creating the eCAL Publisher. An eCAL Process can create multiple publishers (and subscribers).
  # The topic we are going to publish is called "person".
  pub = ProtobufPublisher[person_pb2.Person](person_pb2.Person, "person")
  
  # Create a new message object of type "Person" and fill it with some data.
  person = person_pb2.Person()
  person.name        = "Max"
  person.stype       = person_pb2.Person.MALE
  person.email       = "max@mail.net"
  person.dog.name    = "Brandy"
  person.house.rooms = 4
  
  loop_count = 0
  # Creating an infinite publish-loop.
  # eCAL Supports a stop signal; when an eCAL Process is stopped, ecal_core.ok() will return false.
  while ecal_core.ok():
    # Change something in the message so we can see that the message is changing, e.g. in eCAL Monitor.
    loop_count = loop_count + 1
    person.id = loop_count
    
    # Send the content to other eCAL Processes that have subscribed to the topic "hello".
    pub.send(person)
    
    print("Sending: {}".format(person))
    
    # Sleep for 500 ms so we send with a frequency of 2 Hz.
    time.sleep(0.5)
  
  # Finalize eCAL.
  # You should always do that before your application exits.
  ecal_core.finalize()

if __name__ == "__main__":
  main()

import os
import sys
import time

# import the eCAL core API
import ecal.core.core as ecal_core
# import the eCAL publisher API supporting protobuf
from ecal.core.publisher import ProtoPublisher

sys.path.insert(1, os.path.join(sys.path[0], '../_protobuf'))
# import the compiled protobuf message classes
import person_pb2

def main():
  print("--------------------------------")
  print(" Python (legacy): PERSON SENDER ")
  print("--------------------------------")

  # Initialize eCAL. You always have to initialize eCAL before using it.
  # The name of our eCAL Process will be "person send python".
  # This name will be visible in the eCAL Monitor, once the process is running.
  ecal_core.initialize("person send python (legacy)")
  
  # Print used eCAL version and date
  print("eCAL {} ({})\n".format(ecal_core.getversion(), ecal_core.getdate()))

  # Set the state for the program.
  # You can vary between different states like healthy, warning, critical ...
  # This can be used to communicate the application state to applications like eCAL Monitor/Sys.
  ecal_core.set_process_state(1, 1, "I feel good!")

  # Creating the eCAL Publisher. An eCAL Process can create multiple publishers (and subscribers).
  # The topic we are going to publish is called "person".
  # Furthermore, we want to tell the publisher that we want to use the protobuf message class "Person" from the file "person.proto".
  pub = ProtoPublisher("person", person_pb2.Person)
  
  # Create a new message object of type "Person" and fill it with some data.
  person = person_pb2.Person()
  person.name        = "Max"
  person.stype       = person_pb2.Person.MALE
  person.email       = "max@mail.net"
  person.dog.name    = "Brandy"
  person.house.rooms = 4
  
  # Creating an inifite publish-loop.
  # eCAL Supports a stop signal; when an eCAL Process is stopped, eCAL::Ok() will return false.
  loop_count = 0
  while ecal_core.ok():
    
    # Change something in the message so we can see that the message is changing, e.g. in eCAL Monitor.
    loop_count = loop_count + 1
    person.id = loop_count
  
    # Send the content to other eCAL Processes that have subscribed to the topic "person".
    pub.send(person)
    print("Sending person {}".format(person.id))
  
    # Sleep for 500 ms so we send with a frequency of 2 Hz.
    time.sleep(0.5)
  
  # Finalize eCAL.
  # You should always do that before your application exits.
  ecal_core.finalize()
  
if __name__ == "__main__":
  main()  


├─  C++
│  └─  person_send.cpp
│
├─  C#
│  └─  person_send_csharp.cs
│
├─  Python
|  └─  person_send.py
|
└─  Python (legacy)
   └─  person_send.py

7.2.3.3. Person Subscriber#

For the subscriber the same changes apply as for the publisher.

// Including the eCAL convenience header
#include <ecal/ecal.h>
// In addition we include the msg protobuf publisher
#include <ecal/msg/protobuf/subscriber.h>

#include <iostream>

#include "person.pb.h"

/*
  Here we create the subscriber callback function that is called everytime,
  when a new message arrived from a publisher.
*/
void OnPerson(const eCAL::STopicId& topic_id_, const pb::People::Person& person_, const long long time_, const long long clock_)
{
  std::cout << "------------------------------------------"                    << "\n";
  std::cout << " Received Protobuf message in C++ "                            << "\n";
  std::cout << "------------------------------------------"                    << "\n";
  std::cout << " topic name   : " << topic_id_.topic_name                      << "\n";
  std::cout << " topic time   : " << time_                                     << "\n";
  std::cout << " topic clock  : " << clock_                                    << "\n";
  std::cout << ""                                                              << "\n";
  std::cout << " Content of message type \"" << person_.GetTypeName()  << "\"" << "\n";
  std::cout << "------------------------------------------"                    << "\n";
  std::cout << " id          : " << person_.id()                               << "\n";
  std::cout << " name        : " << person_.name()                             << "\n";
  std::cout << " stype       : " << person_.stype()                            << "\n";
  std::cout << " email       : " << person_.email()                            << "\n";
  std::cout << " dog.name    : " << person_.dog().name()                       << "\n";
  std::cout << " house.rooms : " << person_.house().rooms()                    << "\n";
  std::cout << "------------------------------------------"                    << "\n";
  std::cout                                                                    << "\n";
}

int main()
{
  std::cout << "----------------------" << std::endl;
  std::cout << " C++: PERSON RECEIVER"  << std::endl;
  std::cout << "----------------------" << std::endl;

  /*
    Initialize eCAL. You always have to initialize eCAL before using its API.
    The name of our eCAL Process will be "person receive". 
    This name will be visible in the eCAL Monitor, once the process is running.
  */
  eCAL::Initialize("person receive");

  /*
    Print some eCAL version information.
  */
  std::cout << "eCAL " << eCAL::GetVersionString() << " (" << eCAL::GetVersionDateString() << ")" << "\n";

  /*
    Set the state for the program.
    You can vary between different states like healthy, warning, critical ...
    This can be used to communicate the application state to applications like eCAL Monitor/Sys.
  */
  eCAL::Process::SetState(eCAL::Process::eSeverity::healthy, eCAL::Process::eSeverityLevel::level1, "I feel good!");

  /*
    Creating the eCAL Subscriber. An eCAL Process can create multiple subscribers (and publishers).
    The topic we are going to receive is called "person".
    The data type is "Pb.People.Person", generated from the protobuf definition.
  */
  eCAL::protobuf::CSubscriber<pb::People::Person> subscriber("person");

  /*
    Create and register a receive callback. The callback will be called whenever a new message is received.
  */
  subscriber.SetReceiveCallback(&OnPerson);

  /*
    Creating an infinite loop.
    eCAL Supports a stop signal; when an eCAL Process is stopped, eCAL::Ok() will return false.
  */
  while(eCAL::Ok())
  {
    /*
      Sleep for 500ms to avoid busy waiting.
    */
    eCAL::Process::SleepMS(500);
  }

  /*
    Finalize eCAL. This will stop all eCAL processes and free all resources.
    You should always finalize eCAL before exiting your application.
  */
  eCAL::Finalize();
  
  return(0);
}

using System;
// include ecal core namespace
using Eclipse.eCAL.Core;

public class PersonReceive
{
  public static void Main()
  {
    Console.WriteLine("---------------------");
    Console.WriteLine(" C#: PERSON RECEIVER");
    Console.WriteLine("---------------------");

    /*
      Initialize eCAL. You always have to initialize eCAL before using its API.
      The name of our eCAL Process will be "person receive c#". 
      This name will be visible in the eCAL Monitor, once the process is running.
    */
    Core.Initialize("person receive c#");

    /*
      Print version info.
    */
    Console.WriteLine(String.Format("eCAL {0} ({1})\n", Core.GetVersionString(), Core.GetVersionDateString()));

    /*
      Set the state for the program.
      You can vary between different states like healthy, warning, critical ...
      This can be used to communicate the application state to applications like eCAL Monitor/Sys.
    */
    Process.SetState(eProcessSeverity.Healthy, eProcessSeverityLevel.Level1, "I feel good!");

    /*
      Creating the eCAL Subscriber. An eCAL Process can create multiple subscribers (and publishers).
      The topic we are going to receive is called "person".
      The data type is "Pb.People.Person", generated from the protobuf definition.
    */
    var subscriber = new ProtobufSubscriber<Pb.People.Person>("person");

    /*
      Create and register a receive callback. The callback will be called whenever a new message is received.
    */
    subscriber.SetReceiveCallback((topicId, dataTypeInfo, person) =>
    {
      Console.WriteLine("------------------------------------------");
      Console.WriteLine(" Received Protobuf message in C#");
      Console.WriteLine("------------------------------------------");
      Console.WriteLine(" topic name   : {0}", topicId.TopicName);
      Console.WriteLine(" topic time   : n/a");
      Console.WriteLine(" topic clock  : n/a");
      Console.WriteLine("");
      Console.WriteLine(" Content of message type \"{0}\"", person.Message.GetType());
      Console.WriteLine("------------------------------------------");
      Console.WriteLine(" id          : {0}", person.Message.Id);
      Console.WriteLine(" name        : {0}", person.Message.Name);
      Console.WriteLine(" stype       : {0}", person.Message.Stype);
      Console.WriteLine(" email       : {0}", person.Message.Email);
      Console.WriteLine(" dog.name    : {0}", person.Message.Dog.Name);
      Console.WriteLine(" house.rooms : {0}", person.Message.House.Rooms);
      Console.WriteLine("------------------------------------------");
      Console.WriteLine("");
    });

    /*
      Creating an infinite loop.
      eCAL Supports a stop signal; when an eCAL Process is stopped, eCAL::Ok() will return false.
    */
    while (Core.Ok())
    {
      /*
        Sleep for 500ms to avoid busy waiting.
      */
      System.Threading.Thread.Sleep(500);
    }

    /*
      Cleanup. Dispose the subscriber to free resources.
    */
    subscriber.Dispose();

    /*
      Terminate eCAL. This will stop all eCAL processes and free all resources.
      You should always terminate eCAL before exiting your application.
    */
    Core.Terminate();
  }
}

import os
import sys
import time
import typing
# import the eCAL core API
import ecal.nanobind_core as ecal_core
# import the eCAL subscriber API
from ecal.msg.proto.core import Subscriber as ProtobufSubscriber
from ecal.msg.common.core import ReceiveCallbackData

import google.protobuf
from packaging import version
proto_version = version.parse(google.protobuf.__version__)
sys.path.insert(1, os.path.join(sys.path[0], f"./_protobuf/v{proto_version.major}"))
# import the compiled protobuf message classes
import person_pb2

# eCAL receive callback
def data_callback(publisher_id : ecal_core.TopicId, data : ReceiveCallbackData[person_pb2.Person]) -> None:
  output = f"""
  ----------------------------------------------
   Received person message from topic {publisher_id.topic_name} in Python
  ----------------------------------------------
  Time         : {data.send_timestamp}
  Clock        : {data.send_clock}
  Message      : 
    {data.message}
  """
  print(output)

def main():
  print("--------------------------------")
  print(" Python:  PERSON RECEIVE        ")
  print("--------------------------------")

  # Initialize eCAL. You always have to initialize eCAL before using its API.
  # The name of our eCAL Process will be "hello receive python". 
  # This name will be visible in the eCAL Monitor, once the process is running.
  ecal_core.initialize("person receive python")

  # print eCAL version and date
  print("eCAL {} ({})\n".format(ecal_core.get_version_string(), ecal_core.get_version_date_string()))

  # Set the state for the program.
  # You can vary between different states like healthy, warning, critical ...
  # This can be used to communicate the application state to applications like eCAL Monitor/Sys.
  ecal_core.process.set_state(ecal_core.process.Severity.HEALTHY, ecal_core.process.SeverityLevel.LEVEL1, "I feel good!")

  # Creating the eCAL Subscriber. An eCAL Process can create multiple subscribers (and publishers).
  # The topic we are going to receive is called "person".
  sub = ProtobufSubscriber[person_pb2.Person](person_pb2.Person, "person")
  sub.set_receive_callback(data_callback)
  
  # Creating an infinite loop.
  # eCAL Supports a stop signal; when an eCAL Process is stopped, eCAL::Ok() will return false.
  while ecal_core.ok():
    time.sleep(0.5) # Sleep for 500ms to reduce CPU usage.
  
  # Finalize eCAL.
  # You should always do that before your application exits.
  ecal_core.finalize()
  
if __name__ == "__main__":
  main()

import os
import sys
import time
# import the eCAL core API
import ecal.core.core as ecal_core
# import the eCAL subscriber API
from ecal.core.subscriber import ProtoSubscriber

sys.path.insert(1, os.path.join(sys.path[0], '../_protobuf'))
# import the protobuf generated classes
import person_pb2

# Create here the eCAL receive callback. This will be called whenever a new message is received.
def callback(topic_name, person, time):
  print("")
  print("Received person ..")
  print("person id    : {}".format(person.id))
  print("person name  : {}".format(person.name))
  print("person stype : {}".format(person.stype))
  print("person email : {}".format(person.email))
  print("dog.name     : {}".format(person.dog.name))
  print("house.rooms  : {}".format(person.house.rooms))

def main():
  print("----------------------------------")
  print(" Python (legacy): PERSON RECEIVER ")
  print("----------------------------------")

  # Initialize eCAL. You always have to initialize eCAL before using it.
  # The name of our eCAL Process will be "person receive python (legacy)".
  # This name will be visible in the eCAL Monitor, once the process is running.
  ecal_core.initialize("person receive python (legacy)")

  # Print used eCAL version and date
  print("eCAL {} ({})\n".format(ecal_core.getversion(), ecal_core.getdate()))
  
  # Set the state for the program.
  # You can vary between different states like healthy, warning, critical ...
  # This can be used to communicate the application state to applications like eCAL Monitor/Sys.
  ecal_core.set_process_state(1, 1, "I feel good!")
  
  # Creating the eCAL Subscriber. An eCAL Process can create multiple subscribers (and publishers).
  # The topic we are going to receive is called "person".
  # The topic type is specified by the protobuf message class.
  sub = ProtoSubscriber("person", person_pb2.Person)
  
  # Register the callback with the subscriber so it can be called.
  # The callback will be called whenever a new message is received.
  sub.set_callback(callback)
  
  # Creating an infinite loop.
  # eCAL Supports a stop signal; when an eCAL Process is stopped, eCAL::Ok() will return false.
  while ecal_core.ok():
    # Sleep for 500ms to avoid busy waiting.
    # You can use eCAL::Process::SleepMS() to sleep in milliseconds.
    time.sleep(0.5)
  
  # Finalize eCAL.
  # You should always do that before your application exits.
  ecal_core.finalize()
  
if __name__ == "__main__":
  main()


├─  C++
│  └─  person_receive.cpp
│
├─  C#
│  └─  person_receive_csharp.cs
│
├─  Python
|  └─  person_receive.py
|
└─  Python (legacy)
   └─  person_receive.py

7.2.3.4. Person Dynamic Subscriber#

Using eCAL and Protobuf, it is possible to receive data without knowing the structure of the incoming data in advance. Hence you can use a dynamic subscriber to receive Protobuf data, even if you do not have access to the corresponding .proto file. This is useful for generic applications, such as the eCAL Monitor, which can display all data types without knowing them in advance.

The dynamic Protobuf API is unfortunately only available for C++ and Python at the moment. C# does not have the capabilities for dynamic message support in Protocol Buffers at this time. Progress on this feature can be tracked in the following GitHub issue.

#include <ecal/ecal.h>
#include <ecal/msg/protobuf/dynamic_subscriber.h>

#include <iostream>

const std::string MESSAGE_NAME("person");

void ProcValue(const std::string& group_, const std::string& name_, const double value_, size_t index_)
{
  std::string var_name;
  if(!group_.empty()) var_name += group_ + ".";
  var_name += name_;
  if(index_ > 0) var_name += "[" + std::to_string(index_) + "]";
  std::cout << var_name << " : " << value_ << std::endl;
}

void ProcString(const std::string& group_, const std::string& name_, const std::string& value_, size_t index_)
{
  std::string var_name;
  if(!group_.empty()) var_name += group_ + ".";
  var_name += name_;
  if(index_ > 0) var_name += "[" + std::to_string(index_) + "]";
  std::cout << var_name << " : " << value_ << std::endl;
}

void ProcProtoType(const std::string& group_, const std::string& name_, google::protobuf::int32 value_, size_t index_)
{
  ProcValue(group_, name_, double(value_), index_);
}

void ProcProtoType(const std::string& group_, const std::string& name_, google::protobuf::int64 value_, size_t index_)
{
  ProcValue(group_, name_, double(value_), index_);
}

void ProcProtoType(const std::string& group_, const std::string& name_, google::protobuf::uint32 value_, size_t index_)
{
  ProcValue(group_, name_, double(value_), index_);
}

void ProcProtoType(const std::string& group_, const std::string& name_, google::protobuf::uint64 value_, size_t index_)
{
  ProcValue(group_, name_, double(value_), index_);
}

void ProcProtoType(const std::string& group_, const std::string& name_, float value_, size_t index_)
{
  ProcValue(group_, name_, double(value_), index_);
}

void ProcProtoType(const std::string& group_, const std::string& name_, double value_, size_t index_)
{
  ProcValue(group_, name_, double(value_), index_);
}

void ProcProtoType(const std::string& group_, const std::string& name_, bool value_, size_t index_)
{
  ProcValue(group_, name_, double(value_), index_);
}

void ProcProtoType(const std::string& group_, const std::string& name_, const std::string& value_, size_t index_)
{
  ProcString(group_, name_, value_, index_);
}

void ProcProtoType(const std::string& group_, const std::string& name_, const google::protobuf::EnumValueDescriptor* value_, size_t index_)
{
  ProcValue(group_, name_, double(value_->number()), index_);
}

void ProcProtoMsg(const google::protobuf::Message& msg_, const std::string& prefix_ /* = "" */)
{
  int count = msg_.GetDescriptor()->field_count();
  const google::protobuf::Reflection* ref_ptr = msg_.GetReflection();

  if(ref_ptr)
  {
    for (int i = 0; i < count; ++i)
    {
      auto field = msg_.GetDescriptor()->field(i);
  
      const google::protobuf::FieldDescriptor::CppType fdt = field->cpp_type();
      switch(fdt)
      {
      case google::protobuf::FieldDescriptor::CPPTYPE_INT32:      // TYPE_INT32, TYPE_SINT32, TYPE_SFIXED32
        if(field->is_repeated())
        {
          int fsize = ref_ptr->FieldSize(msg_, field);
          for(int fnum = 0; fnum < fsize; ++fnum)
          {
            ProcProtoType(prefix_, field->name(), ref_ptr->GetRepeatedInt32(msg_, field, fnum), static_cast<size_t>(fnum));
          }
        }
        else
        {
          ProcProtoType(prefix_, field->name(), ref_ptr->GetInt32(msg_, field), 0);
        }
        break;
      case google::protobuf::FieldDescriptor::CPPTYPE_INT64:      // TYPE_INT64, TYPE_SINT64, TYPE_SFIXED64
        if(field->is_repeated())
        {
          int fsize = ref_ptr->FieldSize(msg_, field);
          for(int fnum = 0; fnum < fsize; ++fnum)
          {
            ProcProtoType(prefix_, field->name(), ref_ptr->GetRepeatedInt64(msg_, field, fnum), static_cast<size_t>(fnum));
          }
        }
        else
        {
          ProcProtoType(prefix_, field->name(), ref_ptr->GetInt64(msg_, field), 0);
        }
        break;
      case google::protobuf::FieldDescriptor::CPPTYPE_UINT32:     // TYPE_UINT32, TYPE_FIXED32
        if(field->is_repeated())
        {
          int fsize = ref_ptr->FieldSize(msg_, field);
          for(int fnum = 0; fnum < fsize; ++fnum)
          {
            ProcProtoType(prefix_, field->name(), ref_ptr->GetRepeatedUInt32(msg_, field, fnum), static_cast<size_t>(fnum));
          }
        }
        else
        {
          ProcProtoType(prefix_, field->name(), ref_ptr->GetUInt32(msg_, field), 0);
        }
        break;
      case google::protobuf::FieldDescriptor::CPPTYPE_UINT64:     // TYPE_UINT64, TYPE_FIXED64
        if(field->is_repeated())
        {
          int fsize = ref_ptr->FieldSize(msg_, field);
          for(int fnum = 0; fnum < fsize; ++fnum)
          {
            ProcProtoType(prefix_, field->name(), ref_ptr->GetRepeatedUInt64(msg_, field, fnum), static_cast<size_t>(fnum));
          }
        }
        else
        {
          ProcProtoType(prefix_, field->name(), ref_ptr->GetUInt64(msg_, field), 0);
        }
        break;
      case google::protobuf::FieldDescriptor::CPPTYPE_DOUBLE:     // TYPE_DOUBLE
        if(field->is_repeated())
        {
          int fsize = ref_ptr->FieldSize(msg_, field);
          for(int fnum = 0; fnum < fsize; ++fnum)
          {
            ProcProtoType(prefix_, field->name(), ref_ptr->GetRepeatedDouble(msg_, field, fnum), static_cast<size_t>(fnum));
          }
        }
        else
        {
          ProcProtoType(prefix_, field->name(), ref_ptr->GetDouble(msg_, field), 0);
        }
        break;
      case google::protobuf::FieldDescriptor::CPPTYPE_FLOAT:      // TYPE_FLOAT
        if(field->is_repeated())
        {
          int fsize = ref_ptr->FieldSize(msg_, field);
          for(int fnum = 0; fnum < fsize; ++fnum)
          {
            ProcProtoType(prefix_, field->name(), ref_ptr->GetRepeatedFloat(msg_, field, fnum), static_cast<size_t>(fnum));
          }
        }
        else
        {
          ProcProtoType(prefix_, field->name(), ref_ptr->GetFloat(msg_, field), 0);
        }
        break;
      case google::protobuf::FieldDescriptor::CPPTYPE_BOOL:       // TYPE_BOOL
        if(field->is_repeated())
        {
          int fsize = ref_ptr->FieldSize(msg_, field);
          for(int fnum = 0; fnum < fsize; ++fnum)
          {
            ProcProtoType(prefix_, field->name(), ref_ptr->GetRepeatedBool(msg_, field, fnum), static_cast<size_t>(fnum));
          }
        }
        else
        {
          ProcProtoType(prefix_, field->name(), ref_ptr->GetBool(msg_, field), 0);
        }
        break;
      case google::protobuf::FieldDescriptor::CPPTYPE_ENUM:       // TYPE_ENUM
        if(field->is_repeated())
        {
          int fsize = ref_ptr->FieldSize(msg_, field);
          for(int fnum = 0; fnum < fsize; ++fnum)
          {
            ProcProtoType(prefix_, field->name(), ref_ptr->GetRepeatedEnum(msg_, field, fnum), static_cast<size_t>(fnum));
          }
        }
        else
        {
          ProcProtoType(prefix_, field->name(), ref_ptr->GetEnum(msg_, field), 0);
        }
        break;
      case google::protobuf::FieldDescriptor::CPPTYPE_STRING:     // TYPE_STRING, TYPE_BYTES
        if(field->is_repeated())
        {
          int fsize = ref_ptr->FieldSize(msg_, field);
          for(int fnum = 0; fnum < fsize; ++fnum)
          {
            ProcProtoType(prefix_, field->name(), ref_ptr->GetRepeatedString(msg_, field, fnum), fnum);
          }
        }
        else
        {
          ProcProtoType(prefix_, field->name(), ref_ptr->GetString(msg_, field), 0);
        }
        break;
      case google::protobuf::FieldDescriptor::CPPTYPE_MESSAGE:    // TYPE_MESSAGE, TYPE_GROUP
        {
          if(field->is_repeated())
          {
            int fsize = ref_ptr->FieldSize(msg_, field);
            for(int fnum = 0; fnum < fsize; ++fnum)
            {
              const google::protobuf::Message& msg = ref_ptr->GetRepeatedMessage(msg_, field, fnum);
              std::string prefix = field->name();
              prefix += "[";
              prefix += std::to_string(fnum);
              prefix += "]";
              if(!prefix_.empty()) prefix = prefix_ + "." + prefix;

              // do not process default messages to avoid infinite recursions.
              std::vector<const google::protobuf::FieldDescriptor*> msg_fields;
              msg.GetReflection()->ListFields(msg, &msg_fields);
              
              if (prefix_.find(field->name()) == std::string::npos || !msg_fields.empty())
                ProcProtoMsg(msg, prefix);
            }
          }
          else
          {
            const google::protobuf::Message& msg = ref_ptr->GetMessage(msg_, field);
            std::string prefix = field->name();
            if(!prefix_.empty()) prefix = prefix_ + "." + prefix;

            // do not process default messages to avoid infinite recursions.
            std::vector<const google::protobuf::FieldDescriptor*> msg_fields;
            msg.GetReflection()->ListFields(msg, &msg_fields);

            if (prefix_.find(field->name()) == std::string::npos || !msg_fields.empty())
              ProcProtoMsg(msg, prefix);
          }
        }
        break;
      default:
        break;
      }
    }
  }
}

void ProtoMsgCallback(const eCAL::STopicId& topic_id_, const std::shared_ptr<google::protobuf::Message>& msg_)
{
  ProcProtoMsg(*msg_, topic_id_.topic_name);
  std::cout << std::endl;
}

int main()
{
  // initialize eCAL API
  eCAL::Initialize("proto_dyn");

  // create dynamic subscribers for receiving and decoding messages
  eCAL::protobuf::CDynamicSubscriber sub(MESSAGE_NAME);
  sub.SetReceiveCallback(std::bind(ProtoMsgCallback, std::placeholders::_1, std::placeholders::_2));

  // enter main loop
  while(eCAL::Ok())
  {
    // sleep main thread for 1 second
    eCAL::Process::SleepMS(1000);
  }

  // finalize eCAL API
  eCAL::Finalize();

  return(0);
}

import sys
import time
import typing
# import the eCAL core API
import ecal.nanobind_core as ecal_core
# import the eCAL subscriber API
from ecal.msg.proto.core import DynamicSubscriber as ProtobufDynamicSubscriber
from ecal.msg.common.core import ReceiveCallbackData

# eCAL receive callback
def data_callback(publisher_id : ecal_core.TopicId, data : ReceiveCallbackData[typing.Any]) -> None:
  output = f"""
  ----------------------------------------------
   Received dynamic person message from topic {publisher_id.topic_name} in Python
  ----------------------------------------------
  Time         : {data.send_timestamp}
  Clock        : {data.send_clock}
  Message      : 
    {data.message}
  """
  print(output)

def main():
  print("--------------------------------")
  print(" Python: DYNAMIC PERSON RECEIVE ")
  print("--------------------------------")

  # Initialize eCAL. You always have to initialize eCAL before using its API.
  # The name of our eCAL Process will be "hello receive python". 
  # This name will be visible in the eCAL Monitor, once the process is running.
  ecal_core.initialize("dynamic person receive python")

  # print eCAL version and date
  print("eCAL {} ({})\n".format(ecal_core.get_version_string(), ecal_core.get_version_date_string()))

  # Set the state for the program.
  # You can vary between different states like healthy, warning, critical ...
  # This can be used to communicate the application state to applications like eCAL Monitor/Sys.
  ecal_core.process.set_state(ecal_core.process.Severity.HEALTHY, ecal_core.process.SeverityLevel.LEVEL1, "I feel good!")

  # Creating the eCAL Subscriber. An eCAL Process can create multiple subscribers (and publishers).
  # The topic we are going to receive is called "person".
  sub = ProtobufDynamicSubscriber("person")
  sub.set_receive_callback(data_callback)
  
  # Creating an infinite loop.
  # eCAL Supports a stop signal; when an eCAL Process is stopped, eCAL::Ok() will return false.
  while ecal_core.ok():
    time.sleep(0.5) # Sleep for 500ms to reduce CPU usage.
  
  # Finalize eCAL.
  # You should always do that before your application exits.
  ecal_core.finalize()
  
if __name__ == "__main__":
  main()


├─  C++
│  └─  person_receive.cpp
│
└─  Python
   └─  person_receive.py

Protobuf: Person

Contents

7.2.3. Protobuf: Person#

7.2.3.1. Person Protobuf#

7.2.3.2. Person Publisher#

7.2.3.3. Person Subscriber#

7.2.3.4. Person Dynamic Subscriber#