From fdae3eb85ce5d7f27b9474b711ff36ab503877dc Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Tue, 3 Nov 2026 17:30:05 -0600 Subject: [PATCH] feat(csapi-mqtt): serve the AsyncAPI description at /asyncapi/mqtt --- .../sensorhub-service-csapi-mqtt/README.md | 28 +- .../service/csapi/mqtt/AsyncApiHandler.java | 276 +++++++++++++ .../service/csapi/mqtt/CsApiMqttService.java | 7 +- .../csapi/mqtt/CsApiMqttServiceConfig.java | 13 + .../service/csapi/mqtt/asyncapi-mqtt.yaml | 386 ++++++++++++++++++ .../impl/service/csapi/mqtt/asyncapi.html | 65 +++ .../csapi/mqtt/TestAsyncApiHandler.java | 114 ++++++ 7 files changed, 887 insertions(+), 2 deletions(-) create mode 100644 services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/AsyncApiHandler.java create mode 100644 services/sensorhub-service-csapi-mqtt/src/main/resources/org/sensorhub/impl/service/csapi/mqtt/asyncapi-mqtt.yaml create mode 100644 services/sensorhub-service-csapi-mqtt/src/main/resources/org/sensorhub/impl/service/csapi/mqtt/asyncapi.html create mode 100644 services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestAsyncApiHandler.java diff --git a/services/sensorhub-service-csapi-mqtt/README.md b/services/sensorhub-service-csapi-mqtt/README.md index 289219b0a..b714e2c4e 100644 --- a/services/sensorhub-service-csapi-mqtt/README.md +++ b/services/sensorhub-service-csapi-mqtt/README.md @@ -27,6 +27,27 @@ Configuration: - `nodeId`: the first level of every topic. The default is `api`. - Use one topic level only. Do not use `/`, `+`, `#` or NUL. Do not start it with `$`. - Give each CS API server on the same broker a different `nodeId`. +- `brokerAddress`: the broker host and port (MQTT over TCP) in the AsyncAPI description, for example `broker.example.org:1883`. + If it is empty, the service uses the HTTP host and port 1883. Port 1883 is the port of the bundled HiveMQ config. Set this value if the node's HiveMQ config uses another port. +- `webSocketAddress`: the broker address for MQTT over WebSocket in the AsyncAPI description, for example `broker.example.org:8083/mqtt` or `wss://broker.example.org/mqtt`. + If it is empty, the description lists TCP only. +- `asyncApiViewerUrl`: the base URL of the AsyncAPI viewer for the HTML page. The default is `https://unpkg.com/@asyncapi/react-component@3.2.2`. + If it is empty, the HTML page shows only the YAML text. + +## Service description + +The service publishes an AsyncAPI 3.0 description of its topics at: + +``` +{apiRoot}/asyncapi/mqtt +``` + +For example, `http://localhost:8181/sensorhub/api/asyncapi/mqtt`. +The media type is `application/asyncapi+yaml`. +The description shows the topics, formats, delivery settings and known limits of this version. + +- A browser gets an HTML page that shows the document. You can also use `?f=html`. The page loads the AsyncAPI viewer from `asyncApiViewerUrl`. If it cannot load the viewer, it shows the YAML text. +- Other clients get YAML. You can also use `?f=yaml`. ## Topics @@ -125,13 +146,14 @@ Done: - `{nodeId}` topic prefix - Resource Data topics (`:data`) with the full `{systemPath}` - Format subtopics (`:data/{format}`) +- AsyncAPI service description at `{apiRoot}/asyncapi/mqtt` Not done yet: - Resource Event topics (topics without `:data`) and CloudEvents messages - Wildcard subscriptions (`+`, `#`) - MQTT 5 Content Type -- AsyncAPI service description +- A link to the AsyncAPI description from the CS API landing page ## Known limits @@ -146,6 +168,10 @@ Not done yet: - A default format can also be used on its format subtopic. The Part 3 draft says this SHOULD NOT be offered, because data is then sent twice. Here data is sent only to topics with subscribers. - A `swe-xml` publication creates the resource, but the reply says "invalid payload". The CS API reads past the end of the record. HTTP has the same problem. - Video datastreams offer only `om-json` and `swe-binary`. In `om-json`, the result does not include the frame. +- The CS API landing page does not link to the AsyncAPI description yet. Use the path above. +- The AsyncAPI description is YAML and HTML only. There is no JSON version (`f=json`). +- The AsyncAPI description lists WebSocket servers only when `webSocketAddress` is set. +- The service cannot read the HiveMQ settings. The description states the delivery settings (QoS 1, retain false, sessions in memory) of the bundled HiveMQ config. Authentication is not stated. ## Legacy module diff --git a/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/AsyncApiHandler.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/AsyncApiHandler.java new file mode 100644 index 000000000..dae8eeeaf --- /dev/null +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/AsyncApiHandler.java @@ -0,0 +1,276 @@ +/***************************** BEGIN LICENSE BLOCK *************************** + +The contents of this file are subject to the Mozilla Public License, v. 2.0. +If a copy of the MPL was not distributed with this file, You can obtain one +at http://mozilla.org/MPL/2.0/. + +Software distributed under the License is distributed on an "AS IS" basis, +WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License +for the specific language governing rights and limitations under the License. + +Copyright (C) 2026 GeoRobotix Innovative Research. All Rights Reserved. + +******************************* END LICENSE BLOCK ***************************/ + +package org.sensorhub.impl.service.csapi.mqtt; + +import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.function.Supplier; +import java.util.stream.Collectors; +import org.sensorhub.impl.service.consys.InvalidRequestException; +import org.sensorhub.impl.service.consys.ServiceErrors; +import org.sensorhub.impl.service.consys.resource.IResourceHandler; +import org.sensorhub.impl.service.consys.resource.RequestContext; +import org.sensorhub.impl.service.consys.resource.ResourceFormat; + + +/** + *

+ * Serves the AsyncAPI 3.0 description of the MQTT binding at + * {@code {apiRoot}/asyncapi/mqtt}, as a service description of + * OGC API - Publish-Subscribe. The document is a template filled in with the + * node ID, the API root, the broker host and the format names. + *

+ * The response is YAML, or an HTML page that shows the YAML for {@code f=html} + * and for browsers. The HTML page uses the same layout and static files as + * the CS API HTML pages. + *

+ */ +class AsyncApiHandler implements IResourceHandler +{ + static final String NAME = "asyncapi"; + static final String BINDING = "mqtt"; + static final String MEDIA_TYPE = "application/asyncapi+yaml"; + static final String SHORT_YAML = "yaml"; + static final String TEMPLATE = "asyncapi-mqtt.yaml"; + static final String HTML_TEMPLATE = "asyncapi.html"; + static final int DEFAULT_MQTT_PORT = 1883; + static final String[][] MQTT_VERSIONS = {{"5", "MQTT 5"}, {"3.1.1", "MQTT 3.1.1"}}; + + final String template; + final String htmlTemplate; + final String version; + final Supplier config; + final Supplier topics; + + + /** + * @param version the module version + * @param config the module config, read for each request + * @param topics the active topic mapper, or null when the service is stopped + */ + AsyncApiHandler(String version, Supplier config, Supplier topics) + { + this.template = readResource(TEMPLATE); + this.htmlTemplate = readResource(HTML_TEMPLATE); + this.version = version != null ? version : "unknown"; + this.config = config; + this.topics = topics; + } + + + static String readResource(String name) + { + try (InputStream is = AsyncApiHandler.class.getResourceAsStream(name)) + { + return new String(is.readAllBytes(), StandardCharsets.UTF_8); + } + catch (IOException e) + { + throw new UncheckedIOException("Cannot read " + name, e); + } + } + + + @Override + public String[] getNames() + { + return new String[] {NAME}; + } + + + @Override + public void doGet(RequestContext ctx) throws InvalidRequestException, IOException, SecurityException + { + var mapper = topics.get(); + if (ctx.isEndOfPath() || !BINDING.equals(ctx.popNextPathElt()) || !ctx.isEndOfPath() || mapper == null) + throw ServiceErrors.notFound(); + + var html = isHtml(ctx.getParameter("f"), ctx.isBrowserHtmlRequest()); + var doc = render(mapper, ctx.getApiRootURL()); + if (html) + doc = renderHtml(mapper, ctx.getApiRootURL(), doc); + + ctx.setResponseContentType(html ? ResourceFormat.HTML.getMimeType() : MEDIA_TYPE); + ctx.getOutputStream().write(doc.getBytes(StandardCharsets.UTF_8)); + } + + + /** + * @return true for f=html, or for a browser when there is no f parameter; + * false for f=yaml or the AsyncAPI YAML media type + */ + static boolean isHtml(String f, boolean isBrowser) throws InvalidRequestException + { + if (f == null) + return isBrowser; + if (ResourceFormat.SHORT_HTML.equals(f) || ResourceFormat.HTML.getMimeType().equals(f)) + return true; + if (SHORT_YAML.equals(f) || MEDIA_TYPE.equals(f)) + return false; + throw ServiceErrors.unsupportedFormat(f); + } + + + String render(TopicMapper mapper, String apiRoot) + { + var formatList = mapper.formats.entrySet().stream() + .map(e -> "- `" + e.getKey() + "`: `" + e.getValue() + "`") + .collect(Collectors.joining("\n ")); + + return template + .replace("${nodeId}", nodeId(mapper)) + .replace("${apiRoot}", apiRoot) + .replace("${version}", version) + .replace("${servers}", servers(apiRoot)) + .replace("${formatEnum}", String.join(", ", mapper.formats.keySet())) + .replace("${formatList}", formatList); + } + + + /** + * HTML page that shows the document with the AsyncAPI viewer from + * unpkg.com. If the viewer cannot load, the page shows the YAML text + * with the CS API page style. + */ + String renderHtml(TopicMapper mapper, String apiRoot, String doc) + { + return htmlTemplate + .replace("${viewerUrl}", escapeHtml(trim(config.get().asyncApiViewerUrl))) + .replace("${apiRoot}", escapeHtml(apiRoot)) + .replace("${yamlUrl}", escapeHtml(apiRoot + "/" + NAME + "/" + BINDING + "?f=" + SHORT_YAML)) + .replace("${brokerList}", brokerList(apiRoot)) + .replace("${nodeId}", escapeHtml(nodeId(mapper))) + .replace("${doc}", escapeHtml(doc)); + } + + + static String nodeId(TopicMapper mapper) + { + var prefix = mapper.getPrefix(); + return prefix.substring(0, prefix.length() - 1); + } + + + /** The TCP broker address from the config, or the HTTP host and the default MQTT port */ + String brokerHost(String apiRoot) + { + var address = trim(config.get().brokerAddress); + if (!address.isEmpty()) + return address; + var host = URI.create(apiRoot).getHost(); + return (host != null ? host : "localhost") + ":" + DEFAULT_MQTT_PORT; + } + + + /** HTML list of the advertised broker addresses */ + String brokerList(String apiRoot) + { + var list = "TCP at " + escapeHtml(brokerHost(apiRoot)) + ""; + var ws = trim(config.get().webSocketAddress); + if (!ws.isEmpty()) + list += " and WebSocket at " + escapeHtml(ws) + ""; + return list; + } + + + /** Server Objects: TCP, and WebSocket if the config has a WebSocket address */ + String servers(String apiRoot) + { + var sb = new StringBuilder(); + var tcpHost = brokerHost(apiRoot); + for (var v: MQTT_VERSIONS) + appendServer(sb, "mqtt" + v[0].replace(".", ""), tcpHost, null, "mqtt", v, "TCP"); + + var ws = trim(config.get().webSocketAddress); + if (!ws.isEmpty()) + { + var protocol = "ws"; + if (ws.startsWith("wss://")) + protocol = "wss"; + ws = ws.replaceFirst("^wss?://", ""); + var slash = ws.indexOf('/'); + var host = slash < 0 ? ws : ws.substring(0, slash); + var path = slash < 0 ? null : ws.substring(slash); + for (var v: MQTT_VERSIONS) + appendServer(sb, "mqtt" + v[0].replace(".", "") + "-" + protocol, host, path, protocol, v, protocol.equals("wss") ? "secure WebSocket" : "WebSocket"); + } + return sb.toString(); + } + + + static void appendServer(StringBuilder sb, String name, String host, String path, String protocol, String[] mqttVersion, String transport) + { + sb.append(" ").append(name).append(":\n") + .append(" host: '").append(host.replace("'", "''")).append("'\n"); + if (path != null) + sb.append(" pathname: '").append(path.replace("'", "''")).append("'\n"); + sb.append(" protocol: ").append(protocol).append('\n') + .append(" protocolVersion: '").append(mqttVersion[0]).append("'\n") + .append(" description: ").append(mqttVersion[1]).append(" over ").append(transport).append('.'); + if ("5".equals(mqttVersion[0])) + sb.append(" Rejections carry a reason code and reason string."); + sb.append('\n'); + } + + + static String trim(String s) + { + return s != null ? s.trim() : ""; + } + + + static String escapeHtml(String s) + { + return s.replace("&", "&").replace("<", "<").replace(">", ">").replace("\"", """); + } + + + @Override + public void doPost(RequestContext ctx) throws InvalidRequestException + { + throw ServiceErrors.unsupportedOperation("Read-only resource"); + } + + + @Override + public void doPut(RequestContext ctx) throws InvalidRequestException + { + throw ServiceErrors.unsupportedOperation("Read-only resource"); + } + + + @Override + public void doDelete(RequestContext ctx) throws InvalidRequestException + { + throw ServiceErrors.unsupportedOperation("Read-only resource"); + } + + + @Override + public void addSubResource(IResourceHandler resource) + { + throw new UnsupportedOperationException(); + } + + + @Override + public void addSubResource(IResourceHandler resource, String... names) + { + throw new UnsupportedOperationException(); + } +} diff --git a/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttService.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttService.java index 57b494351..f104b744b 100644 --- a/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttService.java +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttService.java @@ -33,7 +33,7 @@ */ public class CsApiMqttService extends AbstractMqttServiceModule implements IServiceModule { - protected CsApiMqttConnector mqttConnector; + protected volatile CsApiMqttConnector mqttConnector; @Override @@ -71,6 +71,11 @@ protected void doStart() throws SensorHubException var topics = new TopicMapper(config.nodeId, formats); mqttConnector = new CsApiMqttConnector(servlet, topics); mqttServer.registerHandler(topics.getPrefix(), mqttConnector); + + // AsyncAPI service description at {apiRoot}/asyncapi/mqtt + var version = new CsApiMqttServiceDescriptor().getModuleVersion(); + servlet.getRootHandler().addSubResource(new AsyncApiHandler(version, () -> config, + () -> mqttConnector != null ? mqttConnector.topics : null)); getLogger().info("CS API MQTT handler registered on '{}', format subtopics {}", topics.getPrefix(), topics.formats.keySet()); setState(ModuleState.STARTED); } diff --git a/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceConfig.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceConfig.java index c0b56d120..a34da7cd9 100644 --- a/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceConfig.java +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceConfig.java @@ -33,4 +33,17 @@ public class CsApiMqttServiceConfig extends ServiceConfig + "It must be one topic level: no '/', '+', '#' or NUL, and it must not start with '$'. " + "Use a different value for each CS API server that shares the MQTT broker.") public String nodeId = "api"; + + @DisplayInfo(label="Broker Address", desc="Broker host and port (MQTT over TCP) to advertise in the AsyncAPI description, " + + "as in 'broker.example.org:1883'. Leave empty to use the HTTP host and port 1883, the port of the bundled " + + "HiveMQ config. Set it if the node's HiveMQ config uses another port.") + public String brokerAddress; + + @DisplayInfo(label="WebSocket Address", desc="Broker address for MQTT over WebSocket to advertise in the AsyncAPI description, " + + "as in 'broker.example.org:8083/mqtt' or 'wss://broker.example.org/mqtt'. Leave empty to advertise TCP only.") + public String webSocketAddress; + + @DisplayInfo(label="AsyncAPI Viewer URL", desc="Base URL of the AsyncAPI viewer (@asyncapi/react-component) " + + "used by the HTML page of the AsyncAPI description. Leave empty to show only the YAML text.") + public String asyncApiViewerUrl = "https://unpkg.com/@asyncapi/react-component@3.2.2"; } diff --git a/services/sensorhub-service-csapi-mqtt/src/main/resources/org/sensorhub/impl/service/csapi/mqtt/asyncapi-mqtt.yaml b/services/sensorhub-service-csapi-mqtt/src/main/resources/org/sensorhub/impl/service/csapi/mqtt/asyncapi-mqtt.yaml new file mode 100644 index 000000000..ae3b1713a --- /dev/null +++ b/services/sensorhub-service-csapi-mqtt/src/main/resources/org/sensorhub/impl/service/csapi/mqtt/asyncapi-mqtt.yaml @@ -0,0 +1,386 @@ +asyncapi: 3.0.0 +id: urn:osh:csapi-mqtt:${nodeId} +info: + title: OSH Connected Systems API - MQTT binding + version: ${version} + description: |- + MQTT binding of the OSH Connected Systems (CS) API, module + `sensorhub-service-csapi-mqtt`. It follows the MQTT binding of + OGC API - Connected Systems Part 3 (draft). The work is not complete. + The legacy topics, under the CS API HTTP endpoint, are in the module + `sensorhub-service-consys-mqtt`. + + CS API HTTP server: ${apiRoot} + + **What it does** + + - Subscribe to a Resource Data topic to get live data. + - Publish to a Resource Data topic to create one resource. MQTT cannot + update or delete resources. + - The CS API checks each request. The same users and permissions apply + as for HTTP. + + **Topics** + + - Every topic starts with `${nodeId}/`. + - Data topics end with `:data` on the last collection name. + - `:data/{format}` selects a format. Plain `:data` uses the default format. + - This document lists topics for resources of a top-level system. For a + subsystem, insert `/subsystems/{id}` after `systems/{id}`, once for + each level, for example + `${nodeId}/systems/{id}/subsystems/{id}/datastreams/{id}/observations:data`. + + **Format names** + + ${formatList} + + A format that the resource does not support is rejected. There is no + fallback to the default format. + + **Not done yet** + + - Resource Events (topics without `:data`) and CloudEvents messages. + These topics are rejected. + - Wildcard subscriptions (`+`, `#`). These topics are rejected. + - MQTT 5 Content Type. The server does not set it. Clients must use + the topic to know the format. + - Command result topics (`.../commands/{id}/result:data`). + + **Delivery** + + - The server publishes with QoS 1 and retain false. + - The broker keeps sessions in memory. Sessions do not survive a restart. + - There are no ordering or replay guarantees beyond MQTT. QoS does not + guarantee exactly-once creation of resources or commands. + - Data goes only to topics with subscribers. There is no retained state. + - The broker uses its default maximum message size. + + **Errors** + + - The embedded OSH broker rejects a bad subscription in SUBACK and a bad + publication in PUBACK, with an MQTT 5 reason code and reason string. + - MQTT 3.1.1 clients get no reason. QoS 0 publications get no reply. + - There is no asynchronous error topic yet. + + **Known limits** + + - The topic shape is checked. That each ID belongs to the resource before + it is not checked by the CS API. + - The status topic of one command carries the status of all commands on + its control stream. + - `swe-xml` publications create the resource but are reported as invalid. + contact: + name: OpenSensorHub + url: https://github.com/opensensorhub/osh-addons + license: + name: Mozilla Public License 2.0 + url: https://mozilla.org/MPL/2.0/ + externalDocs: + description: OGC API - Connected Systems Part 3 working draft + url: https://github.com/opengeospatial/ogcapi-connected-systems/tree/master/api/part3 + +defaultContentType: application/json + +servers: +${servers} +channels: + observations: + address: '${nodeId}/systems/{systemId}/datastreams/{dataStreamId}/observations:data' + title: Observations of a datastream (default format) + messages: + observation: + $ref: '#/components/messages/observation' + parameters: + systemId: + $ref: '#/components/parameters/systemId' + dataStreamId: + $ref: '#/components/parameters/dataStreamId' + x-ogc-link: + rel: collection + type: application/json + href: '${apiRoot}/datastreams' + observationsFormat: + address: '${nodeId}/systems/{systemId}/datastreams/{dataStreamId}/observations:data/{format}' + title: Observations of a datastream (selected format) + description: The datastream lists its formats in `formats` over HTTP. Video + datastreams offer only `om-json` and `swe-binary`. + messages: + observation: + $ref: '#/components/messages/formatted' + parameters: + systemId: + $ref: '#/components/parameters/systemId' + dataStreamId: + $ref: '#/components/parameters/dataStreamId' + format: + $ref: '#/components/parameters/format' + commands: + address: '${nodeId}/systems/{systemId}/controlstreams/{controlStreamId}/commands:data' + title: Commands of a control stream (default format) + messages: + command: + $ref: '#/components/messages/command' + parameters: + systemId: + $ref: '#/components/parameters/systemId' + controlStreamId: + $ref: '#/components/parameters/controlStreamId' + x-ogc-link: + rel: collection + type: application/json + href: '${apiRoot}/controlstreams' + commandsFormat: + address: '${nodeId}/systems/{systemId}/controlstreams/{controlStreamId}/commands:data/{format}' + title: Commands of a control stream (selected format) + messages: + command: + $ref: '#/components/messages/formatted' + parameters: + systemId: + $ref: '#/components/parameters/systemId' + controlStreamId: + $ref: '#/components/parameters/controlStreamId' + format: + $ref: '#/components/parameters/format' + commandStatus: + address: '${nodeId}/systems/{systemId}/controlstreams/{controlStreamId}/commands/{commandId}/status:data' + title: Status of a command + description: Known limit - this topic carries the status of all commands + on the control stream, not only command `{commandId}`. + messages: + status: + $ref: '#/components/messages/commandStatus' + parameters: + systemId: + $ref: '#/components/parameters/systemId' + controlStreamId: + $ref: '#/components/parameters/controlStreamId' + commandId: + $ref: '#/components/parameters/commandId' + systemEvents: + address: '${nodeId}/systems/{systemId}/events:data' + title: System events of a system + messages: + event: + $ref: '#/components/messages/systemEvent' + parameters: + systemId: + $ref: '#/components/parameters/systemId' + systems: + address: '${nodeId}/systems:data' + title: Top-level systems + messages: + system: + $ref: '#/components/messages/system' + x-ogc-link: + rel: collection + type: application/json + href: '${apiRoot}/systems' + subsystems: + address: '${nodeId}/systems/{systemId}/subsystems:data' + title: Subsystems of a system + messages: + system: + $ref: '#/components/messages/system' + parameters: + systemId: + $ref: '#/components/parameters/systemId' + dataStreams: + address: '${nodeId}/systems/{systemId}/datastreams:data' + title: Datastreams of a system + messages: + dataStream: + $ref: '#/components/messages/dataStream' + parameters: + systemId: + $ref: '#/components/parameters/systemId' + controlStreams: + address: '${nodeId}/systems/{systemId}/controlstreams:data' + title: Control streams of a system + messages: + controlStream: + $ref: '#/components/messages/controlStream' + parameters: + systemId: + $ref: '#/components/parameters/systemId' + +# Operations are seen from the server: 'send' means the server publishes and +# clients subscribe; 'receive' means clients publish and the server creates. +operations: + sendObservations: + action: send + channel: + $ref: '#/channels/observations' + summary: Stream new observations of a datastream. + bindings: + mqtt: {qos: 1, retain: false, bindingVersion: 0.2.0} + receiveObservations: + action: receive + channel: + $ref: '#/channels/observations' + summary: Create one observation per publication. + bindings: + mqtt: {qos: 1, bindingVersion: 0.2.0} + sendObservationsFormat: + action: send + channel: + $ref: '#/channels/observationsFormat' + summary: Stream new observations in the selected format. + bindings: + mqtt: {qos: 1, retain: false, bindingVersion: 0.2.0} + receiveObservationsFormat: + action: receive + channel: + $ref: '#/channels/observationsFormat' + summary: Create one observation per publication, in the selected format. + description: Tested with `om-json`, `swe-json` and `swe-csv`. + bindings: + mqtt: {qos: 1, bindingVersion: 0.2.0} + receiveCommands: + action: receive + channel: + $ref: '#/channels/commands' + summary: Send one command per publication to the system. + description: Put the command fields in `parameters`. Give all fields of the + control stream schema. + bindings: + mqtt: {qos: 1, bindingVersion: 0.2.0} + sendCommands: + action: send + channel: + $ref: '#/channels/commands' + summary: Deliver commands to a client that executes them. + description: A subscriber becomes the command executor of the control + stream. The subscription is rejected if a driver or another client + already executes the commands. + bindings: + mqtt: {qos: 1, retain: false, bindingVersion: 0.2.0} + receiveCommandsFormat: + action: receive + channel: + $ref: '#/channels/commandsFormat' + summary: Send one command per publication, in the selected format. + description: Tested with `json`, `swe-json` and `swe-csv`. + bindings: + mqtt: {qos: 1, bindingVersion: 0.2.0} + sendCommandStatus: + action: send + channel: + $ref: '#/channels/commandStatus' + summary: Stream command status reports. + bindings: + mqtt: {qos: 1, retain: false, bindingVersion: 0.2.0} + sendSystemEvents: + action: send + channel: + $ref: '#/channels/systemEvents' + summary: Stream new system events. + description: The subscription is accepted. Delivery is not tested yet. + bindings: + mqtt: {qos: 1, retain: false, bindingVersion: 0.2.0} + receiveSystems: + action: receive + channel: + $ref: '#/channels/systems' + summary: Create a top-level system. Subscriptions are rejected. + bindings: + mqtt: {qos: 1, bindingVersion: 0.2.0} + receiveSubsystems: + action: receive + channel: + $ref: '#/channels/subsystems' + summary: Create a subsystem. Subscriptions are rejected. + bindings: + mqtt: {qos: 1, bindingVersion: 0.2.0} + receiveDataStreams: + action: receive + channel: + $ref: '#/channels/dataStreams' + summary: Create a datastream. Subscriptions are rejected. + bindings: + mqtt: {qos: 1, bindingVersion: 0.2.0} + receiveControlStreams: + action: receive + channel: + $ref: '#/channels/controlStreams' + summary: Create a control stream. Subscriptions are rejected. + bindings: + mqtt: {qos: 1, bindingVersion: 0.2.0} + +components: + parameters: + systemId: + description: Local ID of a system in the CS API. + dataStreamId: + description: Local ID of a datastream in the CS API. + controlStreamId: + description: Local ID of a control stream in the CS API. + commandId: + description: Local ID of a command in the CS API. + format: + description: Format name, made from the media type (remove parameters and + `application/`, change `+` to `-`, use lowercase). + enum: [${formatEnum}] + messages: + observation: + name: observation + title: Observation + contentType: application/json + summary: One observation in the default JSON format. + payload: + type: object + externalDocs: + description: The result schema of each datastream + url: '${apiRoot}/datastreams' + formatted: + name: formatted + title: Resource in the selected format + summary: One resource in the format of the `{format}` topic level. The + payload is text or binary, as the format sets. + payload: {} + command: + name: command + title: Command + contentType: application/json + summary: One command. The fields go in `parameters`. + payload: + type: object + required: [parameters] + properties: + parameters: + type: object + commandStatus: + name: commandStatus + title: Command status + contentType: application/json + payload: + type: object + properties: + command@id: {type: string} + statusCode: {type: string} + reportTime: {type: string, format: date-time} + systemEvent: + name: systemEvent + title: System event + contentType: application/json + payload: + type: object + system: + name: system + title: System + contentType: application/geo+json + summary: A system in GeoJSON. Use the `sml-json` format subtopic for SensorML. + payload: + type: object + dataStream: + name: dataStream + title: Datastream + contentType: application/json + payload: + type: object + controlStream: + name: controlStream + title: Control stream + contentType: application/json + payload: + type: object diff --git a/services/sensorhub-service-csapi-mqtt/src/main/resources/org/sensorhub/impl/service/csapi/mqtt/asyncapi.html b/services/sensorhub-service-csapi-mqtt/src/main/resources/org/sensorhub/impl/service/csapi/mqtt/asyncapi.html new file mode 100644 index 000000000..a5ba25b7e --- /dev/null +++ b/services/sensorhub-service-csapi-mqtt/src/main/resources/org/sensorhub/impl/service/csapi/mqtt/asyncapi.html @@ -0,0 +1,65 @@ + + + + + + AsyncAPI - MQTT binding + + + +
+ +
+ + +
+ + +
+

AsyncAPI - MQTT binding

+

AsyncAPI 3.0 description of the MQTT binding of this API. Clients and tools use the + YAML document (application/asyncapi+yaml).

+
+
Broker
+

MQTT 5 and MQTT 3.1.1 over ${brokerList}. Topics start with ${nodeId}/.

+
+
+
Document
+ 
${doc}
+
+
+ + + + diff --git a/services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestAsyncApiHandler.java b/services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestAsyncApiHandler.java new file mode 100644 index 000000000..3e362d410 --- /dev/null +++ b/services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestAsyncApiHandler.java @@ -0,0 +1,114 @@ +/***************************** BEGIN LICENSE BLOCK *************************** + +The contents of this file are subject to the Mozilla Public License, v. 2.0. +If a copy of the MPL was not distributed with this file, You can obtain one +at http://mozilla.org/MPL/2.0/. + +Software distributed under the License is distributed on an "AS IS" basis, +WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License +for the specific language governing rights and limitations under the License. + +Copyright (C) 2026 GeoRobotix Innovative Research. All Rights Reserved. + +******************************* END LICENSE BLOCK ***************************/ + +package org.sensorhub.impl.service.csapi.mqtt; + +import static org.junit.Assert.*; +import org.junit.Test; + + +public class TestAsyncApiHandler +{ + static final String API_ROOT = "http://host.example:8181/sensorhub/api"; + final TopicMapper mapper = new TopicMapper("node1", TopicMapper.STANDARD_FORMATS); + final CsApiMqttServiceConfig config = new CsApiMqttServiceConfig(); + final AsyncApiHandler handler = new AsyncApiHandler("0.1.0", () -> config, () -> mapper); + + + @Test + public void testRender() + { + var doc = handler.render(mapper, API_ROOT); + + assertFalse("unfilled placeholder", doc.contains("${")); + assertTrue(doc.contains("version: 0.1.0")); + assertTrue(doc.contains("address: 'node1/systems/{systemId}/datastreams/{dataStreamId}/observations:data/{format}'")); + assertTrue(doc.contains("href: 'http://host.example:8181/sensorhub/api/datastreams'")); + assertTrue(doc.contains("enum: [geo-json, json, om-json, sml-json, sml-xml, swe-binary, swe-csv, swe-json, swe-xml]")); + assertTrue(doc.contains("- `swe-csv`: `application/swe+csv`")); + } + + + @Test + public void testTcpServers() + { + var doc = handler.render(mapper, API_ROOT); + assertTrue(doc.contains(" mqtt5:\n host: 'host.example:1883'\n protocol: mqtt\n protocolVersion: '5'")); + assertTrue(doc.contains(" mqtt311:\n host: 'host.example:1883'\n protocol: mqtt\n protocolVersion: '3.1.1'")); + assertFalse(doc.contains("protocol: ws")); + + // config is read for each request + config.brokerAddress = " broker.example:2883 "; + doc = handler.render(mapper, API_ROOT); + assertEquals(2, count(doc, "host: 'broker.example:2883'")); + } + + + @Test + public void testWebSocketServers() + { + config.webSocketAddress = "broker.example:8083/mqtt"; + var doc = handler.render(mapper, API_ROOT); + assertTrue(doc.contains(" mqtt5-ws:\n host: 'broker.example:8083'\n pathname: '/mqtt'\n protocol: ws\n protocolVersion: '5'")); + assertTrue(doc.contains(" mqtt311-ws:\n host: 'broker.example:8083'\n pathname: '/mqtt'\n protocol: ws\n protocolVersion: '3.1.1'")); + + config.webSocketAddress = "wss://broker.example"; + doc = handler.render(mapper, API_ROOT); + assertEquals(2, count(doc, "protocol: wss")); + assertFalse(doc.contains("pathname")); + } + + + @Test + public void testRenderHtml() + { + var html = handler.renderHtml(mapper, API_ROOT, handler.render(mapper, API_ROOT)); + + assertFalse("unfilled placeholder", html.contains("${")); + assertTrue(html.contains("data-viewer-url=\"" + config.asyncApiViewerUrl + "\"")); + // fallback content + assertTrue(html.contains("href=\"" + API_ROOT + "/static/css/bootstrap.min.css\"")); + assertTrue(html.contains("href=\"" + API_ROOT + "/asyncapi/mqtt?f=yaml\"")); + assertTrue(html.contains("over TCP at host.example:1883.")); + assertTrue(html.contains("asyncapi: 3.0.0")); + assertEquals("a <b> & "c"", AsyncApiHandler.escapeHtml("a & \"c\"")); + + // WebSocket address, no viewer + config.webSocketAddress = "broker.example:8083/mqtt"; + config.asyncApiViewerUrl = null; + html = handler.renderHtml(mapper, API_ROOT, handler.render(mapper, API_ROOT)); + assertTrue(html.contains("data-viewer-url=\"\"")); + assertFalse(html.contains("unpkg")); + assertTrue(html.contains("and WebSocket at broker.example:8083/mqtt.")); + } + + + @Test + public void testFormatSelection() throws Exception + { + assertTrue(AsyncApiHandler.isHtml(null, true)); + assertFalse(AsyncApiHandler.isHtml(null, false)); + assertTrue(AsyncApiHandler.isHtml("html", false)); + assertTrue(AsyncApiHandler.isHtml("text/html", false)); + assertFalse(AsyncApiHandler.isHtml("yaml", true)); + assertFalse(AsyncApiHandler.isHtml("application/asyncapi+yaml", true)); + assertThrows(Exception.class, () -> AsyncApiHandler.isHtml("json", false)); + } + + + static int count(String s, String part) + { + return s.split(java.util.regex.Pattern.quote(part), -1).length - 1; + } +}