+{
+ protected CsApiMqttConnector mqttConnector;
+
+
+ @Override
+ protected void doInit() throws SensorHubException
+ {
+ super.doInit();
+
+ try
+ {
+ TopicMapper.checkNodeId(config.nodeId);
+ }
+ catch (IllegalArgumentException e)
+ {
+ throw new SensorHubException("Invalid configuration: " + e.getMessage());
+ }
+ }
+
+
+ @Override
+ protected void doStart() throws SensorHubException
+ {
+ super.doStart();
+
+ // try to attach to CONSYS API REST service
+ getParentHub().getModuleRegistry().waitForModuleType(ConSysApiService.class, ModuleState.STARTED)
+ .thenAccept(service -> {
+ var servlet = service.getServlet();
+
+ if (mqttServer != null)
+ {
+ // format subtopics: standard formats and the API's custom formats
+ var formats = new ArrayList<>(TopicMapper.STANDARD_FORMATS);
+ for (var f: service.getConfiguration().customFormats)
+ formats.add(f.mimeType);
+ var topics = new TopicMapper(config.nodeId, formats);
+ mqttConnector = new CsApiMqttConnector(servlet, topics);
+ mqttServer.registerHandler(topics.getPrefix(), mqttConnector);
+ getLogger().info("CS API MQTT handler registered on '{}', format subtopics {}", topics.getPrefix(), topics.formats.keySet());
+ setState(ModuleState.STARTED);
+ }
+ })
+ .orTimeout(30, TimeUnit.SECONDS)
+ .exceptionally(e -> {
+ reportError("Could not attach to CONSYS API service", e);
+ return null;
+ });
+ }
+
+
+ @Override
+ protected void doStop() throws SensorHubException
+ {
+ super.doStop();
+
+ // stop MQTT extension
+ if (mqttConnector != null && mqttServer != null)
+ {
+ // the connector keeps the prefix it was registered with
+ mqttServer.unregisterHandler(mqttConnector.topics.getPrefix(), mqttConnector);
+ getLogger().info("CS API MQTT handler unregistered");
+ mqttConnector.stop();
+ mqttConnector = null;
+ }
+ }
+}
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
new file mode 100644
index 000000000..c0b56d120
--- /dev/null
+++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceConfig.java
@@ -0,0 +1,36 @@
+/***************************** 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) 2021 Sensia Software LLC. All Rights Reserved.
+
+******************************* END LICENSE BLOCK ***************************/
+
+package org.sensorhub.impl.service.csapi.mqtt;
+
+import org.sensorhub.api.config.DisplayInfo;
+import org.sensorhub.api.service.ServiceConfig;
+
+
+/**
+ *
+ * Configuration class for the Connected Systems API MQTT service module
+ *
+ *
+ * @author Alex Robin
+ * @since May 9, 2023
+ */
+public class CsApiMqttServiceConfig extends ServiceConfig
+{
+ @DisplayInfo.Required
+ @DisplayInfo(label="Node ID", desc="Prefix of all MQTT topics, as in '{nodeId}/systems/{id}'. "
+ + "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";
+}
diff --git a/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceDescriptor.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceDescriptor.java
new file mode 100644
index 000000000..8e4c2f4f1
--- /dev/null
+++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceDescriptor.java
@@ -0,0 +1,47 @@
+/***************************** 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) 2020 Sensia Software LLC. All Rights Reserved.
+
+******************************* END LICENSE BLOCK ***************************/
+
+package org.sensorhub.impl.service.csapi.mqtt;
+
+import org.sensorhub.api.module.IModule;
+import org.sensorhub.api.module.IModuleProvider;
+import org.sensorhub.api.module.ModuleConfig;
+import org.sensorhub.impl.module.JarModuleProvider;
+
+
+/**
+ *
+ * Descriptor of Connected Systems API MQTT service module, needed for automatic
+ * discovery by the ModuleRegistry.
+ *
+ *
+ * @author Alex Robin
+ * @since May 9, 2023
+ */
+public class CsApiMqttServiceDescriptor extends JarModuleProvider implements IModuleProvider
+{
+ @Override
+ public Class> getModuleClass()
+ {
+ return CsApiMqttService.class;
+ }
+
+
+ @Override
+ public Class getModuleConfigClass()
+ {
+ return CsApiMqttServiceConfig.class;
+ }
+
+}
diff --git a/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/TopicMapper.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/TopicMapper.java
new file mode 100644
index 000000000..599ed20b9
--- /dev/null
+++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/TopicMapper.java
@@ -0,0 +1,208 @@
+/***************************** 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.net.URI;
+import java.net.URISyntaxException;
+import java.util.Collection;
+import java.util.TreeMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.regex.Pattern;
+import org.sensorhub.api.comm.mqtt.InvalidTopicException;
+import org.sensorhub.impl.service.consys.resource.ResourceFormat;
+
+
+/**
+ *
+ * Maps MQTT topics to CS API resource paths, per the OGC CS API Part 3
+ * topic grammar: every topic starts with {@code {nodeId}/}, followed by
+ * the resource path. Topics do not carry query strings.
+ *
+ * Only Resource Data topics are accepted, for example
+ * {@code node1/systems/134/datastreams/12/observations:data}. The plain
+ * {@code :data} topic selects the default representation, and
+ * {@code :data/{format}} selects another one by its format token. The
+ * topic shape is checked here; that each ID belongs to its parent, and
+ * that the resource offers the format, is left to the CS API. Resource
+ * Event topics and wildcard topic filters are rejected until they are
+ * supported.
+ *
+ */
+final class TopicMapper
+{
+ static final String DATA_MARKER = ":data";
+
+ // Resource Data topic templates (OGC CS API Part 3, Resource Data Topics
+ // table), without '{nodeId}/' and ':data'. An ID is one level with no ':'.
+ static final Pattern DATA_PATH = Pattern.compile(
+ "systems(/[^/:]+(/subsystems/[^/:]+)*/("
+ + "subsystems|datastreams|controlstreams|events"
+ + "|datastreams/[^/:]+/observations"
+ + "|controlstreams/[^/:]+/commands(/[^/:]+/(status|result))?))?");
+
+ // formats that the CS API offers without custom formats
+ static final List STANDARD_FORMATS = List.of(
+ ResourceFormat.JSON.getMimeType(),
+ ResourceFormat.GEOJSON.getMimeType(),
+ ResourceFormat.SML_JSON.getMimeType(),
+ ResourceFormat.SML_XML.getMimeType(),
+ ResourceFormat.OM_JSON.getMimeType(),
+ ResourceFormat.SWE_JSON.getMimeType(),
+ ResourceFormat.SWE_TEXT.getMimeType(),
+ ResourceFormat.SWE_XML.getMimeType(),
+ ResourceFormat.SWE_BINARY.getMimeType());
+
+ /** A topic mapped to a CS API resource. {@code format} is null for the default. */
+ record ResourceTopic(URI uri, String format) {}
+
+ final String prefix;
+ final Map formats = new TreeMap<>(); // token -> media type
+
+
+ /**
+ * @param nodeId the node identifier, see {@link #checkNodeId(String)}
+ * @param mediaTypes the media types that format subtopics can select.
+ * Types with no valid token, or with the same token as another
+ * type, get no subtopic.
+ * @throws IllegalArgumentException if the node identifier is not valid
+ */
+ TopicMapper(String nodeId, Collection mediaTypes)
+ {
+ this.prefix = checkNodeId(nodeId) + "/";
+
+ var duplicates = new HashSet();
+ for (var type: mediaTypes)
+ {
+ var token = formatToken(type);
+ if (token.isEmpty() || token.indexOf('/') >= 0 || token.indexOf('#') >= 0)
+ continue;
+ var other = formats.put(token, type);
+ if (other != null && !other.equals(type))
+ duplicates.add(token);
+ }
+ formats.keySet().removeAll(duplicates);
+ }
+
+
+ /**
+ * Make the format token of a media type (CS API Part 3 format name):
+ * remove parameters and the {@code application/} prefix, change
+ * {@code +} to {@code -}, and use lowercase.
+ */
+ static String formatToken(String mediaType)
+ {
+ var token = mediaType;
+ var sep = token.indexOf(';');
+ if (sep >= 0)
+ token = token.substring(0, sep);
+ // media types are case-insensitive, so lowercase first
+ token = token.trim().toLowerCase(Locale.ROOT);
+ if (token.startsWith("application/"))
+ token = token.substring("application/".length());
+ return token.replace('+', '-');
+ }
+
+
+ /**
+ * Check that {@code nodeId} is one MQTT topic level that can be used in
+ * both topic names and topic filters.
+ * @return the node identifier
+ * @throws IllegalArgumentException if it is empty, contains {@code /},
+ * {@code +}, {@code #} or NUL, or starts with {@code $}
+ */
+ static String checkNodeId(String nodeId)
+ {
+ if (nodeId == null || nodeId.isEmpty())
+ throw new IllegalArgumentException("nodeId must not be empty");
+ if (nodeId.charAt(0) == '$')
+ throw new IllegalArgumentException("nodeId must not start with '$'");
+ for (var c: nodeId.toCharArray())
+ {
+ if (c == '/' || c == '+' || c == '#' || c == '\0')
+ throw new IllegalArgumentException("nodeId must not contain '" + (c == '\0' ? "NUL" : c) + "'");
+ }
+ return nodeId;
+ }
+
+
+ /**
+ * @return the topic prefix, {@code {nodeId}/}
+ */
+ String getPrefix()
+ {
+ return prefix;
+ }
+
+
+ /**
+ * @param topic an MQTT topic name
+ * @return the CS API resource that the topic identifies. The URI has no
+ * {@code :data}, and has an {@code f} parameter if the topic has
+ * a format subtopic.
+ * @throws InvalidTopicException if the topic is not under this node,
+ * has a query string, is a wildcard filter, is not a Resource
+ * Data topic, or has an unknown format token
+ */
+ ResourceTopic parse(String topic) throws InvalidTopicException
+ {
+ if (topic == null || !topic.startsWith(prefix))
+ throw new InvalidTopicException("Topic must start with '" + prefix + "'");
+
+ var path = topic.substring(prefix.length());
+ if (path.isEmpty())
+ throw new InvalidTopicException("Topic has no resource path");
+ if (path.indexOf('?') >= 0)
+ throw new InvalidTopicException("Topics must not have a query string");
+
+ // Wildcard filters are not supported yet (see #249, wildcard subscriptions).
+ // Reject them here: URI would read '#' as a fragment, and map
+ // 'node1/systems/134/#' to the parent resource '/systems/134/'.
+ if (path.indexOf('+') >= 0 || path.indexOf('#') >= 0)
+ throw new InvalidTopicException("Wildcard topic filters are not supported");
+
+ // Only Resource Data topics are supported. Resource Event topics
+ // (no ':data') come later (see #249).
+ var marker = path.lastIndexOf(DATA_MARKER);
+ if (marker < 0)
+ throw new InvalidTopicException("Only Resource Data topics (with '" + DATA_MARKER + "') are supported");
+ var tail = path.substring(marker + DATA_MARKER.length());
+ path = path.substring(0, marker);
+ if (!DATA_PATH.matcher(path).matches())
+ throw new InvalidTopicException("Not a Resource Data topic: " + topic);
+
+ // plain ':data' or ':data/{format}', with no fallback for an unknown token
+ String format = null;
+ if (!tail.isEmpty())
+ {
+ var token = tail.startsWith("/") ? tail.substring(1) : "";
+ format = formats.get(token);
+ if (format == null)
+ throw new InvalidTopicException("Unknown format subtopic: " + tail);
+ }
+
+ try
+ {
+ var uri = new URI(null, null, "/" + path, format == null ? null : "f=" + format, null);
+ return new ResourceTopic(uri, format);
+ }
+ catch (URISyntaxException e)
+ {
+ throw new InvalidTopicException("Invalid resource path in topic: " + topic);
+ }
+ }
+}
diff --git a/services/sensorhub-service-csapi-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider b/services/sensorhub-service-csapi-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider
new file mode 100644
index 000000000..58e0550a5
--- /dev/null
+++ b/services/sensorhub-service-csapi-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider
@@ -0,0 +1 @@
+org.sensorhub.impl.service.csapi.mqtt.CsApiMqttServiceDescriptor
diff --git a/services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestTopicMapper.java b/services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestTopicMapper.java
new file mode 100644
index 000000000..e789914cb
--- /dev/null
+++ b/services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestTopicMapper.java
@@ -0,0 +1,175 @@
+/***************************** 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 java.util.List;
+import org.junit.Test;
+import org.sensorhub.api.comm.mqtt.InvalidTopicException;
+
+
+public class TestTopicMapper
+{
+ static TopicMapper mapper()
+ {
+ return new TopicMapper("node1", TopicMapper.STANDARD_FORMATS);
+ }
+
+
+ @Test
+ public void testValidNodeIds()
+ {
+ for (var id: new String[] {"api", "node1", "my-node", "a.b_c", "x"})
+ assertEquals(id, TopicMapper.checkNodeId(id));
+ }
+
+
+ @Test
+ public void testInvalidNodeIds()
+ {
+ for (var id: new String[] {null, "", "a/b", "a+b", "a#b", "a\0b", "$sys", "/api"})
+ {
+ try
+ {
+ TopicMapper.checkNodeId(id);
+ fail("Accepted invalid nodeId: " + id);
+ }
+ catch (IllegalArgumentException e)
+ {
+ // expected
+ }
+ }
+ }
+
+
+ @Test
+ public void testPrefix()
+ {
+ assertEquals("node1/", mapper().getPrefix());
+ }
+
+
+ @Test
+ public void testToResourceUri() throws InvalidTopicException
+ {
+ // designated topics from the Part 3 Resource Data Topics table
+ var topics = mapper();
+ var sys = "systems/134/subsystems/7";
+ for (var path: new String[] {
+ "systems",
+ "systems/134/subsystems",
+ sys + "/subsystems",
+ sys + "/datastreams",
+ sys + "/controlstreams",
+ sys + "/events",
+ "systems/134/datastreams/12/observations",
+ sys + "/datastreams/12/observations",
+ sys + "/controlstreams/3/commands",
+ sys + "/controlstreams/3/commands/99/status",
+ sys + "/controlstreams/3/commands/99/result"
+ })
+ {
+ var res = topics.parse("node1/" + path + ":data");
+ assertEquals("/" + path, res.uri().toString());
+ assertNull(res.format());
+ }
+ }
+
+
+ @Test
+ public void testFormatTokens()
+ {
+ assertEquals("json", TopicMapper.formatToken("application/json"));
+ assertEquals("om-json", TopicMapper.formatToken("application/om+json"));
+ assertEquals("swe-binary", TopicMapper.formatToken("application/swe+binary"));
+ assertEquals("swe-csv", TopicMapper.formatToken("application/swe+csv"));
+ assertEquals("geo-json", TopicMapper.formatToken("application/geo+json"));
+ assertEquals("sml-xml", TopicMapper.formatToken("Application/SML+XML; version=2.0"));
+ assertEquals("text/csv", TopicMapper.formatToken("text/csv"));
+ }
+
+
+ @Test
+ public void testFormatSubtopics() throws InvalidTopicException
+ {
+ var topics = mapper();
+ var obs = "node1/systems/1/datastreams/2/observations:data/";
+ for (var type: TopicMapper.STANDARD_FORMATS)
+ {
+ var res = topics.parse(obs + TopicMapper.formatToken(type));
+ assertEquals(type, res.format());
+ assertEquals("/systems/1/datastreams/2/observations", res.uri().getPath());
+ assertEquals("f=" + type, res.uri().getQuery());
+ }
+ }
+
+
+ @Test
+ public void testFormatsWithoutSubtopic()
+ {
+ // tokens with '/' and tokens shared by two types get no subtopic
+ var topics = new TopicMapper("node1", List.of(
+ "application/json", "video/x-motion-jpeg", "text/csv", "application/a+b", "application/a-b"));
+ assertEquals(List.of("json"), List.copyOf(topics.formats.keySet()));
+ }
+
+
+ @Test
+ public void testRejectedTopics()
+ {
+ var topics = mapper();
+ for (var topic: new String[] {
+ null,
+ "node1", // no resource path
+ "node1/", // no resource path
+ "node2/systems:data", // other node
+ "/api/systems:data", // 1.x topic
+ "node1x/systems:data", // prefix of a longer level
+ "node1/systems/1/datastreams/2/observations:data?f=json", // query string
+ "node1/systems/+/datastreams:data", // wildcard, not supported yet
+ "node1/systems/134/#", // wildcard, not supported yet
+ "node1/systems/1/datastreams/2/observations", // event topic, not supported yet
+ "node1/systems/134", // event topic, not supported yet
+ "node1/systems/1/datastreams/2/observations:data/swe-proto", // unknown format
+ "node1/systems/1/datastreams/2/observations:data/", // empty format
+ "node1/systems/1/datastreams/2/observations:data/SWE-JSON", // tokens are lowercase
+ "node1/systems/1/datastreams/2/observations:data/swe-json/x", // format is one level
+ "node1/systems/1/datastreams/2/observations:datax", // not ':data'
+ "node1/systems:data/subsystems", // ':data' not on the last level
+ "node1/systems/134:data", // ':data' on an ID
+ "node1/systems/1:x/datastreams:data", // ':' in an ID
+ "node1/systems/1/datastreams/2:data", // ':data' on an ID
+ "node1/observations:data", // flat, not allowed
+ "node1/commands:data", // flat, not allowed
+ "node1/datastreams/2/observations:data", // no system path
+ "node1/systems/1/samplingFeatures:data", // not an allowed type
+ "node1/systems/1/datastreams/2/schema:data", // not an allowed type
+ "node1/systems/1/subsystems:data/x", // unknown format
+ "node1/systems/1//datastreams:data", // empty ID
+ "node1/:data" // no collection
+ })
+ {
+ try
+ {
+ topics.parse(topic);
+ fail("Accepted invalid topic: " + topic);
+ }
+ catch (InvalidTopicException e)
+ {
+ // expected
+ }
+ }
+ }
+}
diff --git a/services/sensorhub-service-csapi-mqtt/src/test/resources/empty b/services/sensorhub-service-csapi-mqtt/src/test/resources/empty
new file mode 100644
index 000000000..e69de29bb