csapi-mqtt: AsyncAPI service description

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
 

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;
+
+
+/**
+ * <p>
+ * 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.
+ * </p><p>
+ * 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.
+ * </p>
+ */
+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<CsApiMqttServiceConfig> config;
+    final Supplier<TopicMapper> 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<CsApiMqttServiceConfig> config, Supplier<TopicMapper> 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 <code>" + escapeHtml(brokerHost(apiRoot)) + "</code>";
+        var ws = trim(config.get().webSocketAddress);
+        if (!ws.isEmpty())
+            list += " and WebSocket at <code>" + escapeHtml(ws) + "</code>";
+        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("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;").replace("\"", "&quot;");
+    }
+
+
+    @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();
+    }
+}

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<CsApiMqttServiceConfig> implements IServiceModule<CsApiMqttServiceConfig>
 {
-    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);
                 }

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