diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 00b5566..49bdbcd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -71,6 +71,12 @@ jobs: - name: Functional tests working-directory: stack/bridge + # + # corTest sorts JSON bodies with the corJson tool and refuses to run without it: built with corJson, + # not on PATH here (as in coraine's ci.yml) + # + env: + CORJSON: ${{ github.workspace }}/stack/corJson/bin/corJson run: make test CORTEST=../corLibs/bin/corTest - name: Keep what a failing test left diff --git a/README.md b/README.md index 4bd5b4a..f7d3725 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ # corBridgeTemplate + A **bridge plugin** for the coraine NGSI-LD broker: `tmpl.so`, a working bridge with a tiny example transport, to be turned into a real one - OPC UA, CKAN, Philips Hue, anything the broker does not speak. It builds, loads into a broker, and carries a value each way before a line of its own is @@ -8,12 +10,15 @@ written. ## Starting a bridge from this template - cp -r corBridgeTemplate corOpcuaBridge && cd corOpcuaBridge # or a fresh clone / export + mkdir corOpcuaBridge && git -C corBridgeTemplate archive HEAD | tar -x -C corOpcuaBridge + cd corOpcuaBridge && git init -b main ./rename.sh opcua # tmpl -> opcua everywhere: opcua.so, src/opcuaRegister.c, ... make && make test # builds, and a broker moves a value both ways through it - grep -rn TODO makefile src test # where the protocol goes + grep -rn --exclude-dir=.git TODO . # where the protocol goes - the README and the CI too `rename.sh` takes the bridge's alias: lowercase letters and digits. It removes this section and itself. +An export, not a copy of the checkout: `cp -r` takes the template's `.git` along - its history, and an +`origin` that pushes to corBridgeTemplate (`rename.sh` warns about that). ## What a bridge is @@ -30,6 +35,12 @@ A bridge connects the broker to a foreign transport. It **knows nothing about en - **Channel** - *this endpoint <-> this entity attribute, this direction*. Channels live in the broker, which configures them from the `--bridgeConfig` file and shows them on `GET /ngsi-ld/v1/channels`. A Channel the bridge refuses stays, `dormant`, with the reason. +- **DataChannel** - for a transport whose messages are **records** (a CKAN row, a CSV line: one + JSON object, one member per column) naming their entity in a column: *this endpoint -> the entities + each record makes*, inbound only - the `dataChannels` section of the file, `GET + /ngsi-ld/v1/dataChannels` (coraine `COR_FEATURE_DATA_CHANNELS`). Not a Channel; the plugin's side is + the same `sampleIn`, one record a call, its time 0 when the record has none (no `observedAt`); + corCkanBridge is the example. - **Bridge** - one transport instance (one connection, one server), `GET /ngsi-ld/v1/bridges`. `--bridges tmpl` loads the plugin; it is never loaded at runtime or inferred from configuration. - **In** - a value arrives on an endpoint; the plugin calls `brokerP->sampleIn(alias, endpoint, json, @@ -50,6 +61,7 @@ offers). In the order the broker uses them: | `init(configFile, brokerP)` | broker, late in startup | keeps `brokerP`; reads its member of the `--bridgeConfig` file; brings the transport up | | `channelAddInfo(endpoint, kind, direction, info)` | broker | per Channel, after `init`: checks the endpoint and `channelInfo`, subscribes | | `brokerP->sampleIn(...)` | the plugin's own | a value arrived | +| `brokerP->channelStatusIn(...)` | the plugin's own, or inside `channelAddInfo` | ABI 12: a Channel carries, or not and why - shown as its `status` and `statusReason` | | `publish(endpoint, json)` | broker, inside the request | hands the value to the transport and returns - must not block | | `channelDel(endpoint)` | broker | stops carrying an endpoint (not called yet: Channels come from the file only) | | `close()` | broker, at shutdown | returns only when no plugin thread can call `sampleIn` any more | @@ -58,7 +70,11 @@ offers). In the order the broker uses them: - **NULL means "not supported".** A slot the plugin leaves NULL is answered `BRIDGE_UNSUPPORTED` by the broker. Optional slots: `args` (command-line options of the plugin's own), services and actions (ABI 2-4), `notifySchemes` + `notify` (notifications to a URI scheme - corMqttBridge), Service - Execution (ABI 11). + Execution (ABI 11) - and `publish`, for a bridge that only reads (its Channels inbound). +- **Why a Channel is dormant** is the plugin's to say: through `channelStatusIn` (ABI 12) before + `channelAddInfo` refuses it, and at any time after - a peer that does not have the endpoint, a peer + that cannot be reached. Without it, `GET /ngsi-ld/v1/channels` shows only the broker's "its address + or channelInfo does not fit the transport", and anything learned later shows nowhere but the log. - **The ABI handshake.** The broker zeroes the driver, writes *its* `BRIDGE_ABI_VERSION` into `abiVersion` and calls `bridgeRegister`. The struct is the broker's, at the size of its header: the plugin fills in only the slots that version has (`channelAddInfo` from ABI 9), then writes its own @@ -100,6 +116,9 @@ broker's - it makes the Channels; everything else in the member is the plugin's. - A Channel's direction is `both` and every configured attribute exists from startup, with the value `"uninitialized"` until the first sample. +- The **catch-all entity** is on unless the section says `"defaultEntity": false`: a sample on an + endpoint no Channel claims is stored on `urn:ngsi-ld:tmpl:default`. A transport that hands over only + the endpoints it was given never reaches it; one that hands over everything it hears does. - `channelInfo` - how to read or carry one endpoint - is checked by the broker for its shape (an array of `{"key", "value"}` strings) and never for its keys: an unknown key is the plugin's to refuse (`BRIDGE_BAD_INPUT`). @@ -126,7 +145,9 @@ a broker built with `coraine-build `: `make COR_LIBS=`, and Starts `coraine` (from `PATH`: the coraine package, or `. /install/env` after `coraine-build`) on port 19026 with ramDB - no database server - loads `./tmpl.so` by path, and runs -`test/funcTests/cases/*.test` with corTest (`PATH`, or `CORTEST=`). `TMPL_BROKER`, `TMPL_PORT` and +`test/funcTests/cases/*.test` with corTest (`PATH`, or `CORTEST=`), with `COR_TEST_NO_REAP=on`: +corTest's start-up reaping is for coraine's own suite - it kills every `coraine` process on the host - +and these tests stop their broker by its pid. `TMPL_BROKER`, `TMPL_PORT` and `TMPL_TEST_DIR` override; the broker's log is `/tmp/tmpl-bridge-test/broker.log`. ## Running diff --git a/makefile b/makefile index 2b3177f..3735d9a 100644 --- a/makefile +++ b/makefile @@ -114,8 +114,12 @@ contract: | $(COR_LIBS)/corBridge/BridgeDriver.h # CORTEST ?= $(shell command -v corTest || echo $(COR_LIBS)/corTest/corTest) +# COR_TEST_NO_REAP=on: corTest's start-up reaping (killall -9 coraine, and coraine's mock clients) is +# for coraine's own suite - here it would take down every broker on the host. These tests stop their +# own broker by pid. +# test: all - $(CORTEST) --noColor + COR_TEST_NO_REAP=on $(CORTEST) --noColor clean: rm -f src/*.o src/*.d *.so *~ .flags diff --git a/rename.sh b/rename.sh index 7c92530..1510a36 100755 --- a/rename.sh +++ b/rename.sh @@ -32,6 +32,15 @@ self=$(basename "$0") [ -f src/tmplRegister.c ] || { echo "$0: no src/tmplRegister.c - renamed already?" >&2; exit 1; } +# +# A copy of the template's checkout carries its .git: the template's history, and an origin that pushes +# to corBridgeTemplate +# +if origin=$(git config --get remote.origin.url 2>/dev/null) && [[ "$origin" == *corBridgeTemplate* ]]; then + echo "$0: warning: this checkout's origin is the template ($origin) - a push goes there." >&2 + echo " start from an export instead: git -C corBridgeTemplate archive HEAD | tar -x -C ; git init" >&2 +fi + Name="$(tr '[:lower:]' '[:upper:]' <<< "${name:0:1}")${name:1}" NAME=$(tr '[:lower:]' '[:upper:]' <<< "$name") repo="cor${Name}Bridge" diff --git a/src/tmplRegister.c b/src/tmplRegister.c index ad40fac..dc389a7 100644 --- a/src/tmplRegister.c +++ b/src/tmplRegister.c @@ -241,6 +241,29 @@ static bool channelInfoText(const char* info, bool* textP, const char** whyP) +// ----------------------------------------------------------------------------- +// +// channelRefused - say on the Channel why it is not carried: GET /ngsi-ld/v1/channels shows it +// +// channelStatusIn (ABI 12) - from inside channelAddInfo, before it refuses, or at any time later from a +// thread of the plugin's own: a peer that turns out not to have the endpoint, one that cannot be +// reached (BridgeChannelDormant), one that is back (BridgeChannelAvailable). Built against corBridge +// headers from before ABI 12, or loaded by a broker without it, the reason is in the log only and the +// Channel shows the broker's own words. +// +static void channelRefused(const char* endpoint, const char* why) +{ +#if BRIDGE_ABI_VERSION >= 12 + if ((brokerP != NULL) && (brokerP->abiVersion >= 12) && (brokerP->channelStatusIn != NULL)) + brokerP->channelStatusIn("tmpl", endpoint, BridgeChannelDormant, why); +#else + (void) endpoint; + (void) why; +#endif +} + + + // ----------------------------------------------------------------------------- // // tmplChannelAddInfo - start carrying an endpoint, ABI 9 @@ -258,16 +281,26 @@ static int tmplChannelAddInfo(const char* endpoint, BridgeChannelKind kind, Brid if (kind != BridgeChannelTopic) { COR_E("tmpl: '%s': this bridge carries topics only", endpoint); + channelRefused(endpoint, "this bridge carries topics only"); return BRIDGE_UNSUPPORTED; } if (channelInfoText(info, &text, &why) == false) { - COR_E("tmpl: '%s': %s", endpoint, (why != NULL) ? why : "channelInfo is not an array of {key, value}"); + if (why == NULL) + why = "channelInfo is not an array of {key, value}"; + + COR_E("tmpl: '%s': %s", endpoint, why); + channelRefused(endpoint, why); return BRIDGE_BAD_INPUT; } - return tmplTransportEndpointAdd(endpoint, direction, text); + int r = tmplTransportEndpointAdd(endpoint, direction, text, &why); + + if (r != BRIDGE_OK) + channelRefused(endpoint, (why != NULL) ? why : "the transport would not carry it"); + + return r; } diff --git a/src/tmplTransport.c b/src/tmplTransport.c index 73a4a63..9bd95c3 100644 --- a/src/tmplTransport.c +++ b/src/tmplTransport.c @@ -314,20 +314,22 @@ void tmplTransportClose(void) // An endpoint is a relative path of letters, digits, '_', '-', '.' and '/', with no '..' - a Channel // cannot reach out of the directory. // -int tmplTransportEndpointAdd(const char* endpoint, BridgeDirection direction, bool text) +int tmplTransportEndpointAdd(const char* endpoint, BridgeDirection direction, bool text, const char** whyP) { const char* allowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-./"; if ((endpoint[0] == 0) || (endpoint[0] == '/') || (endpoint[strspn(endpoint, allowed)] != 0) || (strstr(endpoint, "..") != NULL)) { - COR_E("tmpl: '%s' is not an endpoint - a relative file path, no '..'", endpoint); + *whyP = "not an endpoint of this transport - a relative file path, no '..'"; + COR_E("tmpl: '%s': %s", endpoint, *whyP); return BRIDGE_BAD_INPUT; } char path[PATH_MAX]; if (snprintf(path, sizeof(path), "%s/%s", dir, endpoint) >= (int) sizeof(path)) { - COR_E("tmpl: '%s': the path is too long", endpoint); + *whyP = "the path is too long"; + COR_E("tmpl: '%s': %s", endpoint, *whyP); return BRIDGE_BAD_INPUT; } @@ -351,7 +353,8 @@ int tmplTransportEndpointAdd(const char* endpoint, BridgeDirection direction, bo if (freeP == NULL) { pthread_mutex_unlock(&mtx); - COR_E("tmpl: '%s': no room for another endpoint", endpoint); + *whyP = "no room for another endpoint"; + COR_E("tmpl: '%s': %s", endpoint, *whyP); return BRIDGE_ERR; } diff --git a/src/tmplTransport.h b/src/tmplTransport.h index 403f36d..1fd4698 100644 --- a/src/tmplTransport.h +++ b/src/tmplTransport.h @@ -78,10 +78,11 @@ extern void tmplTransportClose(void); // tmplTransportEndpointAdd - start carrying an endpoint // // @param text the endpoint carries plain text, not JSON: an arriving value becomes a JSON string +// @param whyP on a refusal, why - one line for a person, static text // // @return BRIDGE_OK, BRIDGE_BAD_INPUT (not an endpoint of this transport) or BRIDGE_ERR // -extern int tmplTransportEndpointAdd(const char* endpoint, BridgeDirection direction, bool text); +extern int tmplTransportEndpointAdd(const char* endpoint, BridgeDirection direction, bool text, const char** whyP); diff --git a/test/funcTests/cases/bridge_tmpl.test b/test/funcTests/cases/bridge_tmpl.test index da3e623..3fdc310 100644 --- a/test/funcTests/cases/bridge_tmpl.test +++ b/test/funcTests/cases/bridge_tmpl.test @@ -12,7 +12,7 @@ # echo the bridge's own write does not come back as a sample - samplesIn stays 0 # # An endpoint the transport cannot carry ('../escape') is refused: the Channel stays, dormant, with -# the reason. The device is a directory of files (see src/tmplTransport.c). +# the transport's reason (channelStatusIn, bridge ABI 12). The device is a directory of files (see src/tmplTransport.c). # # TODO: in a real bridge the device is the protocol's peer - a test server the INIT section starts # (corMqttBridge's tests start mosquitto) - and deviceWrite / deviceAwait talk to it. @@ -107,7 +107,7 @@ rm -rf "${TMPL_TEST_DIR:?}/device" "$TMPL_TEST_DIR/bridges.json" # broker. 04. GET /channels - '../escape' dormant and why; setpoint sent once and not taken back in ========================================================================================== -../escape dormant samplesIn=0 samplesOut=0 - the bridge would not carry it: its address or channelInfo does not fit the transport +../escape dormant samplesIn=0 samplesOut=0 - not an endpoint of this transport - a relative file path, no '..' tank1/level available samplesIn=1 samplesOut=0 tank1/setpoint available samplesIn=0 samplesOut=1 tank1/state available samplesIn=1 samplesOut=0