dynsub

The dynsub example is a PoC for a C-based JSON printer of arbitrary data. It assumes that topic discovery is enabled, but doesn’t require it.

Running the example

Pass the name of a topic to dynsub and it waits for a writer of that topic to show up. When it finds one in the DCPSPublication topic, it tries to subscribe and print the received samples as JSON.

For example: Start the HelloworldPublisher in one shell:

# bin/HelloworldPublisher
=== [Publisher]  Waiting for a reader to be discovered ...

In another shell start dynsub:

# bin/dynsub HelloWorldData_Msg
{"userID":1,"message":"Hello World"}
{"userID":1}

The second line is the “invalid sample” generated because of the termination of the publisher. In Cyclone DDS, only the key fields are valid, and therefore printed.

Instead of the HelloWorldData_Msg, the small publisher program “variouspub” can publish a number of different types. Pass it the name of the type to publish. For example:

# bin/variouspub B

This publishes samples at 1Hz until killed.

Source code

dynsub.c

// Copyright(c) 2022 to 2023 ZettaScale Technology and others
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License v. 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0, or the Eclipse Distribution License
// v. 1.0 which is available at
// http://www.eclipse.org/org/documents/edl-v10.php.
//
// SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include <locale.h>
#include <signal.h>

#include "dds/dds.h"
#include "dynsub.h"

// Interpreting the data of an arbitrary topic requires interpreting the type object that describes the data.
// The type object type is defined by the XTypes specification (https://www.omg.org/spec/DDS-XTypes/) and it
// comes in two forms: MinimalTypeObject and CompleteTypeObject.  Only the latter includes field names, so
// that's what need.
//
// Cyclone DDS includes a copy of the IDL as well as the corresponding type definitions in C as generated by
// IDLC.  So instead of including yet another copy, we simply refer to those.  These files are not (yet?)
// part of the stable API of Cyclone DDS and so the updates to Cyclone may change the locations or the names
// of the relevant header files.
//
// The API uses `dds_typeobj_t` and `dds_typeinfo_t` that are opaque types but really amount to the
// corresponding XTypes objects: DDS_XTypes_TypeObject and DDS_XTypes_TypeInformation.  Rather than casting
// pointers like we do here, they should be defined in a slightly different way so that they are not really
// opaque.  For now, this'll have to do.
#include "dds/ddsc/dds_public_alloc.h"
#include "dds/ddsi/ddsi_sertype.h"
#include "dds/ddsi/ddsi_xt_typeinfo.h"

#include "dds/ddsrt/threads.h"
#include "dds/ddsi/ddsi_serdata.h"

// For convenience, the DDS participant is global
static dds_entity_t participant;

static dds_entity_t termcond;

// Helper function to wait for a DCPSPublication/DCPSSubscription to show up with the desired topic name,
// then calls dds_find_topic to create a topic for that data writer's/reader's type up the retrieves the
// type object.
static dds_return_t get_topic_and_typeobj (const char *topic_name, dds_duration_t timeout, dds_entity_t *topic, DDS_XTypes_TypeObject **xtypeobj)
{
  const dds_entity_t waitset = dds_create_waitset (participant);
  const dds_entity_t dcpspublication_reader = dds_create_reader (participant, DDS_BUILTIN_TOPIC_DCPSPUBLICATION, NULL, NULL);
  const dds_entity_t dcpspublication_readcond = dds_create_readcondition (dcpspublication_reader, DDS_ANY_STATE);
  const dds_entity_t dcpssubscription_reader = dds_create_reader (participant, DDS_BUILTIN_TOPIC_DCPSSUBSCRIPTION, NULL, NULL);
  const dds_entity_t dcpssubscription_readcond = dds_create_readcondition (dcpssubscription_reader, DDS_ANY_STATE);
  (void) dds_waitset_attach (waitset, dcpspublication_readcond, dcpspublication_reader);
  (void) dds_waitset_attach (waitset, dcpssubscription_readcond, dcpssubscription_reader);
  const dds_time_t abstimeout = (timeout == DDS_INFINITY) ? DDS_NEVER : dds_time () + timeout;
  dds_return_t ret = DDS_RETCODE_OK;
  *xtypeobj = NULL;
  struct ppc ppc;
  ppc_init (&ppc);
  dds_attach_t triggered_reader_x;
  while (*xtypeobj == NULL && dds_waitset_wait_until (waitset, &triggered_reader_x, 1, abstimeout) > 0)
  {
    void *epraw = NULL;
    dds_sample_info_t si;
    dds_entity_t triggered_reader = (dds_entity_t) triggered_reader_x;
    if (dds_take (triggered_reader, &epraw, &si, 1, 1) <= 0)
      continue;
    dds_builtintopic_endpoint_t *ep = epraw;
    const dds_typeinfo_t *typeinfo = NULL;
    // We are only interested in DCPSPublications where the topic name matches and that carry type information
    // (a non-XTypes capable DDS would not provide type information) because without that information there is
    // no way we can do anything interesting with it.
    if (strcmp (ep->topic_name, topic_name) == 0 && dds_builtintopic_get_endpoint_type_info (ep, &typeinfo) == 0 && typeinfo)
    {
      // Using dds_find_topic allows us to "clone" the topic definition including the topic QoS, but it does
      // require that topic discovery is enabled in the configuration.  The advantage of using dds_find_topic
      // is that it creates a topic with the same name, type *and QoS*.  That distinction only matters if
      // topic is discovery is enabled and/or if the topic has a durability kind of of transient or persistent:
      // - using a different topic QoS might result in an incompatible QoS notification if topic discovery is
      //   enabled (everything would still work).
      // - transient/persistent data behaviour is defined in terms of the topic QoS actually really matters
      //
      // So we try to use dds_find_topic, and if that fails, try to go the other route using the writer's QoS
      // as an approximation of the topic QoS.
      if ((*topic = dds_find_topic (DDS_FIND_SCOPE_GLOBAL, participant, ep->topic_name, typeinfo, DDS_SECS (2))) < 0)
      {
        fprintf (stderr, "dds_find_topic: %s ... continuing on the assumption that topic discovery is disabled\n", dds_strretcode (*topic));
        dds_topic_descriptor_t *descriptor;
        if ((ret = dds_create_topic_descriptor(DDS_FIND_SCOPE_GLOBAL, participant, typeinfo, DDS_SECS (10), &descriptor)) < 0)
        {
          fprintf (stderr, "dds_create_topic_descriptor: %s\n", dds_strretcode (ret));
          dds_return_loan (triggered_reader, &epraw, 1);
          goto error;
        }
        dds_qset_data_representation (ep->qos, 0, NULL);
        if ((*topic = dds_create_topic (participant, descriptor, ep->topic_name, ep->qos, NULL)) < 0)
        {
          fprintf (stderr, "dds_create_topic_descriptor: %s (be sure to enable topic discovery in the configuration)\n", dds_strretcode (*topic));
          dds_delete_topic_descriptor (descriptor);
          dds_return_loan (triggered_reader, &epraw, 1);
          goto error;
        }
        dds_delete_topic_descriptor (descriptor);
      }
      // The topic suffices for creating a reader, but we also need the TypeObject to make sense of the data
      if ((*xtypeobj = load_type_with_deps (participant, typeinfo, &ppc)) == NULL)
      {
        fprintf (stderr, "loading type with all dependencies failed\n");
        dds_return_loan (triggered_reader, &epraw, 1);
        goto error;
      }
      if (load_type_with_deps_min (participant, typeinfo, &ppc) == NULL)
      {
        fprintf (stderr, "loading minimal type with all dependencies failed\n");
        dds_return_loan (triggered_reader, &epraw, 1);
        goto error;
      }
    }
    dds_return_loan (triggered_reader, &epraw, 1);
  }
  if (*xtypeobj)
  {
    // If we got the type object, populate the type cache
    size_t align, size;
    build_typecache_to (&(*xtypeobj)->_u.complete, &align, &size);
    fflush (stdout);
    struct typeinfo templ = { .key = { .key = (uintptr_t) *xtypeobj } } , *info;
    if ((info = type_cache_lookup (&templ)) != NULL)
    {
      assert (info->release == NULL);
      info->release = *xtypeobj;
    }
    else
    {
      // not sure whether this is at all possible
      info = malloc (sizeof (*info));
      assert (info);
      *info = (struct typeinfo){ .key = { .key = (uintptr_t) *xtypeobj }, .typeobj = &(*xtypeobj)->_u.complete, .release = *xtypeobj, .align = align, .size = size };
      type_cache_add (info);
    }
  }
error:
  dds_delete (dcpspublication_reader);
  dds_delete (dcpssubscription_reader);
  dds_delete (waitset);
  return (*xtypeobj != NULL) ? DDS_RETCODE_OK : DDS_RETCODE_TIMEOUT;
}

static bool print_sample_normal (dds_entity_t reader, const DDS_XTypes_TypeObject *xtypeobj)
{
  void *raw = NULL;
  dds_sample_info_t si;
  dds_return_t ret;
  if ((ret = dds_take (reader, &raw, &si, 1, 1)) < 0)
    return false;
  else if (ret != 0)
  {
    // ... that we then print
    print_sample (si.valid_data, raw, &xtypeobj->_u.complete);
    if (dds_return_loan (reader, &raw, 1) < 0)
      return false;
  }
  return true;
}

static void hexdump (const unsigned char *msg, const size_t len)
{
  for (size_t off16 = 0; off16 < len; off16 += 16)
  {
    printf ("%04" PRIxSIZE " ", off16);
    size_t off1;
    for (off1 = 0; off1 < 16 && off16 + off1 < len; off1++)
      printf ("%s %02x", (off1 == 8) ? " " : "", msg[off16 + off1]);
    for (; off1 < 16; off1++)
      printf ("%s   ", (off1 == 8) ? " " : "");
    printf ("  |");
    for (off1 = 0; off1 < 16 && off16 + off1 < len; off1++)
    {
      unsigned char c = msg[off16 + off1];
      printf ("%c", (c >= 32 && c < 127) ? c : '.');
    }
    printf ("|\n");
  }
  fflush (stdout);
}

static const char *encodingstr (const struct ddsi_serdata *sd)
{
  uint16_t encoding;
  ddsi_serdata_to_ser (sd, 0, 2, &encoding);
  switch (encoding)
  {
    case DDSI_RTPS_CDR_BE:
    case DDSI_RTPS_CDR_LE:
      return "CDR";
    case DDSI_RTPS_PL_CDR_BE:
    case DDSI_RTPS_PL_CDR_LE:
      return "PL_CDR";
    case DDSI_RTPS_CDR2_BE:
    case DDSI_RTPS_CDR2_LE:
      return "CDR2";
    case DDSI_RTPS_D_CDR2_BE:
    case DDSI_RTPS_D_CDR2_LE:
      return "D_CDR2";
    case DDSI_RTPS_PL_CDR2_BE:
    case DDSI_RTPS_PL_CDR2_LE:
      return "PL_CDR2";
    default:
      return "unknown";
  }
}

static bool print_sample_cdr (dds_entity_t reader, const DDS_XTypes_TypeObject *xtypeobj)
{
  // Note: doesn't print the exact CDR received, but the "normalised" one where byteswapping
  // has been performed, booleans have been mapped to 0 or 1, and perhaps some other similar
  // changes have been made.
  struct ddsi_serdata *sd = NULL;
  dds_sample_info_t si;
  dds_return_t ret;
  if ((ret = dds_takecdr (reader, &sd, 1, &si, 0)) < 0)
    return false;
  else if (ret != 0)
  {
    printf ("encoding: %s", encodingstr (sd));
    if (!si.valid_data)
      printf (" (expect XCDR2 because it is an invalid sample)");
    printf ("\n");
    if (ddsi_serdata_size (sd) == 4)
      printf ("(no payload)\n");
    else
    {
      ddsrt_iovec_t iov;
      struct ddsi_serdata *refsd;
      refsd = ddsi_serdata_to_ser_ref (sd, 4, ddsi_serdata_size (sd) - 4, &iov);
      hexdump (iov.iov_base, iov.iov_len);
      ddsi_serdata_to_ser_unref (refsd, &iov);
    }

    void *raw = calloc (1, sd->type->sizeof_type);
    if (raw == NULL)
      abort ();

    bool ok;
    if (si.valid_data)
      ok = ddsi_serdata_to_sample (sd, raw, NULL, NULL);
    else
    {
      const struct ddsi_sertype *st;
      dds_get_entity_sertype (reader, &st);
      ok = ddsi_serdata_untyped_to_sample (st, sd, raw, NULL, NULL);
    }
    if (ok)
      print_sample (si.valid_data, raw, &xtypeobj->_u.complete);
    else
      printf ("(conversion to sample failed)\n");
    ddsi_sertype_free_sample (sd->type, raw, DDS_FREE_CONTENTS);
    free (raw);

    ddsi_serdata_unref (sd);
  }
  return true;
}

#if !DDSRT_WITH_FREERTOS && !__ZEPHYR__
static void signal_handler (int sig)
{
  (void) sig;
  dds_set_guardcondition (termcond, true);
}
#endif

#if !_WIN32 && !DDSRT_WITH_FREERTOS && !__ZEPHYR__
static uint32_t sigthread (void *varg)
{
  sigset_t *set = varg;
  int sig;
  if (sigwait (set, &sig) == 0)
    signal_handler (sig);
  return 0;
}
#endif

int main (int argc, char **argv)
{
  dds_return_t ret = 0;
  dds_entity_t topic = 0;
  bool raw_mode;
  const char *topic_name;

  // for printf("%ls")
  setlocale (LC_CTYPE, "");

  if (argc == 2)
  {
    raw_mode = false;
    topic_name = argv[1];
  }
  else if (argc == 3 && strcmp (argv[1], "-r") == 0)
  {
    raw_mode = true;
    topic_name = argv[2];
  }
  else
  {
    fprintf (stderr, "usage: %s [-r] topicname\n", argv[0]);
    return 2;
  }

  participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
  if (participant < 0)
  {
    fprintf (stderr, "dds_create_participant: %s\n", dds_strretcode (participant));
    return 1;
  }

  // The one magic step: get a topic and type object ...
  DDS_XTypes_TypeObject *xtypeobj;
  type_cache_init ();
  if ((ret = get_topic_and_typeobj (topic_name, DDS_SECS (10), &topic, &xtypeobj)) < 0)
  {
    fprintf (stderr, "get_topic_and_typeobj: %s\n", dds_strretcode (ret));
    goto error;
  }
  // ... given those, we can create a reader just like we do normally ...
  const dds_entity_t reader = dds_create_reader (participant, topic, NULL, NULL);
  // ... and create a waitset that allows us to wait for any incoming data ...
  const dds_entity_t waitset = dds_create_waitset (participant);
  const dds_entity_t readcond = dds_create_readcondition (reader, DDS_ANY_STATE);
  (void) dds_waitset_attach (waitset, readcond, 0);

  termcond = dds_create_guardcondition (participant);
  (void) dds_waitset_attach (waitset, termcond, 0);

#ifdef _WIN32
  signal (SIGINT, signal_handler);
#elif !DDSRT_WITH_FREERTOS && !__ZEPHYR__
  ddsrt_thread_t sigtid;
  sigset_t sigset, osigset;
  sigemptyset (&sigset);
#ifdef __APPLE__
  DDSRT_WARNING_GNUC_OFF(sign-conversion)
#endif
  sigaddset (&sigset, SIGHUP);
  sigaddset (&sigset, SIGINT);
  sigaddset (&sigset, SIGTERM);
#ifdef __APPLE__
  DDSRT_WARNING_GNUC_ON(sign-conversion)
#endif
  sigprocmask (SIG_BLOCK, &sigset, &osigset);
  {
    ddsrt_threadattr_t tattr;
    ddsrt_threadattr_init (&tattr);
    ddsrt_thread_create (&sigtid, "sigthread", &tattr, sigthread, &sigset);
  }
#endif

  bool termflag = false;
  while (!termflag)
  {
    (void) dds_waitset_wait (waitset, NULL, 0, DDS_INFINITY);
    dds_read_guardcondition (termcond, &termflag);

    bool ok = raw_mode ? print_sample_cdr (reader, xtypeobj) : print_sample_normal (reader, xtypeobj);
    if (!ok)
      break;
  }

#if _WIN32
  signal_handler (SIGINT);
#elif !DDSRT_WITH_FREERTOS && !__ZEPHYR__
  {
    /* get the attention of the signal handler thread */
    void (*osigint) (int);
    void (*osigterm) (int);
    kill (getpid (), SIGTERM);
    ddsrt_thread_join (sigtid, NULL);
    osigint = signal (SIGINT, SIG_IGN);
    osigterm = signal (SIGTERM, SIG_IGN);
    sigprocmask (SIG_SETMASK, &osigset, NULL);
    signal (SIGINT, osigint);
    signal (SIGINT, osigterm);
  }
#endif

error:
  type_cache_free ();
  dds_delete (participant);
  return ret < 0;
}

variouspub_types.idl

struct A {
  @key
  string name;
  string message;
  unsigned long count;
};

struct T {
  short s;
  long l;
};

struct B {
  A a;
  sequence<T> ts;
};

struct C {
  B b;
  @key
  short k;
};

module M1 {
  @appendable
  struct O {
    @optional long x;
  };
};

struct D {
  wstring ws;
  wchar wc;
  unsigned long count;
};

struct U {
  uint32 w;
  @key string x;
  string y;
  @key uint32 z;
};

struct E {
  uint32 a;
  @key sequence<U> b[2];
  @key uint32 c;
};

variouspub.c

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <signal.h>
#include <assert.h>

#include "dds/dds.h"
#include "variouspub_types.h"

static void *samples_a[] = {
  &(A){ "Mariken",   "Wie sidi, vrient?", 0 },
  &(A){ "Die duvel", "Een meester vol consten,", 0 },
  &(A){ "Die duvel", "Nieuwers af falende, wes ic besta.", 0 },
  &(A){ "Mariken",   "'t Comt mi alleleens met wien dat ick ga,", 0 },
  &(A){ "Mariken",   "Also lief gae ic metten quaetsten als metten besten.", 0 },
  NULL
};

static void *samples_b[] = {
  &(B){ {"Die duvel", "Wildi u liefde te mi werts vesten,", 0},
        { ._length = 2, ._maximum = 2, ._release = false,
          ._buffer = (T[]){ {2,3},{5,7} } } },
  &(B){ {"Die duvel", "Ick sal u consten leeren sonder ghelijcke,", 0},
        { ._length = 3, ._maximum = 3, ._release = false,
          ._buffer = (T[]){ {11,13},{17,19},{23,29} } } },
  &(B){ {"Die duvel", "Die seven vrie consten: rethorijcke, musijcke,", 0},
        { ._length = 5, ._maximum = 5, ._release = false,
          ._buffer = (T[]){ {31,37},{41,43},{47,52},{59,61},{67,71} } } },
  &(B){ {"Die duvel", "Logica, gramatica ende geometrie,", 0},
        { ._length = 7, ._maximum = 7, ._release = false,
          ._buffer = (T[]){ {73,79},{83,89},{97,101},{103,107},{109,113},
                            {127,131},{137,139} } } },
  &(B){ {"Die duvel", "Aristmatica ende alkenie,", 0},
        { ._length = 11, ._maximum = 11, ._release = false,
          ._buffer = (T[]){ {149,151},{157,163},{167,173},{179,181},
                            {191,193},{197, 199},{211,223},{227,229},
                            {233,239},{241,251},{257,263} } } },
  NULL
};

static void *samples_c[] = {
  &(C){ { {"Die duvel", "Dwelc al consten sijn seer curable.", 0},
          { ._length = 13, ._maximum = 13, ._release = false,
            ._buffer = (T[]){ {269,271},{277,281},{283,293},{307,311},
                              {313,317},{331,337},{347,349},{353,359},
                              {367,373},{379,383},{389,397},{401,409},
                              {419,421} } } },
        8936 },
  &(C){ { {"Die duvel", "Noyt vrouwe en leefde op eerde so able", 0},
          { ._length = 17, ._maximum = 17, ._release = false,
            ._buffer = (T[]){ {431,433},{439,443},{449,457},{461,463},
                              {467,479},{487,491},{499,503},{509,521},
                              {523,541},{547,557},{563,569},{571,577},
                              {587,593},{599,601},{607,613},{617,619},
                              {631,641} } } },
        18088 },
  &(C){ { {"Die duvel", "Als ic u maken sal.", 0},
          { ._length = 19, ._maximum = 19, ._release = false,
            ._buffer = (T[]){ {643,647},{653,659},{661,673},{677,683},
                              {691,701},{709,719},{727,733},{739,743},
                              {751,757},{761,769},{773,787},{797,809},
                              {811,821},{823,827},{829,839},{853,857},
                              {859,863},{877,881},{883,887} } } },
        29172 },
  &(C){ { {"Mariken",  "So moetti wel zijn een constich man.", 0},
          { ._length = 23, ._maximum = 23, ._release = false,
            ._buffer = (T[]){ {907,911},{919,929},{937,941},{947,953},
                              {967,971},{977,983},{991,997},{1009,1013},
                              {1019,1021},{1031,1033},{1039,1049},{1051,1061},
                              {1063,1069},{1087,1091},{1093,1097},{1103,1109},
                              {1117,1123},{1129,1151},{1153,1163},{1171,1181},
                              {1187,1193},{1201,1213},{1217,1223} } } },
        16022 },
  &(C){ { {"Mariken",  "Wie sidi dan?", 0},
          { ._length = 29, ._maximum = 29, ._release = false,
            ._buffer = (T[]){ {1229,1231},{1237,1249},{1259,1277},{1279,1283},
                              {1289,1291},{1297,1301},{1303,1307},{1319,1321},
                              {1327,1361},{1367,1373},{1381,1399},{1409,1423},
                              {1427,1429},{1433,1439},{1447,1451},{1453,1459},
                              {1471,1481},{1483,1487},{1489,1493},{1499,1511},
                              {1523,1531},{1543,1549},{1553,1559},{1567,1571},
                              {1579,1583},{1597,1601},{1607,1609},{1613,1619},
                              {1621,1627} } } },
        17880 },
  NULL
};

static int32_t long_4 = 4;
static void *samples_M1_O[] = {
  &(M1_O){ .x = NULL },
  &(M1_O){ .x = &long_4 },
  NULL
};

static void *samples_d[] = {
  &(D){ L"😀 Een Kruyck gaat soo langh te water tot datse barst.", 0x2206, 0 },
  &(D){ L"🙃 Men treckt een Boogh soo lang tot datse stucken knarst.", 0x2207, 0 },
  &(D){ L"😊 De Steel-kunst doet zyn Meester de dood vaak verwerven.", 0x22a5, 0 },
  NULL
};

static void *samples_e[] = {
  &(E){ 0, {{0},{0}}, 456 },
  &(E){ 0, {{0},{._length=1, ._maximum=1, ._buffer=&(U){34,"aap","noot",56}}}, 456},
  &(E){ 0, {{0},{._length=1, ._maximum=1, ._buffer=&(U){78,"aap","zus",56}}}, 456},
  &(E){ 0, {{0},{._length=1, ._maximum=1, ._buffer=&(U){78,"wim","zus",90}}}, 456},
  NULL
};

static struct tpentry {
  const char *name;
  const dds_topic_descriptor_t *descr;
  void **samples;
  size_t count_offset;
} tptab[] = {
  { "A", &A_desc, samples_a, offsetof (A, count) },
  { "B", &B_desc, samples_b, offsetof (B, a.count) },
  { "C", &C_desc, samples_c, offsetof (C, b.a.count) },
  { "M1::O", &M1_O_desc, samples_M1_O, SIZE_MAX },
  { "D", &D_desc, samples_d, offsetof (D, count) },
  { "E", &E_desc, samples_e, offsetof (E, a) },
  { NULL, NULL, NULL, 0 }
};

static void usage (const char *argv0)
{
  fprintf (stderr, "usage: %s {", argv0);
  const char *sep = "";
  for (struct tpentry *tpentry = &tptab[0]; tpentry->name; tpentry++)
  {
    fprintf (stderr, "%s%s", sep, tpentry->name);
    sep = "|";
  }
  fprintf (stderr, "}\n");
  exit (2);
}

static volatile sig_atomic_t interrupted;

static void sigint (int sig)
{
  (void) sig;
  interrupted = 1;
}

int main (int argc, char **argv)
{
  if (argc != 2)
    usage (argv[0]);
  struct tpentry *tpentry;
  for (tpentry = &tptab[0]; tpentry->name; tpentry++)
    if (strcmp (tpentry->name, argv[1]) == 0)
      break;
  if (tpentry->name == NULL)
    usage (argv[0]);

  const dds_entity_t participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
  if (participant < 0)
  {
    fprintf (stderr, "dds_create_participant: %s\n", dds_strretcode (participant));
    return 1;
  }

  const dds_entity_t topic = dds_create_topic (participant, tpentry->descr, tpentry->name, NULL, NULL);
  const dds_entity_t writer = dds_create_writer (participant, topic, NULL, NULL);
  uint32_t sample_idx = 0;
  uint32_t count = 0;
  signal (SIGINT, sigint);
  while (!interrupted)
  {
    dds_return_t ret = 0;
    void *sample = tpentry->samples[sample_idx];
    uint32_t * const countp =
      (tpentry->count_offset != SIZE_MAX)
      ? (uint32_t *) ((unsigned char *) sample + tpentry->count_offset)
      : 0;
    if (countp)
      *countp = count++;
    if ((ret = dds_write (writer, sample)) < 0)
    {
      fprintf (stderr, "dds_write: %s\n", dds_strretcode (ret));
      dds_delete (participant);
      return 1;
    }
    if (tpentry->samples[++sample_idx] == NULL)
    {
      sample_idx = 0;
    }
    dds_sleepfor (DDS_SECS (1));
  }

  dds_delete (participant);
  return 0;
}