corTestClient.c

//
// FILE            corTestClient.c
//
// AUTHOR          Ken Zangelin
//
// Copyright 2026 Seamware
// SPDX-License-Identifier: Apache-2.0
//
// Test notification receiver for functional tests.
// Accumulates all incoming requests (e.g. POST /notify) in a global array.
// Provides:
//   GET    /dump   - return the requests received SINCE THE LAST DUMP, and
//                    consume them. A dump reads; read notifications are gone.
//   GET    /die    - exit the process
//   POST   /**     - accumulate the request (catch-all)
//
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <pthread.h>

#include <mosquitto.h>

#include "corArgs/corArgs.h"
#include "corArgs/corArgsBuiltins.h"
#include "corTree/CorNode.h"
#include "corJson/CorJson.h"
#include "corTree/corTreeBuilder.h"
#include "corTree/corTreeLookup.h"
#include "corJson/corJsonRender.h"
#include "corJson/corJsonRenderSize.h"
#include "corTree/corTreeClone.h"
#include "corTree/corTreeFree.h"
#include "corAlloc/corAlloc.h"
#include "corAlloc/corAlloc.h"
#include "corLog/corLog.h"
#include "corBase/corFileRead.h"                   // corFileRead
#include "corJson/corJsonParse.h"                 // corJsonParse
#include "corAlloc/corAllocBufferInit.h"          // corAllocBufferInit
#include "corAlloc/corAllocBufferReset.h"         // corAllocBufferReset
#include "corJson/corJsonCreate.h"                // corJsonCreate
#include "corPlugin/corPlugin.h"                  // corPluginSetBaseDir, corPluginResolve, corPluginOpen
#include "corBridge/BridgeDriver.h"                // BridgeDriver, BridgeRegisterFunc, BRIDGES_MAX
#include "corBridge/BridgeBroker.h"                // BridgeBroker, BRIDGE_*
#include "corBridge/BridgeServer.h"                // BridgeServer, BridgeServiceRequestFunc
#include "corRest/corRest.h"



// -----------------------------------------------------------------------------
//
// Globals
//
static CorNode* dumpArray    = NULL;     // accumulates requests (malloc allocator)
static int     dumpCount     = 0;
static int     probeCount    = 0;        // sourceIdentity discovery probes seen (kept OUT of dumpArray)

//
// dumpMutex - guards dumpArray + dumpCount, on every path that touches them.
//
// corTestClient is thread-per-connection, so two notifications in flight at once run
// dumpAccumulate on two threads and spliced into the same tail unguarded: the
// order of the two entries flipped, and one of them could be lost outright.
// GET /dump rendered the list while it was being appended to, and DELETE /dump
// corTreeFree'd it out from under a writer.
//
// The mutex existed already - as mqttDumpMutex, taken only by the MQTT
// listener, which shares this very array with the HTTP path. It was never
// MQTT's mutex; it guards the accumulator, and it is named for that now.
//
static pthread_mutex_t dumpMutex = PTHREAD_MUTEX_INITIALIZER;



// -----------------------------------------------------------------------------
//
// Programmable response stubs (the mock-reply API)
//
// A test POSTs to /mock/reply to program how corTestClient answers a forwarded
// request — { "verb": "POST", "path": "/attrs/", "status": 207, "body": {...} }.
// On each forwarded request the stub list is consulted (verb match + path
// substring); the first match's status + body is served, else the --status
// fallback applies. Stubs are persistent until DELETE /mock/reply.
//
typedef struct FtStub
{
  char*           verb;     // "POST", "GET", ...
  char*           path;     // substring to find in the request urlPath
  int             status;   // HTTP status to return
  int             delayMs;  // sleep this many ms before responding (per-path timeout injection), 0 = none
  char*           body;     // rendered JSON body to return (malloc), or NULL
  struct FtStub*  next;
} FtStub;

static FtStub* ftStubs = NULL;



// -----------------------------------------------------------------------------
//
// Command line arguments
//
unsigned short ftPort       = 7701;
bool           ftFg         = true;
unsigned short ftPostStatus = 200;     // status returned for accumulate POSTs; override with --status
unsigned int   ftDelayMs    = 0;       // sleep before responding — for timeout tests
bool           ftDiscard    = false;   // POSTs counted, not kept - a benchmark's receiver
//
// ftMqttSubscribed - set when the MQTT SUBACK arrives, read by GET /mqttReady.
//
// Not "the thread started" and not "subscribe was called": a publish that
// reaches the broker before the SUBACK is delivered to nobody and is DISCARDED,
// not queued. So the only readiness that means anything to a test is the
// acknowledgement, which is why this is set in the on-subscribe callback.
//
static volatile bool ftMqttSubscribed = false;

//
// ftMqttFailed - set when the MQTT connection is given up on for good.
//
// The connect loop used to be unbounded ("try forever, the broker may not be up
// yet"), which is right for a broker that is merely slow and wrong for one that
// is never coming: corTestClient span silently, printed nothing, and the only symptom
// was a readiness barrier timing out with an EMPTY log to show for it. A bounded
// retry that says why is strictly better - the failure is reported by the process
// that actually observed it.
//
static volatile bool ftMqttFailed = false;

//
// ftMqttConn - the subscriber's connection, published on by POST /mqtt/publish once it is up
//
static struct mosquitto* volatile ftMqttConn = NULL;

unsigned short ftMqttPort   = 0;       // 0 = no MQTT subscription
char*          ftMqttTopic  = (char*) "#"; // default: catch every topic
char*          ftMqttUser   = NULL;    // MQTT username (for an auth-required broker)
char*          ftMqttPass   = NULL;    // MQTT password
bool           ftMqttTls    = false;   // subscribe over TLS (mqtts), insecure
char*          ftHttpsKey   = NULL;    // path to a PEM private key  (enables TLS when both set)
char*          ftHttpsCert  = NULL;    // path to a PEM certificate  (enables TLS when both set)

//
// Bridge hosting.
//
// corTestClient loads the SAME plugin the broker loads, rather than speaking a
// transport of its own. Two reasons, and the second is the better one:
//
//   the plugin is C++ and links 21 MiB of transport libraries, and corTestClient is
//   built inside PROJECT(coraine C) - linking that here would put a C++
//   compiler and the whole eProsima stack into coraine's own build tree, which
//   is exactly what the plugin architecture exists to avoid.
//
//   and a test that drives the broker's actual plugin proves the broker's
//   actual plugin works. A second, independently written client can be wrong
//   in its own way and agree with nothing.
//
// So corTestClient is a bridge HOST: it supplies its own BridgeBroker, whose
// sampleIn records into the dump instead of writing an entity.
//
char*         ftBridges      = NULL;   // comma-separated bridge plugins to load
char*         ftBridgeConfig = NULL;   // the shared bridge configuration file

static BridgeDriver  ftBridgeV[BRIDGES_MAX];
static int           ftBridgeCount = 0;


static char* ftTraceLevels = (char*) "0-255";

static CorArg ftArgV[] =
{
  { "--traceLevels",     NULL,  CorArgString, _vp &ftTraceLevels, CorArgOpt, _vp "0-255", NULL, NULL,  "trace levels to log - every one by default (the functests read them); \"\" for none - a benchmark's notification receiver logs a full trace of every request otherwise" },
  { "--port",            "-p",  CorArgUShort, _vp &ftPort,   CorArgOpt, _vp 7701,  _vp 1, _vp 65535, "TCP port to listen on" },
  { "--foreground",      "-fg", CorArgBool,   _vp &ftFg,         CorArgOpt, _vp true, _vp false, _vp true, "run in foreground" },
  { "--status",          "-s",  CorArgUShort, _vp &ftPostStatus, CorArgOpt, _vp 200,   _vp 100, _vp 599, "HTTP status for accumulate POSTs (misbehave mode)" },
  { "--delay",           NULL,  CorArgUInt,   _vp &ftDelayMs,    CorArgOpt, _vp 0, _vp 0, _vp 600000, "sleep N ms before responding (timeout tests)" },
  { "--discard",         NULL,  CorArgBool,   _vp &ftDiscard,    CorArgOpt, _vp false, _vp false, _vp true, "POSTs counted (GET /count), not kept - a benchmark's notification receiver" },
  { "--mqttPort",        NULL,  CorArgUShort, _vp &ftMqttPort,   CorArgOpt, _vp 0, _vp 0, _vp 65535, "MQTT broker port to subscribe to (0 = disabled)" },
  { "--mqttTopic",       NULL,  CorArgString, _vp &ftMqttTopic,  CorArgOpt, _vp "#",   NULL,  NULL,      "MQTT topic to subscribe (default '#')" },
  { "--mqttUser",        NULL,  CorArgString, _vp &ftMqttUser,   CorArgOpt, NULL,  NULL,  NULL,      "MQTT username (auth-required broker)" },
  { "--mqttPassword",    NULL,  CorArgString, _vp &ftMqttPass,   CorArgOpt, NULL,  NULL,  NULL,      "MQTT password" },
  { "--mqttTls",         NULL,  CorArgBool,   _vp &ftMqttTls,    CorArgOpt, _vp false, _vp false, _vp true, "subscribe over TLS (mqtts), accept self-signed" },
  { "--httpsKey",        "-k",  CorArgString, _vp &ftHttpsKey,   CorArgOpt, NULL,  NULL,  NULL,      "PEM private key file (serve HTTPS; needs --httpsCertificate)" },
  { "--httpsCertificate","-c",  CorArgString, _vp &ftHttpsCert,  CorArgOpt, NULL,  NULL,  NULL,      "PEM certificate file (serve HTTPS; needs --httpsKey)" },
  { "--bridges",         "-br", CorArgString, _vp &ftBridges,    CorArgOpt, NULL,  NULL,  NULL,      "bridge plugins to host (comma-separated)" },
  { "--bridgeConfig",    "-brc",CorArgString, _vp &ftBridgeConfig, CorArgOpt, NULL,    NULL,  NULL,      "bridge configuration file (the broker's own)" },
  CORARGS_END
};



// -----------------------------------------------------------------------------
//
// dumpInit - create the accumulator array
//
static void dumpInit(void)
{
  dumpArray  = corTreeArray(NULL, NULL);
  dumpCount  = 0;
  probeCount = 0;
}



// -----------------------------------------------------------------------------
//
// dumpDrain - free the accumulated entries and start a fresh, empty array
//
// Not dumpClear: probeCount counts sourceIdentity probes, which are deliberately
// kept OUT of dumpArray and are not "read" by a dump. Draining the one must not
// reset the other.
//
// Caller holds dumpMutex.
//
static void dumpDrain(void)
{
  if (dumpArray != NULL)
    corTreeFree(dumpArray);

  dumpArray = corTreeArray(NULL, NULL);
  dumpCount = 0;
}



// -----------------------------------------------------------------------------
//
// ftBridgeSampleIn - a sample arrived on a transport
//
// ⚠ A PLUGIN THREAD. The entry is built with the malloc allocator, like every
// other dump entry, so it outlives whatever the transport is about to reuse.
//
// This is the whole of corTestClient's side of the seam: the broker's
// implementation stores an attribute, this one writes down that the value
// arrived. Same contract, different job, which is the point of the contract
// being about endpoints and bytes rather than entities.
//
static int ftBridgeSampleIn(const char* bridgeName, const char* endpoint, const char* json, int64_t publishTime)
{
  CorNode* entry = corTreeObject(NULL, NULL);

  corTreeChildAdd(entry, corTreeString(NULL, "bridge", (bridgeName != NULL) ? bridgeName : "?"));
  corTreeChildAdd(entry, corTreeString(NULL, "endpoint", (endpoint != NULL) ? endpoint : "?"));
  corTreeChildAdd(entry, corTreeString(NULL, "payload", (json != NULL) ? json    : ""));

  if (publishTime > 0)
    corTreeChildAdd(entry, corTreeInteger(NULL, "publishTime", (long long) publishTime));

  //
  // dumpMutex, like every other path that touches dumpArray - and this one is
  // reached from a transport thread, which is the case it exists for.
  //
  pthread_mutex_lock(&dumpMutex);
  if (dumpArray == NULL)
    dumpArray = corTreeArray(NULL, NULL);
  corTreeChildAdd(dumpArray, entry);
  ++dumpCount;
  pthread_mutex_unlock(&dumpMutex);

  return BRIDGE_OK;
}



//
// Defined below, beside the rest of the bridge plumbing; the service handler
// above it is the one caller that comes first.
//
static BridgeDriver* ftBridgeDriverLookup(const char* alias);



// -----------------------------------------------------------------------------
//
// Canned service replies.
//
// ⭐ corTestClient IS THE PEER, and a peer to a service has to have an answer.
// Computing one is exactly what a context broker cannot do and what a test must
// not have to do either, so the answer is told to corTestClient in advance: the test
// says "when somebody asks this service, say that", then makes the broker ask.
//
// Which also keeps the assertion where it belongs. What the test is checking is
// that the request left the broker and the reply came back into the attribute -
// not that anything clever happened in between.
//
#define FT_SERVICE_MAX  8

typedef struct FtServiceReply
{
  char*  endpoint;
  char*  payload;                                      // rendered JSON, malloc'd
} FtServiceReply;

static FtServiceReply  ftServiceReplyV[FT_SERVICE_MAX];
static int             ftServiceReplyCount = 0;
static pthread_mutex_t ftServiceMutex      = PTHREAD_MUTEX_INITIALIZER;



// -----------------------------------------------------------------------------
//
// ftServiceReplyLookup - the answer set for an endpoint, or NULL
//
// Returns a COPY: the caller is on a transport thread and the table can be
// rewritten by a REST thread between this call and the send.
//
static char* ftServiceReplyLookup(const char* endpoint)
{
  char* copy = NULL;

  pthread_mutex_lock(&ftServiceMutex);

  for (int i = 0; i < ftServiceReplyCount; i++)
  {
    if (strcmp(ftServiceReplyV[i].endpoint, endpoint) == 0)
    {
      copy = strdup(ftServiceReplyV[i].payload);
      break;
    }
  }

  pthread_mutex_unlock(&ftServiceMutex);

  return copy;
}



// -----------------------------------------------------------------------------
//
// ftBridgeServiceRequest - the broker asked us something
//
// ⚠ A PLUGIN THREAD, as ftBridgeSampleIn.
//
// Two things happen, and both matter to a test: the request is written down, so
// it can be asserted on exactly as an arriving sample is, and it is answered,
// so the other half of the exchange can be asserted on in the broker.
//
static int ftBridgeServiceRequest(const char* bridgeName,
                                  const char* endpoint,
                                  const char* json,
                                  uint64_t    requestId,
                                  int64_t     publishTime)
{
  CorNode* entry = corTreeObject(NULL, NULL);

  corTreeChildAdd(entry, corTreeString(NULL, "bridge", (bridgeName != NULL) ? bridgeName : "?"));
  corTreeChildAdd(entry, corTreeString(NULL, "endpoint", (endpoint != NULL) ? endpoint : "?"));
  corTreeChildAdd(entry, corTreeString(NULL, "kind", "serviceRequest"));
  corTreeChildAdd(entry, corTreeString(NULL, "payload", (json != NULL) ? json     : ""));

  if (publishTime > 0)
    corTreeChildAdd(entry, corTreeInteger(NULL, "publishTime", (long long) publishTime));

  pthread_mutex_lock(&dumpMutex);
  if (dumpArray == NULL)
    dumpArray = corTreeArray(NULL, NULL);
  corTreeChildAdd(dumpArray, entry);
  ++dumpCount;
  pthread_mutex_unlock(&dumpMutex);

  //
  // ⛔ The requestId is NOT written into the dump. It is the transport's own
  // counter, it runs across every service in the process, and a test that
  // asserted on it would be asserting on how many requests happened to precede
  // this one.
  //
  char* reply = ftServiceReplyLookup(endpoint);

  if (reply == NULL)
  {
    COR_W("corTestClient: asked on '%s' with no answer set - the request stands unanswered", endpoint);
    return BRIDGE_OK;
  }

  BridgeDriver*       driverP = ftBridgeDriverLookup((bridgeName != NULL) ? bridgeName : "dds");
  const BridgeServer* serverP = ((driverP != NULL) && (driverP->serverIface != NULL)) ? driverP->serverIface() : NULL;

  if ((serverP != NULL) && (serverP->serviceReply != NULL))
  {
    if (serverP->serviceReply(endpoint, requestId, reply) != BRIDGE_OK)
      COR_W("corTestClient: could not answer on '%s'", endpoint);
    else
      COR_T(1, "corTestClient: answered on '%s' with %s", endpoint, reply);
  }

  free(reply);

  return BRIDGE_OK;
}



// -----------------------------------------------------------------------------
//
// ftBridgeLog - the transport library's own log messages
//
static void ftBridgeLog(int severity, const char* fileName, int lineNo, const char* funcName, const char* msg)
{
  char sev;

  switch (severity)
  {
  case BRIDGE_LOG_ERROR:    sev = 'E';  break;
  case BRIDGE_LOG_WARNING:  sev = 'W';  break;
  case BRIDGE_LOG_INFO:     sev = 'I';  break;
  default:                  sev = 'T';  break;
  }

  corLogOut((char*) fileName, lineNo, (char*) funcName, sev, -1, "%s", msg);
}



// -----------------------------------------------------------------------------
//
// ftBridgeQualifiedIn - a reply, or anything else that is not a plain sample
//
// corTestClient hosts the plugin as the broker does, so it is offered the same
// arrivals. It writes them down with their qualifiers, which is what lets a
// test tell "the answer came back" from "a value arrived".
//
// ⚠ A PLUGIN THREAD, as ftBridgeSampleIn.
//
static int ftBridgeQualifiedIn(const char* bridgeName,
                               const char* endpoint,
                               const char* datasetId,
                               const char* subAttrName,
                               const char* json,
                               int64_t     publishTime)
{
  CorNode* entry = corTreeObject(NULL, NULL);

  corTreeChildAdd(entry, corTreeString(NULL, "bridge", (bridgeName != NULL) ? bridgeName : "?"));
  corTreeChildAdd(entry, corTreeString(NULL, "endpoint", (endpoint != NULL) ? endpoint : "?"));
  corTreeChildAdd(entry, corTreeString(NULL, "payload", (json != NULL) ? json    : ""));

  if (subAttrName != NULL)
    corTreeChildAdd(entry, corTreeString(NULL, "subAttribute", (char*) subAttrName));

  if (datasetId != NULL)
    corTreeChildAdd(entry, corTreeString(NULL, "datasetId", (char*) datasetId));

  if (publishTime > 0)
    corTreeChildAdd(entry, corTreeInteger(NULL, "publishTime", (long long) publishTime));

  pthread_mutex_lock(&dumpMutex);
  if (dumpArray == NULL)
    dumpArray = corTreeArray(NULL, NULL);
  corTreeChildAdd(dumpArray, entry);
  ++dumpCount;
  pthread_mutex_unlock(&dumpMutex);

  return BRIDGE_OK;
}



static BridgeBroker ftBridgeBroker =
{
  BRIDGE_ABI_VERSION,
  ftBridgeSampleIn,
  ftBridgeLog,
  ftBridgeQualifiedIn
};



// -----------------------------------------------------------------------------
//
// ftBridgeDriverLookup - which loaded plugin answers to this name
//
static BridgeDriver* ftBridgeDriverLookup(const char* alias)
{
  for (int i = 0; i < ftBridgeCount; i++)
  {
    if ((ftBridgeV[i].alias != NULL) && (strcmp(ftBridgeV[i].alias, alias) == 0))
      return &ftBridgeV[i];
  }

  return (ftBridgeCount > 0) ? &ftBridgeV[0] : NULL;   // one bridge is the common case
}



// -----------------------------------------------------------------------------
//
// ftBridgeTopicsCarry - subscribe to every endpoint named in the config file
//
// The same file the broker reads. corTestClient has no Channels and does not want
// any - it is the peer on the other end of the wire, not a second broker - but
// it does need to know which endpoints to listen on, and the file already says.
//
static void ftBridgeTopicsCarry(BridgeDriver* driverP, const char* configFile)
{
  char*  buf    = NULL;
  int    bufLen = 0;

  if ((driverP->channelAdd == NULL) || (configFile == NULL))
    return;

  if (corFileRead((char*) "", (char*) configFile, &buf, &bufLen) != 0)
    return;

  //
  // Its own buffer. This runs BEFORE corRestInit, so corRest.corJsonP is still
  // NULL - parsing into it takes the process down before it serves anything,
  // which is a confusing way to be told the order is wrong.
  //
  char      kallocBuffer[8192];
  CorAlloc  kalloc;
  CorJson   corJson;

  corAllocBufferInit(&kalloc, kallocBuffer, sizeof(kallocBuffer), 8 * 1024, NULL, "ftBridge");

  CorJson* corJsonP = corJsonCreate(&corJson, &kalloc);
  CorNode* treeP = corJsonParse(corJsonP, buf);

  if (treeP == NULL)
  {
    corAllocBufferReset(&kalloc, true);
    return;
  }

  CorNode* bridgeP  = corTreeLookup(treeP, (driverP->alias != NULL) ? driverP->alias : "dds");
  CorNode* ngsildP  = (bridgeP != NULL) ? corTreeLookup(bridgeP, "ngsild") : NULL;
  CorNode* topicsP  = (ngsildP != NULL) ? corTreeLookup(ngsildP, "topics") : NULL;
  CorNode* servicesP = (ngsildP != NULL) ? corTreeLookup(ngsildP, "services") : NULL;

  for (CorNode* entryP = (topicsP != NULL) ? topicsP->value.head : NULL; entryP != NULL; entryP = entryP->next)
  {
    //
    // BOTH directions: corTestClient stands in for whatever is at the far end, and
    // the far end both publishes and listens.
    //
    if (driverP->channelAdd(entryP->name, BridgeChannelTopic, BridgeDirectionBoth) != BRIDGE_OK)
      COR_W("corTestClient: bridge '%s' would not carry '%s'", driverP->alias, entryP->name);
  }

  //
  // ⭐ AND THE SERVICES, THE OTHER WAY UP.
  //
  // A topic is carried; a service is SERVED. The broker reads these same
  // entries and becomes the client of each - it is the only thing it can be -
  // so for the exchange to exist at all, something has to be the server, and
  // that is what corTestClient is for. Without it there is nothing on the domain to
  // answer and a service test could only ever assert that nothing happened.
  //
  const BridgeServer* serverP = (driverP->serverIface != NULL) ? driverP->serverIface() : NULL;

  if ((servicesP != NULL) && ((serverP == NULL) || (serverP->serviceServe == NULL)))
    COR_W("corTestClient: bridge '%s' has services configured but cannot serve them", driverP->alias);
  else if (servicesP != NULL)
  {
    for (CorNode* entryP = servicesP->value.head; entryP != NULL; entryP = entryP->next)
    {
      if (serverP->serviceServe(entryP->name, ftBridgeServiceRequest) != BRIDGE_OK)
        COR_W("corTestClient: bridge '%s' would not serve '%s'", driverP->alias, entryP->name);
    }
  }

  corAllocBufferReset(&kalloc, true);
}



// -----------------------------------------------------------------------------
//
// ftBridgesLoad - dlopen each named plugin and bring its transport up
//
static void ftBridgesLoad(void)
{
  if (ftBridges == NULL)
    return;

  corPluginSetBaseDir("/opt/seamware/plugins", "SEAMWARE_PLUGIN_DIR");

  char buf[512];
  strncpy(buf, ftBridges, sizeof(buf) - 1);
  buf[sizeof(buf) - 1] = 0;

  char* saveptr = NULL;
  char* token   = strtok_r(buf, ",", &saveptr);

  while ((token != NULL) && (ftBridgeCount < BRIDGES_MAX))
  {
    char path[512];
    char openErr[512];

    corPluginResolve(corPluginBaseDir(), "bridge", NULL, token, path, sizeof(path));

    BridgeRegisterFunc registerFunc = (BridgeRegisterFunc) corPluginOpen(path, "bridgeRegister", openErr, sizeof(openErr));

    if (registerFunc == NULL)
    {
      COR_X(1, "corTestClient: bridge plugin '%s' (%s): %s", token, path, openErr);
      return;
    }

    BridgeDriver* driverP = &ftBridgeV[ftBridgeCount];

    memset(driverP, 0, sizeof(BridgeDriver));

    //
    // What THIS host speaks, so a plugin built against a newer contract knows
    // where our struct ends - the same handshake the broker's loader does, and
    // for the same reason. See BridgeDriver.h.
    //
    driverP->abiVersion = BRIDGE_ABI_VERSION;

    registerFunc(driverP);
    ++ftBridgeCount;

    if ((driverP->init != NULL) && (driverP->init(ftBridgeConfig, &ftBridgeBroker) != BRIDGE_OK))
      COR_X(1, "corTestClient: init failed for bridge '%s'", token);

    ftBridgeTopicsCarry(driverP, ftBridgeConfig);

    COR_I("corTestClient: hosting bridge '%s'", (driverP->alias != NULL) ? driverP->alias : token);

    token = strtok_r(NULL, ",", &saveptr);
  }
}



// -----------------------------------------------------------------------------
//
// postBridgePublish - POST /bridge/publish
//
// Body: { "endpoint": "rt/pose", "payload": <any json> }
//
// Puts a value on a transport, which is how a test plays the far end: the
// broker is then supposed to store it, and the test asks the broker.
//
static bool postBridgePublish(void)
{
  CorNode* bodyP = corRest.in.requestTree;

  if ((ftBridgeCount == 0) || (bodyP == NULL))
  {
    corRest.out.httpStatusCode = 400;
    return true;
  }

  CorNode* endpointP = corTreeLookup(bodyP, "endpoint");
  CorNode* payloadP = corTreeLookup(bodyP, "payload");

  if ((endpointP == NULL) || (endpointP->type != CorString) || (payloadP == NULL))
  {
    corRest.out.httpStatusCode = 400;
    return true;
  }

  //
  // The payload goes on the wire as the application's own JSON, so the member
  // name and the sibling link come off for the render - see
  // bridgeRequestsBeforeWrite in the broker, where the same two things bit.
  //
  static char   rendered[16384];
  char*         savedName = payloadP->name;
  CorNode*      savedNext = payloadP->next;

  payloadP->name = NULL;
  payloadP->next = NULL;
  corJsonFastRender(payloadP, rendered);
  payloadP->name = savedName;
  payloadP->next = savedNext;

  BridgeDriver* driverP = ftBridgeDriverLookup("dds");

  if ((driverP == NULL) || (driverP->publish == NULL))
  {
    corRest.out.httpStatusCode = 501;
    return true;
  }

  corRest.out.httpStatusCode = (driverP->publish(endpointP->value.s, rendered) == BRIDGE_OK) ? 204 : 502;

  return true;
}



// -----------------------------------------------------------------------------
//
// dumpAccumulate - record an incoming request
//
// Stores: { "verb": "POST", "url": "/notify", "payload": "..." }
// Uses malloc allocator (NULL) so entries persist across requests.
//
static void dumpAccumulate(void)
{
  //
  // All strings come from per-request allocators that will be freed.
  // Use corTreeString with NULL allocator (malloc) — it strdup's the value.
  //
  CorNode* entry = corTreeObject(NULL, NULL);

  corTreeChildAdd(entry, corTreeString(NULL, "verb", corRest.in.verbString ? corRest.in.verbString : "?"));
  corTreeChildAdd(entry, corTreeString(NULL, "url", corRest.in.urlPath ? corRest.in.urlPath : "?"));

  // Headers
  CorNode* hdrs = corTreeObject(NULL, "headers");
  for (int i = 0; i < corRest.in.httpHeaderCount; i++)
    corTreeChildAdd(hdrs, corTreeString(NULL, corRest.in.httpHeaderV[i].key, corRest.in.httpHeaderV[i].value));
  corTreeChildAdd(entry, hdrs);

  // URI params
  if (corRest.in.uriParamCount > 0)
  {
    CorNode* params = corTreeObject(NULL, "params");
    for (int i = 0; i < corRest.in.uriParamCount; i++)
      corTreeChildAdd(params, corTreeString(NULL, corRest.in.uriParamV[i].key, corRest.in.uriParamV[i].value));
    corTreeChildAdd(entry, params);
  }

  // Body — deep-clone the request tree since the original is per-request allocated
  if (corRest.in.requestTree != NULL)
  {
    CorNode* bodyClone = corTreeClone(NULL, corRest.in.requestTree);
    if (bodyClone != NULL)
    {
      bodyClone->name = (char*) "body";
      corTreeChildAdd(entry, bodyClone);
    }
  }
  else if (corRest.in.payload != NULL && corRest.in.payloadSize > 0)
    corTreeChildAdd(entry, corTreeString(NULL, "body", corRest.in.payload));

  pthread_mutex_lock(&dumpMutex);
  corTreeChildAdd(dumpArray, entry);
  dumpCount++;
  pthread_mutex_unlock(&dumpMutex);
}



// -----------------------------------------------------------------------------
//
// getDump - GET /dump — the requests received since the last dump, and consume
//
static bool getDump(void)
{
  pthread_mutex_lock(&dumpMutex);

  if (dumpArray == NULL || dumpCount == 0)
  {
    // Return empty array
    corRest.out.payload     = (char*) "[]";
    corRest.out.payloadSize = 2;
    pthread_mutex_unlock(&dumpMutex);
    return true;
  }

  //
  // Render dumpArray to JSON.
  // Use a scratch buffer allocated via malloc so the output survives
  // beyond the per-request kalloc lifetime — corRest copies the response
  // body (MHD_RESPMEM_MUST_COPY), so we can free this buffer after
  // the call returns by using corAlloc from the per-request allocator.
  //
  int   bufSize = corJsonFastRenderSize(dumpArray) + 1;
  char* buf     = (char*) corAlloc(&corRest.kalloc, bufSize);

  corJsonFastRender(dumpArray, buf);

  //
  // A dump READS the notifications, and read notifications are gone. What the
  // accumulator holds from here on is what arrived SINCE - so the next dump in
  // the test answers "what did this step produce", not "everything so far".
  //
  // That is what makes an unexpected notification visible. While a dump was
  // cumulative, an extra one silently padded every later expectation, and the
  // test that would have caught it was the one that had already passed. It also
  // retires DELETE /dump: a test no longer has to throw the accumulator away to
  // get a clean read, which was the same error with an explicit verb on it.
  //
  // buf already holds the rendered text, so freeing the tree here is safe.
  //
  dumpDrain();

  pthread_mutex_unlock(&dumpMutex);

  corRest.out.payload     = buf;
  corRest.out.payloadSize = strlen(buf);

  return true;
}



// -----------------------------------------------------------------------------
//
// getCount - GET /count — number of accumulated requests as a bare integer
//
// A parser-free alternative to GET /dump for the (many) tests that only need
// the count of received requests: returns e.g. "3" (or "0" when empty) as plain
// text, never JSON. A caller reads it with plain curl and never has to pipe a
// possibly-empty/invalid body through a JSON parser — which, on an empty dump,
// would throw to stderr and flake the test under parallel load.
//
static bool getCount(void)
{
  char* buf = (char*) corAlloc(&corRest.kalloc, 16);

  pthread_mutex_lock(&dumpMutex);
  snprintf(buf, 16, "%d", dumpCount);
  pthread_mutex_unlock(&dumpMutex);

  corRest.out.payload     = buf;
  corRest.out.payloadSize = strlen(buf);

  return true;
}



// -----------------------------------------------------------------------------
//
// getMqttReady - GET /mqttReady — 1 once the MQTT SUBACK has arrived.
//
// A bare integer, like /count: the harness polls this from bash and a JSON
// parser has no business in a readiness barrier. Answers 1 when no MQTT was
// asked for, so a caller need not special-case that.
//
// Three answers, not two:
//   1  subscribed, or none was asked for
//   0  not yet - keep polling
//  -1  given up on; polling longer cannot help, and the log says why
//
// -1 is what turns a barrier timeout into an immediate, explained failure.
//
static bool getMqttReady(void)
{
  char* buf = (char*) corAlloc(&corRest.kalloc, 4);

  int state = ((ftMqttPort == 0) || (ftMqttSubscribed == true))?  1 :
              (ftMqttFailed == true)?                            -1 : 0;

  snprintf(buf, 4, "%d", state);
  corRest.out.payload     = buf;
  corRest.out.payloadSize = strlen(buf);

  return true;
}



// -----------------------------------------------------------------------------
//
// getProbeCount - GET /probeCount — number of sourceIdentity probes seen.
//
// Probes are infrastructure (§ 5.15 alias discovery) and are deliberately kept
// out of the request dump, so a test can't count them there. This bare-integer
// endpoint lets a test assert the probe fired (no contextSourceAlias supplied)
// or did NOT fire (alias supplied in the registration → probe skipped).
//
static bool getProbeCount(void)
{
  char* buf = (char*) corAlloc(&corRest.kalloc, 16);

  snprintf(buf, 16, "%d", probeCount);
  corRest.out.payload     = buf;
  corRest.out.payloadSize = strlen(buf);

  return true;
}



// -----------------------------------------------------------------------------
//
// getDie - GET /die — exit the process
//
static bool getDie(void)
{
  corRest.out.httpStatusCode = 200;
  corRest.out.payload        = (char*) "bye";
  corRest.out.payloadSize    = 3;

  // Schedule exit after the response is sent
  // (MHD_RESPMEM_MUST_COPY means corRest copies the buffer before we exit)
  _exit(0);

  return true;  // unreachable
}



// -----------------------------------------------------------------------------
//
// ftStubMatch - first stub matching the verb (exact) and path (substring)
//
static FtStub* ftStubMatch(const char* verb, const char* path)
{
  if (verb == NULL || path == NULL)
    return NULL;

  for (FtStub* s = ftStubs; s != NULL; s = s->next)
    if ((strcmp(s->verb, verb) == 0) && (strstr(path, s->path) != NULL))
      return s;

  return NULL;
}



// -----------------------------------------------------------------------------
//
// stubServe - if a stub matches the current request, serve it (return true)
//
static bool stubServe(void)
{
  FtStub* s = ftStubMatch(corRest.in.verbString, corRest.in.urlPath);
  if (s == NULL)
    return false;

  // Per-path delay — lets a test time out ONLY a specific endpoint (e.g. the
  // sourceIdentity probe) without the global --delay slowing the forwards too.
  if (s->delayMs > 0)
    usleep(s->delayMs * 1000);

  corRest.out.httpStatusCode = s->status;

  if (s->body != NULL)
  {
    int   n   = strlen(s->body);
    char* buf = (char*) corAlloc(&corRest.kalloc, n + 1);
    memcpy(buf, s->body, n + 1);
    corRest.out.payload     = buf;
    corRest.out.payloadSize = n;
  }

  COR_T(1, "stub served: %s %s -> %d (bodyLen=%d)", s->verb, corRest.in.urlPath, s->status, s->body ? (int) strlen(s->body) : 0);
  return true;
}



// -----------------------------------------------------------------------------
//
// postMockReply - POST /mock/reply — program a stubbed response
//
// Body: { "verb": "POST", "path": "/attrs/", "status": 207, "body": {...} }
// verb/path/status default to "POST" / "/" / 200; body is optional.
//
static bool postMockReply(void)
{
  CorNode* tree = corRest.in.requestTree;
  if (tree == NULL)
  {
    corRest.out.httpStatusCode = 400;
    return true;
  }

  CorNode* verbP  = corTreeLookup(tree, "verb");
  CorNode* pathP  = corTreeLookup(tree, "path");
  CorNode* statusP = corTreeLookup(tree, "status");
  CorNode* delayP = corTreeLookup(tree, "delayMs"); // sleep before responding (per-path timeout injection)
  CorNode* bodyP  = corTreeLookup(tree, "body");
  CorNode* rawP   = corTreeLookup(tree, "raw"); // verbatim body (e.g. malformed JSON for a 502 test)

  FtStub* s = (FtStub*) calloc(1, sizeof(FtStub));
  s->verb    = strdup(((verbP != NULL) && (verbP->type == CorString)) ? verbP->value.s : "POST");
  s->path    = strdup(((pathP != NULL) && (pathP->type == CorString)) ? pathP->value.s : "/");
  s->status  = ((statusP != NULL) && (statusP->type == CorInt)) ? (int) statusP->value.i : 200;
  s->delayMs = ((delayP  != NULL) && (delayP->type  == CorInt)) ? (int) delayP->value.i : 0;
  s->body    = NULL;

  if ((rawP != NULL) && (rawP->type == CorString))
    s->body = strdup(rawP->value.s);
  else if (bodyP != NULL)
  {
    bodyP->name = NULL;   // render the value alone, not  "body": {...}
    int n = corJsonFastRenderSize(bodyP) + 1;
    s->body = (char*) malloc(n);
    corJsonFastRender(bodyP, s->body);
  }

  s->next = ftStubs;
  ftStubs = s;

  COR_T(1, "stub programmed: %s <path-contains %s> -> %d", s->verb, s->path, s->status);
  corRest.out.httpStatusCode = 201;
  return true;
}



// -----------------------------------------------------------------------------
//
// deleteMockReply - DELETE /mock/reply — clear all programmed stubs
//
static bool deleteMockReply(void)
{
  while (ftStubs != NULL)
  {
    FtStub* next = ftStubs->next;
    free(ftStubs->verb);
    free(ftStubs->path);
    free(ftStubs->body);
    free(ftStubs);
    ftStubs = next;
  }

  corRest.out.httpStatusCode = 204;
  return true;
}



// -----------------------------------------------------------------------------
//
// postAccumulate - POST /** — catch-all, accumulate the request, return 200
//
static bool postAccumulate(void)
{
  //
  // --discard: a benchmark's notification receiver - counted (GET /count), kept nowhere. Accumulating
  // made the receiver grow with every notification, and a long run measured the receiver.
  //
  if (ftDiscard)
  {
    pthread_mutex_lock(&dumpMutex);
    ++dumpCount;
    pthread_mutex_unlock(&dumpMutex);

    corRest.out.httpStatusCode = ftPostStatus;
    return true;
  }

  dumpAccumulate();
  COR_T(1, "POST %s received (total: %d, status=%u, delay=%ums)", corRest.in.urlPath, dumpCount, ftPostStatus, ftDelayMs);

  if (ftDelayMs > 0)
    usleep(ftDelayMs * 1000);

  // A programmed stub (POST /mock/reply) wins over the --status fallback.
  if (stubServe())
    return true;

  corRest.out.httpStatusCode = ftPostStatus;

  //
  // For 4xx/5xx responses, emit an NGSI-LD ProblemDetails body so coraine's
  // forward-failure surfacing has something spec-shaped to log / include
  // as the "reason" in notCreated entries.
  //
  if (ftPostStatus >= 400)
  {
    static char errBuf[256];
    snprintf(errBuf, sizeof(errBuf),
             "{\"type\":\"https://uri.etsi.org/ngsi-ld/errors/InternalError\","
             "\"title\":\"Mock Error\","
             "\"detail\":\"corTestClient configured with --status %u\"}", ftPostStatus);
    corRest.out.payload     = errBuf;
    corRest.out.payloadSize = strlen(errBuf);
  }

  return true;
}



// -----------------------------------------------------------------------------
//
// getAccumulate - GET /** — catch-all for forwarded queries (queryEntities,
//                 retrieveEntity), accumulates the request and returns an
//                 empty entity array. Honors --status like postAccumulate
//                 so timeout / error simulations work for GETs too.
//
// /info/sourceIdentity is a discovery probe (broker learns the CSR's
// alias for loop-detection at registration time) — infrastructure, not
// a forwarded operation. Don't pollute the dump with it.
//
static bool getAccumulate(void)
{
  // Match the sourceIdentity probe by SUFFIX: the resource lives under the API
  // root, so the broker sends "<endpoint>/ngsi-ld/v1/info/sourceIdentity". A
  // bare-path match (old "/info/sourceIdentity") silently stops recognizing the
  // probe once the caller carries the /ngsi-ld/v1 prefix, letting it pollute the
  // dump. Suffix-match covers both spellings.
  static const char probeSuffix[] = "/info/sourceIdentity";
  bool isProbe = false;
  if (corRest.in.urlPath != NULL)
  {
    int urlLen    = (int) strlen(corRest.in.urlPath);
    int suffixLen = (int) (sizeof(probeSuffix) - 1);
    isProbe = (urlLen >= suffixLen) &&
              (strcmp(corRest.in.urlPath + urlLen - suffixLen, probeSuffix) == 0);
  }

  if (isProbe)
    probeCount++;
  else
    dumpAccumulate();
  COR_T(1, "GET %s received (probe=%d total: %d, status=%u, delay=%ums)",
        corRest.in.urlPath, isProbe, dumpCount, ftPostStatus, ftDelayMs);

  if (ftDelayMs > 0)
    usleep(ftDelayMs * 1000);

  // A programmed stub (POST /mock/reply) wins over the --status fallback.
  if (stubServe())
    return true;

  corRest.out.httpStatusCode = (ftPostStatus == 201) ? 200 : ftPostStatus;

  if (ftPostStatus >= 400)
  {
    static char errBuf[256];
    snprintf(errBuf, sizeof(errBuf),
             "{\"type\":\"https://uri.etsi.org/ngsi-ld/errors/InternalError\","
             "\"title\":\"Mock Error\","
             "\"detail\":\"corTestClient configured with --status %u\"}", ftPostStatus);
    corRest.out.payload     = errBuf;
    corRest.out.payloadSize = strlen(errBuf);
  }
  else
  {
    corRest.out.payload     = (char*) "[]";
    corRest.out.payloadSize = 2;
  }

  return true;
}



// -----------------------------------------------------------------------------
//
// MQTT subscriber — runs in its own thread, accumulates received messages
// into the same dumpArray under verb="MQTT", url=<topic>, body=<payload>.
//
static void mqttOnConnect(struct mosquitto* m, void* ud, int rc)
{
  (void) ud;

  //
  // rc is the CONNACK return code, and it was being ignored. A refused CONNECT -
  // bad credentials, most usefully - completes the TCP connection and then fails
  // at the MQTT level, so subscribing anyway produced no SUBACK and no reason.
  //
  if (rc != 0)
  {
    fprintf(stderr, "corTestClient: MQTT broker on port %d refused the connection: %s\n",
            ftMqttPort, mosquitto_connack_string(rc));
    fflush(stderr);
    ftMqttFailed = true;
    return;
  }

  mosquitto_subscribe(m, NULL, ftMqttTopic, 0);
}



// -----------------------------------------------------------------------------
//
// mqttOnSubscribe - the SUBACK. From here on, a publish on the topic arrives.
//
static void mqttOnSubscribe(struct mosquitto* m, void* ud, int mid, int qosCount, const int* grantedQos)
{
  (void) m; (void) ud; (void) mid; (void) qosCount; (void) grantedQos;
  ftMqttSubscribed = true;
}

static void mqttOnMessage(struct mosquitto* m, void* ud, const struct mosquitto_message* msg)
{
  (void) m; (void) ud;
  if (msg == NULL || msg->payload == NULL) return;

  CorNode* entry = corTreeObject(NULL, NULL);
  corTreeChildAdd(entry, corTreeString(NULL, "verb", "MQTT"));
  corTreeChildAdd(entry, corTreeString(NULL, "url", msg->topic ? msg->topic : ""));

  // Try to parse payload as JSON; fall back to a string body.
  char* payloadStr = (char*) malloc(msg->payloadlen + 1);
  if (payloadStr == NULL) return;
  memcpy(payloadStr, msg->payload, msg->payloadlen);
  payloadStr[msg->payloadlen] = '\0';

  // Store the payload as a string body — same shape as HTTP non-JSON
  // bodies. Tests typically grep on the payload text directly.
  corTreeChildAdd(entry, corTreeString(NULL, "body", payloadStr));
  free(payloadStr);

  pthread_mutex_lock(&dumpMutex);
  corTreeChildAdd(dumpArray, entry);
  dumpCount++;
  pthread_mutex_unlock(&dumpMutex);
}

static void* mqttListenerThread(void* arg)
{
  (void) arg;

  if (mosquitto_lib_init() != MOSQ_ERR_SUCCESS)
    return NULL;

  struct mosquitto* mosq = mosquitto_new(NULL, true, NULL);
  if (mosq == NULL) return NULL;

  if (ftMqttUser != NULL)
    mosquitto_username_pw_set(mosq, ftMqttUser, ftMqttPass);

  // --mqttTls: subscribe over TLS, accepting a self-signed broker cert
  // (test rig only — same insecure posture as the broker's --insecureNotif).
  if (ftMqttTls)
  {
    mosquitto_tls_set(mosq, NULL, "/etc/ssl/certs", NULL, NULL, NULL);
    mosquitto_tls_opts_set(mosq, 0 /* SSL_VERIFY_NONE */, NULL, NULL);
    mosquitto_tls_insecure_set(mosq, true);
  }

  mosquitto_connect_callback_set(mosq, mqttOnConnect);
  mosquitto_subscribe_callback_set(mosq, mqttOnSubscribe);
  mosquitto_message_callback_set(mosq, mqttOnMessage);

  //
  // Retry, but not forever. The broker may legitimately not be up yet, so this
  // keeps trying - and if it never appears, it says so and stops, rather than
  // spinning in silence and leaving the reader of the log nothing to go on.
  //
  // 5 seconds: far longer than a local mosquitto needs even on a loaded runner,
  // and comfortably inside the readiness barrier that is waiting on this, so the
  // barrier hears the verdict instead of timing out over the top of it.
  //
  int rc = MOSQ_ERR_SUCCESS;

  for (int attempt = 0; attempt < 25; attempt++)          // 25 x 200ms = 5s
  {
    rc = mosquitto_connect(mosq, "localhost", ftMqttPort, 60);
    if (rc == MOSQ_ERR_SUCCESS)
      break;
    usleep(200000);
  }

  if (rc != MOSQ_ERR_SUCCESS)
  {
    fprintf(stderr, "corTestClient: no MQTT broker on port %d after 5s: %s\n",
            ftMqttPort, mosquitto_strerror(rc));
    fflush(stderr);
    ftMqttFailed = true;
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
    return NULL;
  }

  ftMqttConn = mosq;
  mosquitto_loop_forever(mosq, -1, 1);
  ftMqttConn = NULL;

  mosquitto_destroy(mosq);
  mosquitto_lib_cleanup();
  return NULL;
}



// -----------------------------------------------------------------------------
//
// postBridgeServiceReply - POST /bridge/serviceReply
//
// Body: { "endpoint": "add_two_ints", "payload": <any json> }
//
// What corTestClient will answer when the broker asks that service. Set before the
// broker is made to ask; replacing it replaces the answer.
//
static bool postBridgeServiceReply(void)
{
  CorNode* bodyP = corRest.in.requestTree;

  if (bodyP == NULL)
  {
    corRest.out.httpStatusCode = 400;
    return true;
  }

  CorNode* endpointP = corTreeLookup(bodyP, "endpoint");
  CorNode* payloadP = corTreeLookup(bodyP, "payload");

  if ((endpointP == NULL) || (endpointP->type != CorString) || (payloadP == NULL))
  {
    corRest.out.httpStatusCode = 400;
    return true;
  }

  //
  // The name and the sibling link come off for the render - the payload goes on
  // the wire as the application's own JSON. Same two things that bite in
  // postBridgePublish and in the broker's own outbound path.
  //
  static char rendered[16384];
  char*       savedName = payloadP->name;
  CorNode*    savedNext = payloadP->next;

  payloadP->name = NULL;
  payloadP->next = NULL;
  corJsonFastRender(payloadP, rendered);
  payloadP->name = savedName;
  payloadP->next = savedNext;

  pthread_mutex_lock(&ftServiceMutex);

  int slot = -1;

  for (int i = 0; i < ftServiceReplyCount; i++)
  {
    if (strcmp(ftServiceReplyV[i].endpoint, endpointP->value.s) == 0)
    {
      slot = i;
      break;
    }
  }

  if (slot == -1)
  {
    if (ftServiceReplyCount >= FT_SERVICE_MAX)
    {
      pthread_mutex_unlock(&ftServiceMutex);
      corRest.out.httpStatusCode = 507;
      return true;
    }

    slot = ftServiceReplyCount++;
    ftServiceReplyV[slot].endpoint = strdup(endpointP->value.s);
  }
  else
    free(ftServiceReplyV[slot].payload);

  ftServiceReplyV[slot].payload = strdup(rendered);

  pthread_mutex_unlock(&ftServiceMutex);

  COR_T(1, "corTestClient: '%s' will be answered with %s", endpointP->value.s, rendered);

  corRest.out.httpStatusCode = 204;

  return true;
}



// -----------------------------------------------------------------------------
//
// postMqttPublish - POST /mqtt/publish: put a message on the test MQTT broker
//
//   { "topic": "plant/tank1/level", "payload": 42.5 }       - any JSON, published as its text
//   { "topic": "plant/tank1/state", "text": "running" }     - published exactly as given (not JSON)
//
// What a device on the bus would do, for the bridge tests: the broker's mqtt.so is subscribed to the
// topic and turns the message into an attribute write.
//
static bool postMqttPublish(void)
{
  CorNode* bodyP = corRest.in.requestTree;
  CorNode* topicP   = (bodyP != NULL) ? corTreeLookup(bodyP, "topic")   : NULL;
  CorNode* payloadP = (bodyP != NULL) ? corTreeLookup(bodyP, "payload") : NULL;
  CorNode* textP    = (bodyP != NULL) ? corTreeLookup(bodyP, "text")    : NULL;

  if ((topicP == NULL) || (topicP->type != CorString) || ((payloadP == NULL) && ((textP == NULL) || (textP->type != CorString))))
  {
    corRest.out.httpStatusCode = 400;
    return true;
  }

  if (ftMqttConn == NULL)
  {
    corRest.out.httpStatusCode = 503;            // no --mqttPort, or not connected (yet)
    return true;
  }

  static char rendered[16384];
  const char* message = rendered;

  if (payloadP != NULL)
  {
    char*    savedName = payloadP->name;
    CorNode* savedNext = payloadP->next;

    payloadP->name = NULL;
    payloadP->next = NULL;
    corJsonFastRender(payloadP, rendered);
    payloadP->name = savedName;
    payloadP->next = savedNext;
  }
  else
    message = textP->value.s;

  int rc = mosquitto_publish(ftMqttConn, NULL, topicP->value.s, strlen(message), message, 1, false);

  corRest.out.httpStatusCode = (rc == MOSQ_ERR_SUCCESS) ? 204 : 500;
  return true;
}



// -----------------------------------------------------------------------------
//
// Service table
//
static CorRestServiceSimplified ftServices[] =
{
  { CorVerbGet,    "/dump",  getDump,         0,                       0 },
  { CorVerbGet,    "/count", getCount,        0,                       0 },
  { CorVerbGet,    "/probeCount", getProbeCount, 0,                    0 },
  { CorVerbGet,    "/mqttReady",  getMqttReady,  0,                    0 },
  { CorVerbGet,    "/die",   getDie,          0,                       0 },
  // Programmable response stubs (mock-reply API).
  { CorVerbPost,   "/mock/reply", postMockReply,   ~(uint64_t)0,       0 },
  { CorVerbPost,   "/bridge/publish", postBridgePublish, ~(uint64_t)0,  0 },
  { CorVerbPost,   "/mqtt/publish", postMqttPublish, ~(uint64_t)0,      0 },
  { CorVerbPost,   "/bridge/serviceReply", postBridgeServiceReply, ~(uint64_t)0, 0 },
  { CorVerbDelete, "/mock/reply", deleteMockReply, 0,                  0 },
  // Catch-all accumulators — every verb lands here and honors --status.
  // supportedParams = ~0ULL: corTestClient mocks any NGSI-LD endpoint and
  // must accept whatever URL params the broker forwards, without 400ing.
  { CorVerbGet,    "/**",    getAccumulate,   ~(uint64_t)0,            0 },
  { CorVerbPost,   "/**",    postAccumulate,  ~(uint64_t)0,            0 },
  { CorVerbDelete, "/**",    postAccumulate,  ~(uint64_t)0,            0 },
  { CorVerbPatch,  "/**",    postAccumulate,  ~(uint64_t)0,            0 },
  { CorVerbPut,    "/**",    postAccumulate,  ~(uint64_t)0,            0 },
};

static int ftServiceCount = sizeof(ftServices) / sizeof(ftServices[0]);



// -----------------------------------------------------------------------------
//
// pemSlurp - read a PEM file into a malloc'd, NUL-terminated string
//
// corRest's HTTPS server options want the key/certificate as in-memory PEM, not
// file paths. Returns NULL (and prints to stderr) on any error.
//
static char* pemSlurp(const char* path)
{
  FILE* f = fopen(path, "rb");
  if (f == NULL)
  {
    fprintf(stderr, "corTestClient: cannot open '%s'\n", path);
    return NULL;
  }

  fseek(f, 0, SEEK_END);
  long size = ftell(f);
  fseek(f, 0, SEEK_SET);

  if ((size <= 0) || (size > 64 * 1024))
  {
    fprintf(stderr, "corTestClient: '%s' has unreasonable size %ld\n", path, size);
    fclose(f);
    return NULL;
  }

  char* buf = (char*) malloc(size + 1);
  size_t n = fread(buf, 1, size, f);
  fclose(f);

  if (n != (size_t) size)
  {
    fprintf(stderr, "corTestClient: short read on '%s'\n", path);
    free(buf);
    return NULL;
  }

  buf[size] = 0;
  return buf;
}



// -----------------------------------------------------------------------------
//
// main
//
int main(int argC, char* argV[])
{
  char* progName = strrchr(argV[0], '/');
  progName = (progName != NULL) ? progName + 1 : argV[0];

  CorArgsStatus ks = corArgsInit(progName, ftArgV, "CORTESTCLIENT");
  if (ks != CorArgsOk)
  {
    fprintf(stderr, "corArgsInit failed\n");
    return 1;
  }

  ks = corArgsParse(argC, argV);
  if (ks != CorArgsOk)
  {
    fprintf(stderr, "corArgsParse failed\n");
    return 1;
  }

  if (corLogInit("corTestClient", "/tmp", false, NULL, ftTraceLevels, corArgsBuiltinVerbose, corArgsBuiltinDebug, false) != 0)
  {
    fprintf(stderr, "corLogInit failed\n");
    return 1;
  }

  signal(SIGINT,  SIG_DFL);
  signal(SIGTERM, SIG_DFL);

  dumpInit();

  corRestSetPrettySpaces(2);

  // corTestClient is a notification RECEIVER — accept geo+json notifications (the
  // broker's § 6.3.4 415 gate otherwise rejects a geo+json POST body).
  corRestAcceptGeoJsonInputSet(true);

  // --httpsKey + --httpsCertificate → serve notifications over TLS
  if ((ftHttpsKey != NULL) && (ftHttpsCert != NULL))
  {
    char* keyPem  = pemSlurp(ftHttpsKey);
    char* certPem = pemSlurp(ftHttpsCert);

    if ((keyPem == NULL) || (certPem == NULL))
      return 1;

    corRestHttpsServerCredentialsSet(keyPem, certPem);
    COR_I("corTestClient serving HTTPS on port %u", ftPort);
  }

  //
  // Bridges before the HTTP server: a transport that is up can deliver at once,
  // and ftBridgeSampleIn only needs the dump, which exists from the start.
  //
  ftBridgesLoad();

  if (corRestInit(ftServices, ftServiceCount, ftPort, 2) != 0)
  {
    fprintf(stderr, "corTestClient: corRestInit failed on port %u\n", ftPort);
    return 1;
  }

  if (ftMqttPort != 0)
  {
    pthread_t tid;
    if (pthread_create(&tid, NULL, mqttListenerThread, NULL) != 0)
    {
      fprintf(stderr, "corTestClient: failed to start MQTT listener thread\n");
      return 1;
    }
    pthread_detach(tid);
    COR_I("corTestClient MQTT listener: localhost:%u topic='%s'", ftMqttPort, ftMqttTopic);
  }

  COR_I("corTestClient running on port %u", ftPort);

  while (1)
    pause();

  return 0;
}