dynsub

The dynsub example is a PoC for a C-based 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 by default. Pass -f xml to print samples as XML instead.

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
  1// Copyright(c) 2022 to 2023 ZettaScale Technology and others
  2//
  3// This program and the accompanying materials are made available under the
  4// terms of the Eclipse Public License v. 2.0 which is available at
  5// http://www.eclipse.org/legal/epl-2.0, or the Eclipse Distribution License
  6// v. 1.0 which is available at
  7// http://www.eclipse.org/org/documents/edl-v10.php.
  8//
  9// SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
 10
 11#include <stdio.h>
 12#include <string.h>
 13#include <stdlib.h>
 14#include <assert.h>
 15#include <locale.h>
 16#include <signal.h>
 17#include <getopt.h>
 18#if !defined _WIN32 && !DDSRT_WITH_FREERTOS && !__ZEPHYR__
 19#include <unistd.h>
 20#endif
 21
 22#include "dds/dds.h"
 23#include "dds/ddsrt/heap.h"
 24
 25#include "dyntypelib.h"
 26
 27// Interpreting the data of an arbitrary topic requires interpreting the type object that describes the data.
 28// The type object type is defined by the XTypes specification (https://www.omg.org/spec/DDS-XTypes/) and it
 29// comes in two forms: MinimalTypeObject and CompleteTypeObject.  Only the latter includes field names, so
 30// that's what need.
 31//
 32// Cyclone DDS includes a copy of the IDL as well as the corresponding type definitions in C as generated by
 33// IDLC.  So instead of including yet another copy, we simply refer to those.  These files are not (yet?)
 34// part of the stable API of Cyclone DDS and so the updates to Cyclone may change the locations or the names
 35// of the relevant header files.
 36//
 37// The API uses `dds_typeobj_t` and `dds_typeinfo_t` that are opaque types but really amount to the
 38// corresponding XTypes objects: DDS_XTypes_TypeObject and DDS_XTypes_TypeInformation.  Rather than casting
 39// pointers like we do here, they should be defined in a slightly different way so that they are not really
 40// opaque.  For now, this'll have to do.
 41#include "dds/ddsc/dds_public_alloc.h"
 42#include "dds/ddsi/ddsi_sertype.h"
 43#include "dds/ddsi/ddsi_xt_typeinfo.h"
 44
 45#include "dds/ddsrt/threads.h"
 46#include "dds/ddsi/ddsi_serdata.h"
 47
 48// For convenience, the DDS participant is global
 49static struct dyntypelib *dyntypelib;
 50static dds_entity_t participant;
 51static enum dtl_sample_format output_format = DTL_SAMPLE_FORMAT_JSON;
 52
 53static dds_entity_t termcond;
 54
 55static void usage (const char *argv0)
 56{
 57  fprintf (stderr, "usage: %s [-r] [-f json|xml] topicname\n", argv0);
 58}
 59
 60static bool parse_sample_format (const char *arg, enum dtl_sample_format *fmt)
 61{
 62  if (strcmp (arg, "json") == 0)
 63  {
 64    *fmt = DTL_SAMPLE_FORMAT_JSON;
 65    return true;
 66  }
 67  if (strcmp (arg, "xml") == 0)
 68  {
 69    *fmt = DTL_SAMPLE_FORMAT_XML;
 70    return true;
 71  }
 72  return false;
 73}
 74
 75static void print_sample (struct dyntypelib *dtl, bool valid_data, const void *sample, const DDS_XTypes_CompleteTypeObject *typeobj)
 76{
 77  struct dyntypelib_error err = { .errmsg = "" };
 78  char *str = NULL;
 79  const struct dtl_sample_print_options opts = {
 80    .format = output_format,
 81    .trailing_newline = true
 82  };
 83  dds_return_t rc = dtl_print_sample_to_string (dtl, valid_data, sample, typeobj, &opts, &str, NULL, &err);
 84  if (rc == DDS_RETCODE_OK)
 85  {
 86    (void) fputs (str, stdout);
 87    ddsrt_free (str);
 88  }
 89  else
 90  {
 91    printf ("(sample print failed: %s)\n", err.errmsg);
 92  }
 93}
 94
 95// Helper function to wait for a DCPSPublication/DCPSSubscription to show up with the desired topic name,
 96// then calls dds_find_topic to create a topic for that data writer's/reader's type up the retrieves the
 97// type object.
 98static dds_return_t get_topic_and_typeobj (const char *topic_name, dds_duration_t timeout, dds_entity_t *topic, const DDS_XTypes_TypeObject **xtypeobj)
 99{
100  const dds_entity_t waitset = dds_create_waitset (participant);
101  const dds_entity_t dcpspublication_reader = dds_create_reader (participant, DDS_BUILTIN_TOPIC_DCPSPUBLICATION, NULL, NULL);
102  const dds_entity_t dcpspublication_readcond = dds_create_readcondition (dcpspublication_reader, DDS_ANY_STATE);
103  const dds_entity_t dcpssubscription_reader = dds_create_reader (participant, DDS_BUILTIN_TOPIC_DCPSSUBSCRIPTION, NULL, NULL);
104  const dds_entity_t dcpssubscription_readcond = dds_create_readcondition (dcpssubscription_reader, DDS_ANY_STATE);
105  (void) dds_waitset_attach (waitset, dcpspublication_readcond, dcpspublication_reader);
106  (void) dds_waitset_attach (waitset, dcpssubscription_readcond, dcpssubscription_reader);
107  const dds_time_t abstimeout = (timeout == DDS_INFINITY) ? DDS_NEVER : dds_time () + timeout;
108  dds_return_t ret = DDS_RETCODE_OK;
109  *xtypeobj = NULL;
110  dds_attach_t triggered_reader_x;
111  while (*xtypeobj == NULL && dds_waitset_wait_until (waitset, &triggered_reader_x, 1, abstimeout) > 0)
112  {
113    void *epraw = NULL;
114    dds_sample_info_t si;
115    dds_entity_t triggered_reader = (dds_entity_t) triggered_reader_x;
116    if (dds_take (triggered_reader, &epraw, &si, 1, 1) <= 0)
117      continue;
118    dds_builtintopic_endpoint_t *ep = epraw;
119    const dds_typeinfo_t *typeinfo = NULL;
120    // We are only interested in DCPSPublications where the topic name matches and that carry type information
121    // (a non-XTypes capable DDS would not provide type information) because without that information there is
122    // no way we can do anything interesting with it.
123    if (strcmp (ep->topic_name, topic_name) == 0 && dds_builtintopic_get_endpoint_type_info (ep, &typeinfo) == 0 && typeinfo)
124    {
125      // Using dds_find_topic allows us to "clone" the topic definition including the topic QoS, but it does
126      // require that topic discovery is enabled in the configuration.  The advantage of using dds_find_topic
127      // is that it creates a topic with the same name, type *and QoS*.  That distinction only matters if
128      // topic is discovery is enabled and/or if the topic has a durability kind of of transient or persistent:
129      // - using a different topic QoS might result in an incompatible QoS notification if topic discovery is
130      //   enabled (everything would still work).
131      // - transient/persistent data behaviour is defined in terms of the topic QoS actually really matters
132      //
133      // So we try to use dds_find_topic, and if that fails, try to go the other route using the writer's QoS
134      // as an approximation of the topic QoS.
135      if ((*topic = dds_find_topic (DDS_FIND_SCOPE_GLOBAL, participant, ep->topic_name, typeinfo, DDS_SECS (2))) < 0)
136      {
137        fprintf (stderr, "dds_find_topic: %s ... continuing on the assumption that topic discovery is disabled\n", dds_strretcode (*topic));
138        dds_topic_descriptor_t *descriptor;
139        if ((ret = dds_create_topic_descriptor(DDS_FIND_SCOPE_GLOBAL, participant, typeinfo, DDS_SECS (10), &descriptor)) < 0)
140        {
141          fprintf (stderr, "dds_create_topic_descriptor: %s\n", dds_strretcode (ret));
142          dds_return_loan (triggered_reader, &epraw, 1);
143          goto error;
144        }
145        dds_qset_data_representation (ep->qos, 0, NULL);
146        if ((*topic = dds_create_topic (participant, descriptor, ep->topic_name, ep->qos, NULL)) < 0)
147        {
148          fprintf (stderr, "dds_create_topic_descriptor: %s (be sure to enable topic discovery in the configuration)\n", dds_strretcode (*topic));
149          dds_delete_topic_descriptor (descriptor);
150          dds_return_loan (triggered_reader, &epraw, 1);
151          goto error;
152        }
153        dds_delete_topic_descriptor (descriptor);
154      }
155      // The topic suffices for creating a reader, but we also need the TypeObject to make sense of the data
156      if ((*xtypeobj = load_type_with_deps (dyntypelib->typecache, participant, typeinfo, &dyntypelib->ppc)) == NULL)
157      {
158        fprintf (stderr, "loading type with all dependencies failed\n");
159        dds_return_loan (triggered_reader, &epraw, 1);
160        goto error;
161      }
162      if (load_type_with_deps_min (dyntypelib->typecache, participant, typeinfo, &dyntypelib->ppc) == NULL)
163      {
164        fprintf (stderr, "loading minimal type with all dependencies failed\n");
165        dds_return_loan (triggered_reader, &epraw, 1);
166        goto error;
167      }
168    }
169    dds_return_loan (triggered_reader, &epraw, 1);
170  }
171  if (*xtypeobj)
172  {
173    // If we got the type object, populate the type cache
174    size_t align, size;
175    build_typecache_to (dyntypelib->typecache, &(*xtypeobj)->_u.complete, &align, &size);
176    fflush (stdout);
177    struct typeinfo templ = { .key = { .key = (uintptr_t) *xtypeobj } } , *info;
178    if ((info = type_cache_lookup (dyntypelib->typecache, &templ)) != NULL)
179    {
180      assert (info->release == NULL);
181    }
182    else
183    {
184      // not sure whether this is at all possible
185      info = ddsrt_malloc (sizeof (*info));
186      *info = (struct typeinfo){ .key = { .key = (uintptr_t) *xtypeobj }, .typeobj = &(*xtypeobj)->_u.complete, .release = NULL, .align = align, .size = size };
187      type_cache_add (dyntypelib->typecache, info);
188    }
189  }
190error:
191  dds_delete (dcpspublication_reader);
192  dds_delete (dcpssubscription_reader);
193  dds_delete (waitset);
194  return (*xtypeobj != NULL) ? DDS_RETCODE_OK : DDS_RETCODE_TIMEOUT;
195}
196
197static bool print_sample_normal (dds_entity_t reader, const DDS_XTypes_TypeObject *xtypeobj)
198{
199  void *raw = NULL;
200  dds_sample_info_t si;
201  dds_return_t ret;
202  if ((ret = dds_take (reader, &raw, &si, 1, 1)) < 0)
203    return false;
204  else if (ret != 0)
205  {
206    // ... that we then print
207    print_sample (dyntypelib, si.valid_data, raw, &xtypeobj->_u.complete);
208    if (dds_return_loan (reader, &raw, 1) < 0)
209      return false;
210  }
211  return true;
212}
213
214static void hexdump (const unsigned char *msg, const size_t len)
215{
216  for (size_t off16 = 0; off16 < len; off16 += 16)
217  {
218    printf ("%04" PRIxSIZE " ", off16);
219    size_t off1;
220    for (off1 = 0; off1 < 16 && off16 + off1 < len; off1++)
221      printf ("%s %02x", (off1 == 8) ? " " : "", msg[off16 + off1]);
222    for (; off1 < 16; off1++)
223      printf ("%s   ", (off1 == 8) ? " " : "");
224    printf ("  |");
225    for (off1 = 0; off1 < 16 && off16 + off1 < len; off1++)
226    {
227      unsigned char c = msg[off16 + off1];
228      printf ("%c", (c >= 32 && c < 127) ? c : '.');
229    }
230    printf ("|\n");
231  }
232  fflush (stdout);
233}
234
235static const char *encodingstr (const struct ddsi_serdata *sd)
236{
237  uint16_t encoding;
238  ddsi_serdata_to_ser (sd, 0, 2, &encoding);
239  switch (encoding)
240  {
241    case DDSI_RTPS_CDR_BE:
242    case DDSI_RTPS_CDR_LE:
243      return "CDR";
244    case DDSI_RTPS_PL_CDR_BE:
245    case DDSI_RTPS_PL_CDR_LE:
246      return "PL_CDR";
247    case DDSI_RTPS_CDR2_BE:
248    case DDSI_RTPS_CDR2_LE:
249      return "CDR2";
250    case DDSI_RTPS_D_CDR2_BE:
251    case DDSI_RTPS_D_CDR2_LE:
252      return "D_CDR2";
253    case DDSI_RTPS_PL_CDR2_BE:
254    case DDSI_RTPS_PL_CDR2_LE:
255      return "PL_CDR2";
256    default:
257      return "unknown";
258  }
259}
260
261static bool print_sample_cdr (dds_entity_t reader, const DDS_XTypes_TypeObject *xtypeobj)
262{
263  // Note: doesn't print the exact CDR received, but the "normalised" one where byteswapping
264  // has been performed, booleans have been mapped to 0 or 1, and perhaps some other similar
265  // changes have been made.
266  struct ddsi_serdata *sd = NULL;
267  dds_sample_info_t si;
268  dds_return_t ret;
269  if ((ret = dds_takecdr (reader, &sd, 1, &si, 0)) < 0)
270    return false;
271  else if (ret != 0)
272  {
273    printf ("encoding: %s", encodingstr (sd));
274    if (!si.valid_data)
275      printf (" (expect XCDR2 because it is an invalid sample)");
276    printf ("\n");
277    if (ddsi_serdata_size (sd) == 4)
278      printf ("(no payload)\n");
279    else
280    {
281      ddsrt_iovec_t iov;
282      struct ddsi_serdata *refsd;
283      refsd = ddsi_serdata_to_ser_ref (sd, 4, ddsi_serdata_size (sd) - 4, &iov);
284      hexdump (iov.iov_base, iov.iov_len);
285      ddsi_serdata_to_ser_unref (refsd, &iov);
286    }
287
288    void *raw = ddsrt_calloc (1, sd->type->sizeof_type);
289    if (raw == NULL)
290      abort ();
291
292    bool ok;
293    if (si.valid_data)
294      ok = ddsi_serdata_to_sample (sd, raw, NULL, NULL);
295    else
296    {
297      const struct ddsi_sertype *st;
298      dds_get_entity_sertype (reader, &st);
299      ok = ddsi_serdata_untyped_to_sample (st, sd, raw, NULL, NULL);
300    }
301    if (ok)
302      print_sample (dyntypelib, si.valid_data, raw, &xtypeobj->_u.complete);
303    else
304      printf ("(conversion to sample failed)\n");
305    ddsi_sertype_free_sample (sd->type, raw, DDS_FREE_CONTENTS);
306    ddsrt_free (raw);
307
308    ddsi_serdata_unref (sd);
309  }
310  return true;
311}
312
313#if !DDSRT_WITH_FREERTOS && !__ZEPHYR__
314static void signal_handler (int sig)
315{
316  (void) sig;
317  dds_set_guardcondition (termcond, true);
318}
319#endif
320
321#if !_WIN32 && !DDSRT_WITH_FREERTOS && !__ZEPHYR__
322static uint32_t sigthread (void *varg)
323{
324  sigset_t *set = varg;
325  int sig;
326  if (sigwait (set, &sig) == 0)
327    signal_handler (sig);
328  return 0;
329}
330#endif
331
332int main (int argc, char **argv)
333{
334  dds_return_t ret = 0;
335  dds_entity_t topic = 0;
336  bool raw_mode;
337  const char *topic_name;
338
339  // for printf("%ls")
340  setlocale (LC_CTYPE, "");
341
342  raw_mode = false;
343  int opt;
344  while ((opt = getopt (argc, argv, "f:r")) != EOF)
345  {
346    switch (opt)
347    {
348      case 'f':
349        if (!parse_sample_format (optarg, &output_format))
350        {
351          fprintf (stderr, "unsupported output format %s\n", optarg);
352          return 2;
353        }
354        break;
355      case 'r':
356        raw_mode = true;
357        break;
358      default:
359        usage (argv[0]);
360        return 2;
361    }
362  }
363  if (argc - optind != 1)
364  {
365    usage (argv[0]);
366    return 2;
367  }
368  topic_name = argv[optind];
369
370  participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
371  if (participant < 0)
372  {
373    fprintf (stderr, "dds_create_participant: %s\n", dds_strretcode (participant));
374    return 1;
375  }
376
377  // The one magic step: get a topic and type object ...
378  dyntypelib = dtl_new (participant);
379  const DDS_XTypes_TypeObject *xtypeobj;
380  if ((ret = get_topic_and_typeobj (topic_name, DDS_SECS (10), &topic, &xtypeobj)) < 0)
381  {
382    fprintf (stderr, "get_topic_and_typeobj: %s\n", dds_strretcode (ret));
383    goto error;
384  }
385  // ... given those, we can create a reader just like we do normally ...
386  const dds_entity_t reader = dds_create_reader (participant, topic, NULL, NULL);
387  // ... and create a waitset that allows us to wait for any incoming data ...
388  const dds_entity_t waitset = dds_create_waitset (participant);
389  const dds_entity_t readcond = dds_create_readcondition (reader, DDS_ANY_STATE);
390  (void) dds_waitset_attach (waitset, readcond, 0);
391
392  termcond = dds_create_guardcondition (participant);
393  (void) dds_waitset_attach (waitset, termcond, 0);
394
395#ifdef _WIN32
396  signal (SIGINT, signal_handler);
397#elif !DDSRT_WITH_FREERTOS && !__ZEPHYR__
398  ddsrt_thread_t sigtid;
399  sigset_t sigset, osigset;
400  sigemptyset (&sigset);
401#ifdef __APPLE__
402  DDSRT_WARNING_GNUC_OFF(sign-conversion)
403#endif
404  sigaddset (&sigset, SIGHUP);
405  sigaddset (&sigset, SIGINT);
406  sigaddset (&sigset, SIGTERM);
407#ifdef __APPLE__
408  DDSRT_WARNING_GNUC_ON(sign-conversion)
409#endif
410  sigprocmask (SIG_BLOCK, &sigset, &osigset);
411  {
412    ddsrt_threadattr_t tattr;
413    ddsrt_threadattr_init (&tattr);
414    ddsrt_thread_create (&sigtid, "sigthread", &tattr, sigthread, &sigset);
415  }
416#endif
417
418  bool termflag = false;
419  while (!termflag)
420  {
421    (void) dds_waitset_wait (waitset, NULL, 0, DDS_INFINITY);
422    dds_read_guardcondition (termcond, &termflag);
423
424    bool ok = raw_mode ? print_sample_cdr (reader, xtypeobj) : print_sample_normal (reader, xtypeobj);
425    if (!ok)
426      break;
427  }
428
429#if _WIN32
430  signal_handler (SIGINT);
431#elif !DDSRT_WITH_FREERTOS && !__ZEPHYR__
432  {
433    /* get the attention of the signal handler thread */
434    void (*osigint) (int);
435    void (*osigterm) (int);
436    kill (getpid (), SIGTERM);
437    ddsrt_thread_join (sigtid, NULL);
438    osigint = signal (SIGINT, SIG_IGN);
439    osigterm = signal (SIGTERM, SIG_IGN);
440    sigprocmask (SIG_SETMASK, &osigset, NULL);
441    signal (SIGINT, osigint);
442    signal (SIGINT, osigterm);
443  }
444#endif
445
446error:
447  dtl_free (dyntypelib);
448  dds_delete (participant);
449  return ret < 0;
450}
variouspub_types.idl
 1struct A {
 2  @key
 3  string name;
 4  string message;
 5  unsigned long count;
 6};
 7
 8struct T {
 9  short s;
10  long l;
11};
12
13struct B {
14  A a;
15  sequence<T> ts;
16};
17
18struct C {
19  B b;
20  @key
21  short k;
22};
23
24module M1 {
25  @appendable
26  struct O {
27    @optional long x;
28  };
29};
30
31struct D {
32  wstring ws;
33  wchar wc;
34  unsigned long count;
35};
36
37struct U {
38  uint32 w;
39  @key string x;
40  string y;
41  @key uint32 z;
42};
43
44struct E {
45  uint32 a;
46  @key sequence<U> b[2];
47  @key uint32 c;
48};
variouspub.c
  1#include <stdio.h>
  2#include <string.h>
  3#include <stdlib.h>
  4#include <signal.h>
  5#include <assert.h>
  6
  7#include "dds/dds.h"
  8#include "variouspub_types.h"
  9
 10static void *samples_a[] = {
 11  &(A){ "Mariken",   "Wie sidi, vrient?", 0 },
 12  &(A){ "Die duvel", "Een meester vol consten,", 0 },
 13  &(A){ "Die duvel", "Nieuwers af falende, wes ic besta.", 0 },
 14  &(A){ "Mariken",   "'t Comt mi alleleens met wien dat ick ga,", 0 },
 15  &(A){ "Mariken",   "Also lief gae ic metten quaetsten als metten besten.", 0 },
 16  NULL
 17};
 18
 19static void *samples_b[] = {
 20  &(B){ {"Die duvel", "Wildi u liefde te mi werts vesten,", 0},
 21        { ._length = 2, ._maximum = 2, ._release = false,
 22          ._buffer = (T[]){ {2,3},{5,7} } } },
 23  &(B){ {"Die duvel", "Ick sal u consten leeren sonder ghelijcke,", 0},
 24        { ._length = 3, ._maximum = 3, ._release = false,
 25          ._buffer = (T[]){ {11,13},{17,19},{23,29} } } },
 26  &(B){ {"Die duvel", "Die seven vrie consten: rethorijcke, musijcke,", 0},
 27        { ._length = 5, ._maximum = 5, ._release = false,
 28          ._buffer = (T[]){ {31,37},{41,43},{47,52},{59,61},{67,71} } } },
 29  &(B){ {"Die duvel", "Logica, gramatica ende geometrie,", 0},
 30        { ._length = 7, ._maximum = 7, ._release = false,
 31          ._buffer = (T[]){ {73,79},{83,89},{97,101},{103,107},{109,113},
 32                            {127,131},{137,139} } } },
 33  &(B){ {"Die duvel", "Aristmatica ende alkenie,", 0},
 34        { ._length = 11, ._maximum = 11, ._release = false,
 35          ._buffer = (T[]){ {149,151},{157,163},{167,173},{179,181},
 36                            {191,193},{197, 199},{211,223},{227,229},
 37                            {233,239},{241,251},{257,263} } } },
 38  NULL
 39};
 40
 41static void *samples_c[] = {
 42  &(C){ { {"Die duvel", "Dwelc al consten sijn seer curable.", 0},
 43          { ._length = 13, ._maximum = 13, ._release = false,
 44            ._buffer = (T[]){ {269,271},{277,281},{283,293},{307,311},
 45                              {313,317},{331,337},{347,349},{353,359},
 46                              {367,373},{379,383},{389,397},{401,409},
 47                              {419,421} } } },
 48        8936 },
 49  &(C){ { {"Die duvel", "Noyt vrouwe en leefde op eerde so able", 0},
 50          { ._length = 17, ._maximum = 17, ._release = false,
 51            ._buffer = (T[]){ {431,433},{439,443},{449,457},{461,463},
 52                              {467,479},{487,491},{499,503},{509,521},
 53                              {523,541},{547,557},{563,569},{571,577},
 54                              {587,593},{599,601},{607,613},{617,619},
 55                              {631,641} } } },
 56        18088 },
 57  &(C){ { {"Die duvel", "Als ic u maken sal.", 0},
 58          { ._length = 19, ._maximum = 19, ._release = false,
 59            ._buffer = (T[]){ {643,647},{653,659},{661,673},{677,683},
 60                              {691,701},{709,719},{727,733},{739,743},
 61                              {751,757},{761,769},{773,787},{797,809},
 62                              {811,821},{823,827},{829,839},{853,857},
 63                              {859,863},{877,881},{883,887} } } },
 64        29172 },
 65  &(C){ { {"Mariken",  "So moetti wel zijn een constich man.", 0},
 66          { ._length = 23, ._maximum = 23, ._release = false,
 67            ._buffer = (T[]){ {907,911},{919,929},{937,941},{947,953},
 68                              {967,971},{977,983},{991,997},{1009,1013},
 69                              {1019,1021},{1031,1033},{1039,1049},{1051,1061},
 70                              {1063,1069},{1087,1091},{1093,1097},{1103,1109},
 71                              {1117,1123},{1129,1151},{1153,1163},{1171,1181},
 72                              {1187,1193},{1201,1213},{1217,1223} } } },
 73        16022 },
 74  &(C){ { {"Mariken",  "Wie sidi dan?", 0},
 75          { ._length = 29, ._maximum = 29, ._release = false,
 76            ._buffer = (T[]){ {1229,1231},{1237,1249},{1259,1277},{1279,1283},
 77                              {1289,1291},{1297,1301},{1303,1307},{1319,1321},
 78                              {1327,1361},{1367,1373},{1381,1399},{1409,1423},
 79                              {1427,1429},{1433,1439},{1447,1451},{1453,1459},
 80                              {1471,1481},{1483,1487},{1489,1493},{1499,1511},
 81                              {1523,1531},{1543,1549},{1553,1559},{1567,1571},
 82                              {1579,1583},{1597,1601},{1607,1609},{1613,1619},
 83                              {1621,1627} } } },
 84        17880 },
 85  NULL
 86};
 87
 88static int32_t long_4 = 4;
 89static void *samples_M1_O[] = {
 90  &(M1_O){ .x = NULL },
 91  &(M1_O){ .x = &long_4 },
 92  NULL
 93};
 94
 95static void *samples_d[] = {
 96  &(D){ L"😀 Een Kruyck gaat soo langh te water tot datse barst.", 0x2206, 0 },
 97  &(D){ L"🙃 Men treckt een Boogh soo lang tot datse stucken knarst.", 0x2207, 0 },
 98  &(D){ L"😊 De Steel-kunst doet zyn Meester de dood vaak verwerven.", 0x22a5, 0 },
 99  NULL
100};
101
102static void *samples_e[] = {
103  &(E){ 0, {{0},{0}}, 456 },
104  &(E){ 0, {{0},{._length=1, ._maximum=1, ._buffer=&(U){34,"aap","noot",56}, ._release = false}}, 456},
105  &(E){ 0, {{0},{._length=1, ._maximum=1, ._buffer=&(U){78,"aap","zus",56}, ._release = false}}, 456},
106  &(E){ 0, {{0},{._length=1, ._maximum=1, ._buffer=&(U){78,"wim","zus",90}, ._release = false}}, 456},
107  NULL
108};
109
110static struct tpentry {
111  const char *name;
112  const dds_topic_descriptor_t *descr;
113  void **samples;
114  size_t count_offset;
115} tptab[] = {
116  { "A", &A_desc, samples_a, offsetof (A, count) },
117  { "B", &B_desc, samples_b, offsetof (B, a.count) },
118  { "C", &C_desc, samples_c, offsetof (C, b.a.count) },
119  { "M1::O", &M1_O_desc, samples_M1_O, SIZE_MAX },
120  { "D", &D_desc, samples_d, offsetof (D, count) },
121  { "E", &E_desc, samples_e, offsetof (E, a) },
122  { NULL, NULL, NULL, 0 }
123};
124
125static void usage (const char *argv0)
126{
127  fprintf (stderr, "usage: %s {", argv0);
128  const char *sep = "";
129  for (struct tpentry *tpentry = &tptab[0]; tpentry->name; tpentry++)
130  {
131    fprintf (stderr, "%s%s", sep, tpentry->name);
132    sep = "|";
133  }
134  fprintf (stderr, "}\n");
135  exit (2);
136}
137
138static volatile sig_atomic_t interrupted;
139
140static void sigint (int sig)
141{
142  (void) sig;
143  interrupted = 1;
144}
145
146int main (int argc, char **argv)
147{
148  if (argc != 2)
149    usage (argv[0]);
150  struct tpentry *tpentry;
151  for (tpentry = &tptab[0]; tpentry->name; tpentry++)
152    if (strcmp (tpentry->name, argv[1]) == 0)
153      break;
154  if (tpentry->name == NULL)
155    usage (argv[0]);
156
157  const dds_entity_t participant = dds_create_participant (DDS_DOMAIN_DEFAULT, NULL, NULL);
158  if (participant < 0)
159  {
160    fprintf (stderr, "dds_create_participant: %s\n", dds_strretcode (participant));
161    return 1;
162  }
163
164  const dds_entity_t topic = dds_create_topic (participant, tpentry->descr, tpentry->name, NULL, NULL);
165  const dds_entity_t writer = dds_create_writer (participant, topic, NULL, NULL);
166  uint32_t sample_idx = 0;
167  uint32_t count = 0;
168  signal (SIGINT, sigint);
169  while (!interrupted)
170  {
171    dds_return_t ret = 0;
172    void *sample = tpentry->samples[sample_idx];
173    uint32_t * const countp =
174      (tpentry->count_offset != SIZE_MAX)
175      ? (uint32_t *) ((unsigned char *) sample + tpentry->count_offset)
176      : 0;
177    if (countp)
178      *countp = count++;
179    if ((ret = dds_write (writer, sample)) < 0)
180    {
181      fprintf (stderr, "dds_write: %s\n", dds_strretcode (ret));
182      dds_delete (participant);
183      return 1;
184    }
185    if (tpentry->samples[++sample_idx] == NULL)
186    {
187      sample_idx = 0;
188    }
189    dds_sleepfor (DDS_SECS (1));
190  }
191
192  dds_delete (participant);
193  return 0;
194}