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;
+ }
+}