From ee2787a292b907c093bc33808d9b0190322c4944 Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 2 Nov 2026 21:06:05 -0600 Subject: [PATCH 1/6] feat(consys-mqtt): add {nodeId} topic prefix --- .../consys/mqtt/ConSysApiMqttConnector.java | 50 +++----- .../consys/mqtt/ConSysApiMqttService.java | 38 ++++-- .../mqtt/ConSysApiMqttServiceConfig.java | 7 +- .../impl/service/consys/mqtt/TopicMapper.java | 108 ++++++++++++++++++ .../service/consys/mqtt/TestTopicMapper.java | 94 +++++++++++++++ 5 files changed, 251 insertions(+), 46 deletions(-) create mode 100644 services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java create mode 100644 services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java index d3f47b48b..8a823914c 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java @@ -16,8 +16,6 @@ import java.io.IOException; import java.io.OutputStream; -import java.net.URI; -import java.net.URISyntaxException; import java.nio.ByteBuffer; import java.security.AccessControlException; import java.util.Map; @@ -52,7 +50,7 @@ public class ConSysApiMqttConnector implements IMqttHandler { ConSysApiServlet servlet; - String endpoint; + TopicMapper topics; Map subscribers = new ConcurrentHashMap<>(); @@ -117,16 +115,19 @@ public void close() } - public ConSysApiMqttConnector(ConSysApiServlet servlet, String endpoint) + public ConSysApiMqttConnector(ConSysApiServlet servlet, TopicMapper topics) { this.servlet = servlet; - this.endpoint = endpoint; + this.topics = topics; } @Override public void onSubscribe(String userID, String topic, IMqttServer server) throws InvalidTopicException { + // reject invalid topics before a subscriber is created + var resourceUri = topics.toResourceUri(topic); + try { // register new subscription if needed @@ -139,7 +140,7 @@ public void onSubscribe(String userID, String topic, IMqttServer server) throws // always evaluate request because we need to check for permissions // even if subscriber was already created - var ctx = getResourceContext(topic, sub); + var ctx = new RequestContext(servlet, resourceUri, sub); servlet.getSecurityHandler().setCurrentUser(userID); servlet.getRootHandler().doGet(ctx); @@ -147,7 +148,7 @@ public void onSubscribe(String userID, String topic, IMqttServer server) throws sub.numSubscribers.incrementAndGet(); sub.maybeStart(); } - catch (SecurityException | InvalidTopicException e) + catch (SecurityException e) { throw e; } @@ -217,6 +218,13 @@ public void onPublish(String userID, String topic, ByteBuffer payload, ByteBuffe { handleInvalidRequestException(e); } + catch (NullPointerException e) + { + // Resource builders reject a missing required field with an NPE, + // for example a command without 'parameters'. Report it as a bad + // payload, not an internal error. + throw new InvalidPayloadException(e.getMessage()); + } catch (Exception e) { throw new IllegalStateException("Internal error publishing to topic " + topic, e); @@ -228,41 +236,15 @@ public void onPublish(String userID, String topic, ByteBuffer payload, ByteBuffe } - private RequestContext getResourceContext(String topic, MqttSubscriber streamHandler) throws InvalidTopicException - { - return new RequestContext( - servlet, - getResourceUri(topic), - streamHandler); - } - - private RequestContext getResourceContext(String topic, ByteBuffer payload) throws InvalidTopicException { return new RequestContext( servlet, - getResourceUri(topic), + topics.toResourceUri(topic), new ByteBufferInputStream(payload)); } - private URI getResourceUri(String topic) throws InvalidTopicException - { - try - { - // remove the base URL part - topic = topic.replaceFirst(endpoint, ""); - - // parse URI (this also URL decodes the query string) - return new URI(topic); - } - catch (URISyntaxException e) - { - throw new InvalidTopicException("Invalid SWE API resource URI"); - } - } - - private void handleInvalidRequestException(InvalidRequestException e) throws MqttException, SecurityException { switch (e.getErrorCode()) diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java index a88873082..fc2e2dfd2 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java @@ -33,25 +33,40 @@ public class ConSysApiMqttService extends AbstractMqttServiceModule implements IServiceModule { protected ConSysApiMqttConnector mqttConnector; - protected String endPoint; + protected TopicMapper topics; + + + @Override + protected void doInit() throws SensorHubException + { + super.doInit(); + + try + { + topics = new TopicMapper(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(); - endPoint = service.getConfiguration().endPoint; - + if (mqttServer != null) { - mqttConnector = new ConSysApiMqttConnector(servlet, endPoint); - mqttServer.registerHandler(endPoint, mqttConnector); - getLogger().info("CONSYS API MQTT handler registered"); + mqttConnector = new ConSysApiMqttConnector(servlet, topics); + mqttServer.registerHandler(topics.getPrefix(), mqttConnector); + getLogger().info("CONSYS API MQTT handler registered on '{}'", topics.getPrefix()); setState(ModuleState.STARTED); } }) @@ -71,10 +86,11 @@ protected void doStop() throws SensorHubException // stop MQTT extension if (mqttConnector != null && mqttServer != null) { - mqttServer.unregisterHandler(endPoint, mqttConnector); - getLogger().info("CONSYS API MQTT handler unregistered"); - mqttConnector.stop(); - mqttConnector = null; + // the connector keeps the prefix it was registered with + mqttServer.unregisterHandler(mqttConnector.topics.getPrefix(), mqttConnector); + getLogger().info("CONSYS API MQTT handler unregistered"); + mqttConnector.stop(); + mqttConnector = null; } } } diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java index 4cd1d5512..790d06142 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java @@ -14,6 +14,7 @@ package org.sensorhub.impl.service.consys.mqtt; +import org.sensorhub.api.config.DisplayInfo; import org.sensorhub.api.service.ServiceConfig; @@ -27,5 +28,9 @@ */ public class ConSysApiMqttServiceConfig 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-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java new file mode 100644 index 000000000..a29871851 --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java @@ -0,0 +1,108 @@ +/***************************** 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.consys.mqtt; + +import java.net.URI; +import java.net.URISyntaxException; +import org.sensorhub.api.comm.mqtt.InvalidTopicException; + + +/** + *

+ * 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. Wildcard topic + * filters are rejected until wildcard subscriptions are supported. + *

+ */ +final class TopicMapper +{ + final String prefix; + + + /** + * @param nodeId the node identifier, see {@link #checkNodeId(String)} + * @throws IllegalArgumentException if the node identifier is not valid + */ + TopicMapper(String nodeId) + { + this.prefix = checkNodeId(nodeId) + "/"; + } + + + /** + * 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 URI of the CS API resource that the topic identifies + * @throws InvalidTopicException if the topic is not under this node, + * has no resource path, has a query string, or is a wildcard filter + */ + URI toResourceUri(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"); + + try + { + return new URI("/" + path); + } + catch (URISyntaxException e) + { + throw new InvalidTopicException("Invalid resource path in topic: " + topic); + } + } +} diff --git a/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java b/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java new file mode 100644 index 000000000..1e31b6dec --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java @@ -0,0 +1,94 @@ +/***************************** 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.consys.mqtt; + +import static org.junit.Assert.*; +import org.junit.Test; +import org.sensorhub.api.comm.mqtt.InvalidTopicException; + + +public class TestTopicMapper +{ + @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/", new TopicMapper("node1").getPrefix()); + } + + + @Test + public void testToResourceUri() throws InvalidTopicException + { + var topics = new TopicMapper("node1"); + assertEquals("/systems/134", topics.toResourceUri("node1/systems/134").toString()); + assertEquals("/datastreams/abc/observations", topics.toResourceUri("node1/datastreams/abc/observations").toString()); + } + + + @Test + public void testRejectedTopics() + { + var topics = new TopicMapper("node1"); + for (var topic: new String[] { + null, + "node1", // no resource path + "node1/", // no resource path + "node2/systems/134", // other node + "/api/systems/134", // 1.x topic + "node1x/systems/134", // prefix of a longer level + "node1/datastreams/abc/observations?f=json", // query string + "node1/systems/+", // wildcard, not supported yet + "node1/systems/134/#", // wildcard, not supported yet + "node1/systems/+/datastreams/+/observations" + }) + { + try + { + topics.toResourceUri(topic); + fail("Accepted invalid topic: " + topic); + } + catch (InvalidTopicException e) + { + // expected + } + } + } +} From 36b834ae31d1ba6b08d16c2cfa37ad362f8d4fb0 Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 2 Nov 2026 21:33:49 -0600 Subject: [PATCH 2/6] feat(consys-mqtt): add :data topics and bump version to 3.0.0 due to breaking changes --- .../sensorhub-service-consys-mqtt/README.md | 97 +++++++++++++++++++ .../build.gradle | 6 ++ .../consys/mqtt/ConSysApiMqttConnector.java | 6 ++ .../impl/service/consys/mqtt/TopicMapper.java | 36 ++++++- .../service/consys/mqtt/TestTopicMapper.java | 52 +++++++--- 5 files changed, 182 insertions(+), 15 deletions(-) create mode 100644 services/sensorhub-service-consys-mqtt/README.md diff --git a/services/sensorhub-service-consys-mqtt/README.md b/services/sensorhub-service-consys-mqtt/README.md new file mode 100644 index 000000000..e411b9834 --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/README.md @@ -0,0 +1,97 @@ +# Connected Systems API MQTT Service + +This module adds MQTT to the OSH Connected Systems (CS) API. +It follows the MQTT binding in OGC CS API Part 3 (draft). +The work is not complete. See [Status](#status) and [Known limits](#known-limits). + +## What it does + +- You subscribe to a topic to get live data. Examples are observations and command status. +- You publish to a topic to create a resource. Examples are a system, an observation or a command. +- Each publication is one create operation. MQTT cannot update or delete resources. +- The CS API checks each request. The same users and permissions apply as for HTTP. + +## Setup + +You need these modules on the node: + +- Connected Systems API Service +- MQTT Server (`sensorhub-service-mqtt-hivemq`) +- Connected Systems API MQTT Service (this module) + +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`. + +## Topics + +All topics start with `{nodeId}/`. Then comes the resource path, as in the HTTP API. +Data topics end with `:data` on the last collection name. + +`{systemPath}` is the full path of the system: `systems/{id}`, then `/subsystems/{id}` for each subsystem level. + +Supported Resource Data topics: + +- `{nodeId}/systems:data`: top-level systems +- `{nodeId}/{systemPath}/subsystems:data` +- `{nodeId}/{systemPath}/datastreams:data` +- `{nodeId}/{systemPath}/controlstreams:data` +- `{nodeId}/{systemPath}/events:data`: system events +- `{nodeId}/{systemPath}/datastreams/{id}/observations:data` +- `{nodeId}/{systemPath}/controlstreams/{id}/commands:data` +- `{nodeId}/{systemPath}/controlstreams/{id}/commands/{id}/status:data` +- `{nodeId}/{systemPath}/controlstreams/{id}/commands/{id}/result:data` + +The service rejects all other topics. + +## Examples + +Get observations from datastream `12` of system `134`: + +``` +subscribe node1/systems/134/datastreams/12/observations:data +``` + +Send a command to control stream `3` of system `134`: + +``` +publish node1/systems/134/controlstreams/3/commands:data +payload {"parameters": {"action": "SET_STEP", "intValue": 5, "countDown": false}} +``` + +The fields in `parameters` come from the control stream schema. Give all of them. + +## Status + +Done: + +- `{nodeId}` topic prefix +- Resource Data topics (`:data`) with the full `{systemPath}` + +Not done yet: + +- Format subtopics (`:data/{format}`). Each topic uses the default format. +- Resource Event topics (topics without `:data`) and CloudEvents messages +- Wildcard subscriptions (`+`, `#`) +- MQTT 5 Content Type +- AsyncAPI service description + +## Known limits + +- The service rejects topics with `+` or `#`. +- The service rejects topics without `:data`. +- The service checks the topic shape only. The CS API does not check that each ID in the path belongs to the resource before it. For example, a command sent under the wrong system path still runs. +- A subscription to the status of one command gets the status of all commands on that control stream. +- Publications to `status:data` and `result:data` are rejected. The CS API accepts them only at `/controlstreams/{id}/status` over HTTP. +- Subscriptions to `systems:data`, `subsystems:data`, `datastreams:data`, `controlstreams:data` and `result:data` are rejected. The CS API cannot stream these resources yet. You can publish to the first four to create resources. +- Error replies (MQTT 5 reason codes) come only from the embedded OSH broker. MQTT 3.1.1 clients get no reason. Asynchronous error reports are not done yet. +- A JSON command must put its fields in `parameters`. If it does not, the service rejects the command. +- Each topic uses the default JSON format of its resource. Other formats need format subtopics, which are not done yet. + +## Change in version 3.0.0 + +This is a breaking change. +Earlier versions used the HTTP path as the topic, for example `/api/datastreams/12/observations`. +These topics do not work now. Use `{nodeId}/{systemPath}/...:data` topics. diff --git a/services/sensorhub-service-consys-mqtt/build.gradle b/services/sensorhub-service-consys-mqtt/build.gradle index fe143ab12..1b68d4f99 100644 --- a/services/sensorhub-service-consys-mqtt/build.gradle +++ b/services/sensorhub-service-consys-mqtt/build.gradle @@ -1,5 +1,6 @@ description = 'Connected Systems API MQTT Extension' ext.details = 'MQTT Extension for OGC API Connected Systems' +version = '3.0.0' dependencies { @@ -8,6 +9,11 @@ dependencies { implementation project(':sensorhub-comm-mqtt') } +// put README.md next to the config class, for the Admin UI README tab +processResources { + from('README.md') { into 'org/sensorhub/impl/service/consys/mqtt' } +} + // add info to OSGI manifest osgi { manifest { diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java index 8a823914c..9139a2a46 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java @@ -156,6 +156,11 @@ public void onSubscribe(String userID, String topic, IMqttServer server) throws { throw new InvalidTopicException(e.getMessage()); } + catch (UnsupportedOperationException e) + { + // some resources cannot be streamed yet (for example command results) + throw new InvalidTopicException("Streaming not supported on this resource"); + } catch (Exception e) { throw new IllegalStateException("Internal error subscribing to topic " + topic, e); @@ -251,6 +256,7 @@ private void handleInvalidRequestException(InvalidRequestException e) throws Mqt { case NOT_FOUND: case BAD_REQUEST: + case UNSUPPORTED_OPERATION: // topic is not writable throw new InvalidTopicException(e.getMessage()); case BAD_PAYLOAD: diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java index a29871851..175a21700 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java @@ -16,6 +16,7 @@ import java.net.URI; import java.net.URISyntaxException; +import java.util.regex.Pattern; import org.sensorhub.api.comm.mqtt.InvalidTopicException; @@ -23,12 +24,27 @@ *

* 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. Wildcard topic - * filters are rejected until wildcard subscriptions are supported. + * 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 topic + * shape is checked here; that each ID belongs to its parent is left to the + * CS API. Resource Event topics, format subtopics 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))?))?"); + final String prefix; @@ -75,9 +91,11 @@ String getPrefix() /** * @param topic an MQTT topic name - * @return the URI of the CS API resource that the topic identifies + * @return the URI of the CS API resource that the topic identifies, + * without {@code :data} * @throws InvalidTopicException if the topic is not under this node, - * has no resource path, has a query string, or is a wildcard filter + * has a query string, is a wildcard filter, or is not a + * Resource Data topic */ URI toResourceUri(String topic) throws InvalidTopicException { @@ -96,6 +114,16 @@ URI toResourceUri(String topic) throws InvalidTopicException 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') and format subtopics come later (see #249). + if (path.contains(DATA_MARKER + "/")) + throw new InvalidTopicException("Format subtopics are not supported"); + if (!path.endsWith(DATA_MARKER)) + throw new InvalidTopicException("Only Resource Data topics (ending with '" + DATA_MARKER + "') are supported"); + path = path.substring(0, path.length() - DATA_MARKER.length()); + if (!DATA_PATH.matcher(path).matches()) + throw new InvalidTopicException("Not a Resource Data topic: " + topic); + try { return new URI("/" + path); diff --git a/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java b/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java index 1e31b6dec..1095eb60d 100644 --- a/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java +++ b/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java @@ -57,9 +57,25 @@ public void testPrefix() @Test public void testToResourceUri() throws InvalidTopicException { + // designated topics from the Part 3 Resource Data Topics table var topics = new TopicMapper("node1"); - assertEquals("/systems/134", topics.toResourceUri("node1/systems/134").toString()); - assertEquals("/datastreams/abc/observations", topics.toResourceUri("node1/datastreams/abc/observations").toString()); + 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" + }) + { + assertEquals("/" + path, topics.toResourceUri("node1/" + path + ":data").toString()); + } } @@ -69,15 +85,29 @@ public void testRejectedTopics() var topics = new TopicMapper("node1"); for (var topic: new String[] { null, - "node1", // no resource path - "node1/", // no resource path - "node2/systems/134", // other node - "/api/systems/134", // 1.x topic - "node1x/systems/134", // prefix of a longer level - "node1/datastreams/abc/observations?f=json", // query string - "node1/systems/+", // wildcard, not supported yet - "node1/systems/134/#", // wildcard, not supported yet - "node1/systems/+/datastreams/+/observations" + "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-json", // format, not supported yet + "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", // trailing level + "node1/systems/1//datastreams:data", // empty ID + "node1/:data" // no collection }) { try From ad4cc6450c39adacdbf7ce87b973bc168c1c3795 Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 2 Nov 2026 22:16:33 -0600 Subject: [PATCH 3/6] feat(consys-mqtt): add :data/{format} subtopics --- .../sensorhub-service-consys-mqtt/README.md | 58 +++++++++- .../consys/mqtt/ConSysApiMqttConnector.java | 38 ++++--- .../consys/mqtt/ConSysApiMqttService.java | 11 +- .../impl/service/consys/mqtt/TopicMapper.java | 106 +++++++++++++++--- .../service/consys/mqtt/TestTopicMapper.java | 65 +++++++++-- 5 files changed, 236 insertions(+), 42 deletions(-) diff --git a/services/sensorhub-service-consys-mqtt/README.md b/services/sensorhub-service-consys-mqtt/README.md index e411b9834..9c5e4c5a0 100644 --- a/services/sensorhub-service-consys-mqtt/README.md +++ b/services/sensorhub-service-consys-mqtt/README.md @@ -46,6 +46,52 @@ Supported Resource Data topics: The service rejects all other topics. +## Formats + +The plain `:data` topic uses the default format of the resource. +To use a different format, add `/{format}` after `:data`: + +``` +node1/systems/134/datastreams/12/observations:data/swe-csv +``` + +The format name comes from the media type: + +1. Remove parameters (`;` and after). +2. Remove `application/`. +3. Change `+` to `-`. +4. Use lowercase. + +Format names: + +- `json`: `application/json` +- `geo-json`: `application/geo+json` +- `sml-json`: `application/sml+json` +- `sml-xml`: `application/sml+xml` +- `om-json`: `application/om+json` +- `swe-json`: `application/swe+json` +- `swe-csv`: `application/swe+csv` +- `swe-xml`: `application/swe+xml` +- `swe-binary`: `application/swe+binary` +- Custom formats of the CS API service also get a name, if the name has no `/` or `#`. + For example, `video/x-motion-jpeg` gets no format subtopic. + +On a format subtopic: + +- A subscription gets the data in that format only. +- A publication must use that format. The service does not detect the format. +- The service rejects an unknown format name. It does not use the default format. +- The service rejects a format that the resource does not support. + Each datastream lists its formats in `formats` over HTTP. + +Formats found to work in tests: + +- Observations, subscribe: `json`, `om-json`, `swe-json`, `swe-csv`, `swe-xml`, `swe-binary` +- Video observations (Axis camera), subscribe: `swe-binary` (MJPEG and H264 frames), `om-json` (no frame data) +- Observations, publish: `om-json`, `swe-json`, `swe-csv` +- Commands, publish: `json`, `swe-json`, `swe-csv` +- Systems, publish: `geo-json`, `sml-json` + ## Examples Get observations from datastream `12` of system `134`: @@ -54,6 +100,12 @@ Get observations from datastream `12` of system `134`: subscribe node1/systems/134/datastreams/12/observations:data ``` +Get the same observations as CSV: + +``` +subscribe node1/systems/134/datastreams/12/observations:data/swe-csv +``` + Send a command to control stream `3` of system `134`: ``` @@ -69,10 +121,10 @@ Done: - `{nodeId}` topic prefix - Resource Data topics (`:data`) with the full `{systemPath}` +- Format subtopics (`:data/{format}`) Not done yet: -- Format subtopics (`:data/{format}`). Each topic uses the default format. - Resource Event topics (topics without `:data`) and CloudEvents messages - Wildcard subscriptions (`+`, `#`) - MQTT 5 Content Type @@ -88,7 +140,9 @@ Not done yet: - Subscriptions to `systems:data`, `subsystems:data`, `datastreams:data`, `controlstreams:data` and `result:data` are rejected. The CS API cannot stream these resources yet. You can publish to the first four to create resources. - Error replies (MQTT 5 reason codes) come only from the embedded OSH broker. MQTT 3.1.1 clients get no reason. Asynchronous error reports are not done yet. - A JSON command must put its fields in `parameters`. If it does not, the service rejects the command. -- Each topic uses the default JSON format of its resource. Other formats need format subtopics, which are 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. ## Change in version 3.0.0 diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java index 9139a2a46..4ae185414 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java @@ -126,7 +126,8 @@ public ConSysApiMqttConnector(ConSysApiServlet servlet, TopicMapper topics) public void onSubscribe(String userID, String topic, IMqttServer server) throws InvalidTopicException { // reject invalid topics before a subscriber is created - var resourceUri = topics.toResourceUri(topic); + var resource = topics.parse(topic); + var accepted = false; try { @@ -140,12 +141,13 @@ public void onSubscribe(String userID, String topic, IMqttServer server) throws // always evaluate request because we need to check for permissions // even if subscriber was already created - var ctx = new RequestContext(servlet, resourceUri, sub); + var ctx = new RequestContext(servlet, resource.uri(), sub); servlet.getSecurityHandler().setCurrentUser(userID); servlet.getRootHandler().doGet(ctx); // start stream if all went well sub.numSubscribers.incrementAndGet(); + accepted = true; sub.maybeStart(); } catch (SecurityException e) @@ -163,11 +165,28 @@ public void onSubscribe(String userID, String topic, IMqttServer server) throws } catch (Exception e) { + // the CS API can wrap a request error, for example an unsupported format + for (var cause = e.getCause(); cause != null; cause = cause.getCause()) + { + if (cause instanceof InvalidRequestException) + throw new InvalidTopicException(cause.getMessage()); + } throw new IllegalStateException("Internal error subscribing to topic " + topic, e); } finally { servlet.getSecurityHandler().clearCurrentUser(); + + // do not keep a subscriber that no client uses + if (!accepted) + { + subscribers.computeIfPresent(topic, (k, sub) -> { + if (sub.numSubscribers.get() > 0) + return sub; + sub.close(); + return null; + }); + } } } @@ -195,8 +214,10 @@ public void onPublish(String userID, String topic, ByteBuffer payload, ByteBuffe { try { - var ctx = getResourceContext(topic, payload); - ctx.setRequestContentType(ResourceFormat.AUTO.getMimeType()); + // the format subtopic selects the payload format, else detect it + var resource = topics.parse(topic); + var ctx = new RequestContext(servlet, resource.uri(), new ByteBufferInputStream(payload)); + ctx.setRequestContentType(resource.format() != null ? resource.format() : ResourceFormat.AUTO.getMimeType()); if (correlData != null) { @@ -241,15 +262,6 @@ public void onPublish(String userID, String topic, ByteBuffer payload, ByteBuffe } - private RequestContext getResourceContext(String topic, ByteBuffer payload) throws InvalidTopicException - { - return new RequestContext( - servlet, - topics.toResourceUri(topic), - new ByteBufferInputStream(payload)); - } - - private void handleInvalidRequestException(InvalidRequestException e) throws MqttException, SecurityException { switch (e.getErrorCode()) diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java index fc2e2dfd2..bbcb4e6c4 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java @@ -14,6 +14,7 @@ package org.sensorhub.impl.service.consys.mqtt; +import java.util.ArrayList; import java.util.concurrent.TimeUnit; import org.sensorhub.api.common.SensorHubException; import org.sensorhub.api.module.ModuleEvent.ModuleState; @@ -33,7 +34,6 @@ public class ConSysApiMqttService extends AbstractMqttServiceModule implements IServiceModule { protected ConSysApiMqttConnector mqttConnector; - protected TopicMapper topics; @Override @@ -43,7 +43,7 @@ protected void doInit() throws SensorHubException try { - topics = new TopicMapper(config.nodeId); + TopicMapper.checkNodeId(config.nodeId); } catch (IllegalArgumentException e) { @@ -64,9 +64,14 @@ protected void doStart() throws SensorHubException 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 ConSysApiMqttConnector(servlet, topics); mqttServer.registerHandler(topics.getPrefix(), mqttConnector); - getLogger().info("CONSYS API MQTT handler registered on '{}'", topics.getPrefix()); + getLogger().info("CONSYS API MQTT handler registered on '{}', format subtopics {}", topics.getPrefix(), topics.formats.keySet()); setState(ModuleState.STARTED); } }) diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java index 175a21700..983090758 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java @@ -16,8 +16,15 @@ 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; /** @@ -27,10 +34,13 @@ * 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 topic - * shape is checked here; that each ID belongs to its parent is left to the - * CS API. Resource Event topics, format subtopics and wildcard topic - * filters are rejected until they are supported. + * {@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 @@ -45,16 +55,66 @@ final class TopicMapper + "|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) + 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('+', '-'); } @@ -91,13 +151,14 @@ String getPrefix() /** * @param topic an MQTT topic name - * @return the URI of the CS API resource that the topic identifies, - * without {@code :data} + * @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, or is not a - * Resource Data topic + * has a query string, is a wildcard filter, is not a Resource + * Data topic, or has an unknown format token */ - URI toResourceUri(String topic) throws InvalidTopicException + ResourceTopic parse(String topic) throws InvalidTopicException { if (topic == null || !topic.startsWith(prefix)) throw new InvalidTopicException("Topic must start with '" + prefix + "'"); @@ -115,18 +176,29 @@ URI toResourceUri(String topic) throws InvalidTopicException throw new InvalidTopicException("Wildcard topic filters are not supported"); // Only Resource Data topics are supported. Resource Event topics - // (no ':data') and format subtopics come later (see #249). - if (path.contains(DATA_MARKER + "/")) - throw new InvalidTopicException("Format subtopics are not supported"); - if (!path.endsWith(DATA_MARKER)) - throw new InvalidTopicException("Only Resource Data topics (ending with '" + DATA_MARKER + "') are supported"); - path = path.substring(0, path.length() - DATA_MARKER.length()); + // (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 { - return new URI("/" + path); + var uri = new URI(null, null, "/" + path, format == null ? null : "f=" + format, null); + return new ResourceTopic(uri, format); } catch (URISyntaxException e) { diff --git a/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java b/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java index 1095eb60d..03a6c401b 100644 --- a/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java +++ b/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java @@ -15,12 +15,19 @@ package org.sensorhub.impl.service.consys.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() { @@ -50,7 +57,7 @@ public void testInvalidNodeIds() @Test public void testPrefix() { - assertEquals("node1/", new TopicMapper("node1").getPrefix()); + assertEquals("node1/", mapper().getPrefix()); } @@ -58,7 +65,7 @@ public void testPrefix() public void testToResourceUri() throws InvalidTopicException { // designated topics from the Part 3 Resource Data Topics table - var topics = new TopicMapper("node1"); + var topics = mapper(); var sys = "systems/134/subsystems/7"; for (var path: new String[] { "systems", @@ -74,15 +81,55 @@ public void testToResourceUri() throws InvalidTopicException sys + "/controlstreams/3/commands/99/result" }) { - assertEquals("/" + path, topics.toResourceUri("node1/" + path + ":data").toString()); + 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 = new TopicMapper("node1"); + var topics = mapper(); for (var topic: new String[] { null, "node1", // no resource path @@ -95,7 +142,11 @@ public void testRejectedTopics() "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-json", // format, 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 @@ -105,14 +156,14 @@ public void testRejectedTopics() "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", // trailing level + "node1/systems/1/subsystems:data/x", // unknown format "node1/systems/1//datastreams:data", // empty ID "node1/:data" // no collection }) { try { - topics.toResourceUri(topic); + topics.parse(topic); fail("Accepted invalid topic: " + topic); } catch (InvalidTopicException e) From ead2ead70b5372ddc9cf3aea459b8e723a5e83bc Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 2 Nov 2026 23:01:46 -0600 Subject: [PATCH 4/6] docs(consys-mqtt): add 3.0.0 migration notes --- .../sensorhub-service-consys-mqtt/README.md | 43 +++++++++++++++++-- 1 file changed, 40 insertions(+), 3 deletions(-) diff --git a/services/sensorhub-service-consys-mqtt/README.md b/services/sensorhub-service-consys-mqtt/README.md index 9c5e4c5a0..f3a4ebe01 100644 --- a/services/sensorhub-service-consys-mqtt/README.md +++ b/services/sensorhub-service-consys-mqtt/README.md @@ -146,6 +146,43 @@ Not done yet: ## Change in version 3.0.0 -This is a breaking change. -Earlier versions used the HTTP path as the topic, for example `/api/datastreams/12/observations`. -These topics do not work now. Use `{nodeId}/{systemPath}/...:data` topics. +This is a breaking change. Clients of version 2.x and earlier must change their topics. +The module version was inherited from osh-core before (2.0.1). Now the module has its own version. + +### Topics + +Old topics used the HTTP endpoint of the CS API service (default `/api`) and the HTTP path. +New topics use `{nodeId}`, the full `{systemPath}` and `:data`. + +- `/api/datastreams/{dsId}/observations` + - becomes `{nodeId}/systems/{sysId}/datastreams/{dsId}/observations:data` +- `/api/controlstreams/{csId}/commands` + - becomes `{nodeId}/systems/{sysId}/controlstreams/{csId}/commands:data` +- `/api/controlstreams/{csId}/status` + - becomes `{nodeId}/systems/{sysId}/controlstreams/{csId}/commands/{cmdId}/status:data` +- `/api/systems` (publish) + - becomes `{nodeId}/systems:data` +- `/api/systems/{sysId}/datastreams` (publish) + - becomes `{nodeId}/systems/{sysId}/datastreams:data` +- `...?f=application/swe+csv` + - becomes `...:data/swe-csv` + +To find `{sysId}` for a datastream or control stream, read `system@id` from the HTTP API, for example `GET /api/datastreams/{dsId}`. +For a subsystem, use the full path: `systems/{sysId}/subsystems/{subId}/...`. + +### Steps + +1. Set `nodeId` in the Connected Systems API MQTT Service config. + - If you do not set it, the value is `api`. Topics then start with `api/` (no leading `/`). + - Use a different `nodeId` for each node on a shared broker. +2. Change each client topic. Use the list above. +3. Remove query strings from topics. Use a format subtopic to select a format. +4. Check JSON commands. Put the fields in `parameters`. +5. For command status, subscribe per command. You get the command ID from the HTTP API. + +### What stops working + +- Topics that start with `/` or with the HTTP endpoint. +- Query strings in topics. Filters such as `?resultTime=` are not available on MQTT. +- Topics without `:data`. In a later version, these topics carry Resource Events. +- The control stream status topic `/api/controlstreams/{csId}/status`. Use the per-command status topic. From 179af26f883913620034c48a839641376262088a Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 9 Nov 2026 16:09:21 -0600 Subject: [PATCH 5/6] refactor(csapi-mqtt): move the Part 3 binding to its own module --- .../org.sensorhub.api.module.IModuleProvider | 1 - .../README.md | 28 +++++++++++-------- .../build.gradle | 10 +++---- .../impl/service/csapi}/mqtt/Activator.java | 2 +- .../csapi}/mqtt/ByteBufferInputStream.java | 2 +- .../csapi/mqtt/CsApiMqttConnector.java} | 6 ++-- .../service/csapi/mqtt/CsApiMqttService.java} | 12 ++++---- .../csapi/mqtt/CsApiMqttServiceConfig.java} | 4 +-- .../mqtt/CsApiMqttServiceDescriptor.java} | 8 +++--- .../impl/service/csapi}/mqtt/TopicMapper.java | 2 +- .../org.sensorhub.api.module.IModuleProvider | 1 + .../service/csapi}/mqtt/TestTopicMapper.java | 2 +- .../src/test/resources/empty | 0 13 files changed, 42 insertions(+), 36 deletions(-) delete mode 100644 services/sensorhub-service-consys-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider rename services/{sensorhub-service-consys-mqtt => sensorhub-service-csapi-mqtt}/README.md (86%) rename services/{sensorhub-service-consys-mqtt => sensorhub-service-csapi-mqtt}/build.gradle (75%) rename services/{sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys => sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi}/mqtt/Activator.java (94%) rename services/{sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys => sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi}/mqtt/ByteBufferInputStream.java (97%) rename services/{sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java => sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttConnector.java} (98%) rename services/{sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java => sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttService.java} (84%) rename services/{sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java => sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceConfig.java} (91%) rename services/{sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceDescriptor.java => sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceDescriptor.java} (83%) rename services/{sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys => sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi}/mqtt/TopicMapper.java (99%) create mode 100644 services/sensorhub-service-csapi-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider rename services/{sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys => sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi}/mqtt/TestTopicMapper.java (99%) rename services/{sensorhub-service-consys-mqtt => sensorhub-service-csapi-mqtt}/src/test/resources/empty (100%) diff --git a/services/sensorhub-service-consys-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider b/services/sensorhub-service-consys-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider deleted file mode 100644 index 7de6a69a8..000000000 --- a/services/sensorhub-service-consys-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider +++ /dev/null @@ -1 +0,0 @@ -org.sensorhub.impl.service.consys.mqtt.ConSysApiMqttServiceDescriptor diff --git a/services/sensorhub-service-consys-mqtt/README.md b/services/sensorhub-service-csapi-mqtt/README.md similarity index 86% rename from services/sensorhub-service-consys-mqtt/README.md rename to services/sensorhub-service-csapi-mqtt/README.md index f3a4ebe01..289219b0a 100644 --- a/services/sensorhub-service-consys-mqtt/README.md +++ b/services/sensorhub-service-csapi-mqtt/README.md @@ -1,9 +1,12 @@ -# Connected Systems API MQTT Service +# Connected Systems API Part 3 MQTT Service This module adds MQTT to the OSH Connected Systems (CS) API. It follows the MQTT binding in OGC CS API Part 3 (draft). The work is not complete. See [Status](#status) and [Known limits](#known-limits). +The legacy MQTT topics stay in the module `sensorhub-service-consys-mqtt`. +You can use the two modules on the same node. See [Legacy module](#legacy-module). + ## What it does - You subscribe to a topic to get live data. Examples are observations and command status. @@ -17,7 +20,7 @@ You need these modules on the node: - Connected Systems API Service - MQTT Server (`sensorhub-service-mqtt-hivemq`) -- Connected Systems API MQTT Service (this module) +- Connected Systems API Part 3 MQTT Service (this module) Configuration: @@ -144,15 +147,18 @@ Not done yet: - 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. -## Change in version 3.0.0 +## Legacy module + +This module was split from `sensorhub-service-consys-mqtt`. That module keeps the legacy topics and does not change. -This is a breaking change. Clients of version 2.x and earlier must change their topics. -The module version was inherited from osh-core before (2.0.1). Now the module has its own version. +- Nodes that use the legacy topics keep `sensorhub-service-consys-mqtt`. Their config does not change. +- To use the Part 3 topics, add this module. Its classes are in `org.sensorhub.impl.service.csapi.mqtt`. +- You can use the two modules on the same node and broker. Legacy topics start with the HTTP endpoint (`/api/`). Part 3 topics start with `{nodeId}/` (`api/`). ### Topics -Old topics used the HTTP endpoint of the CS API service (default `/api`) and the HTTP path. -New topics use `{nodeId}`, the full `{systemPath}` and `:data`. +Legacy topics use the HTTP endpoint of the CS API service (default `/api`) and the HTTP path. +Part 3 topics use `{nodeId}`, the full `{systemPath}` and `:data`. - `/api/datastreams/{dsId}/observations` - becomes `{nodeId}/systems/{sysId}/datastreams/{dsId}/observations:data` @@ -170,9 +176,9 @@ New topics use `{nodeId}`, the full `{systemPath}` and `:data`. To find `{sysId}` for a datastream or control stream, read `system@id` from the HTTP API, for example `GET /api/datastreams/{dsId}`. For a subsystem, use the full path: `systems/{sysId}/subsystems/{subId}/...`. -### Steps +### Move a client to Part 3 topics -1. Set `nodeId` in the Connected Systems API MQTT Service config. +1. Add this module to the node. Set `nodeId` in its config. - If you do not set it, the value is `api`. Topics then start with `api/` (no leading `/`). - Use a different `nodeId` for each node on a shared broker. 2. Change each client topic. Use the list above. @@ -180,9 +186,9 @@ For a subsystem, use the full path: `systems/{sysId}/subsystems/{subId}/...`. 4. Check JSON commands. Put the fields in `parameters`. 5. For command status, subscribe per command. You get the command ID from the HTTP API. -### What stops working +### What this module does not support -- Topics that start with `/` or with the HTTP endpoint. +- Topics that start with `/` or with the HTTP endpoint. Use `sensorhub-service-consys-mqtt` for these. - Query strings in topics. Filters such as `?resultTime=` are not available on MQTT. - Topics without `:data`. In a later version, these topics carry Resource Events. - The control stream status topic `/api/controlstreams/{csId}/status`. Use the per-command status topic. diff --git a/services/sensorhub-service-consys-mqtt/build.gradle b/services/sensorhub-service-csapi-mqtt/build.gradle similarity index 75% rename from services/sensorhub-service-consys-mqtt/build.gradle rename to services/sensorhub-service-csapi-mqtt/build.gradle index 1b68d4f99..aa98e0936 100644 --- a/services/sensorhub-service-consys-mqtt/build.gradle +++ b/services/sensorhub-service-csapi-mqtt/build.gradle @@ -1,6 +1,6 @@ -description = 'Connected Systems API MQTT Extension' -ext.details = 'MQTT Extension for OGC API Connected Systems' -version = '3.0.0' +description = 'Connected Systems API Part 3 MQTT Service' +ext.details = 'MQTT binding of OGC API Connected Systems Part 3' +version = '0.1.0' dependencies { @@ -11,14 +11,14 @@ dependencies { // put README.md next to the config class, for the Admin UI README tab processResources { - from('README.md') { into 'org/sensorhub/impl/service/consys/mqtt' } + from('README.md') { into 'org/sensorhub/impl/service/csapi/mqtt' } } // add info to OSGI manifest osgi { manifest { attributes('Bundle-Vendor': 'Sensia Software LLC') - attributes('Bundle-Activator': 'org.sensorhub.impl.service.consys.mqtt.Activator') + attributes('Bundle-Activator': 'org.sensorhub.impl.service.csapi.mqtt.Activator') } } diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/Activator.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/Activator.java similarity index 94% rename from services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/Activator.java rename to services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/Activator.java index 4fad89510..9f8284484 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/Activator.java +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/Activator.java @@ -12,7 +12,7 @@ ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.service.consys.mqtt; +package org.sensorhub.impl.service.csapi.mqtt; import org.osgi.framework.BundleActivator; import org.sensorhub.utils.OshBundleActivator; diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ByteBufferInputStream.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/ByteBufferInputStream.java similarity index 97% rename from services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ByteBufferInputStream.java rename to services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/ByteBufferInputStream.java index 1980d753e..81b08e4b4 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ByteBufferInputStream.java +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/ByteBufferInputStream.java @@ -12,7 +12,7 @@ ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.service.consys.mqtt; +package org.sensorhub.impl.service.csapi.mqtt; import java.io.IOException; import java.io.InputStream; diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttConnector.java similarity index 98% rename from services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java rename to services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttConnector.java index 4ae185414..25b8729ae 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttConnector.java @@ -12,7 +12,7 @@ ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.service.consys.mqtt; +package org.sensorhub.impl.service.csapi.mqtt; import java.io.IOException; import java.io.OutputStream; @@ -47,7 +47,7 @@ * @author Alex Robin * @since May 9, 2023 */ -public class ConSysApiMqttConnector implements IMqttHandler +public class CsApiMqttConnector implements IMqttHandler { ConSysApiServlet servlet; TopicMapper topics; @@ -115,7 +115,7 @@ public void close() } - public ConSysApiMqttConnector(ConSysApiServlet servlet, TopicMapper topics) + public CsApiMqttConnector(ConSysApiServlet servlet, TopicMapper topics) { this.servlet = servlet; this.topics = topics; diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttService.java similarity index 84% rename from services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java rename to services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttService.java index bbcb4e6c4..57b494351 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttService.java @@ -12,7 +12,7 @@ ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.service.consys.mqtt; +package org.sensorhub.impl.service.csapi.mqtt; import java.util.ArrayList; import java.util.concurrent.TimeUnit; @@ -31,9 +31,9 @@ * @author Alex Robin * @since May 9, 2023 */ -public class ConSysApiMqttService extends AbstractMqttServiceModule implements IServiceModule +public class CsApiMqttService extends AbstractMqttServiceModule implements IServiceModule { - protected ConSysApiMqttConnector mqttConnector; + protected CsApiMqttConnector mqttConnector; @Override @@ -69,9 +69,9 @@ protected void doStart() throws SensorHubException for (var f: service.getConfiguration().customFormats) formats.add(f.mimeType); var topics = new TopicMapper(config.nodeId, formats); - mqttConnector = new ConSysApiMqttConnector(servlet, topics); + mqttConnector = new CsApiMqttConnector(servlet, topics); mqttServer.registerHandler(topics.getPrefix(), mqttConnector); - getLogger().info("CONSYS API MQTT handler registered on '{}', format subtopics {}", topics.getPrefix(), topics.formats.keySet()); + getLogger().info("CS API MQTT handler registered on '{}', format subtopics {}", topics.getPrefix(), topics.formats.keySet()); setState(ModuleState.STARTED); } }) @@ -93,7 +93,7 @@ protected void doStop() throws SensorHubException { // the connector keeps the prefix it was registered with mqttServer.unregisterHandler(mqttConnector.topics.getPrefix(), mqttConnector); - getLogger().info("CONSYS API MQTT handler unregistered"); + getLogger().info("CS API MQTT handler unregistered"); mqttConnector.stop(); mqttConnector = null; } diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceConfig.java similarity index 91% rename from services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java rename to services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceConfig.java index 790d06142..c0b56d120 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceConfig.java @@ -12,7 +12,7 @@ ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.service.consys.mqtt; +package org.sensorhub.impl.service.csapi.mqtt; import org.sensorhub.api.config.DisplayInfo; import org.sensorhub.api.service.ServiceConfig; @@ -26,7 +26,7 @@ * @author Alex Robin * @since May 9, 2023 */ -public class ConSysApiMqttServiceConfig extends ServiceConfig +public class CsApiMqttServiceConfig extends ServiceConfig { @DisplayInfo.Required @DisplayInfo(label="Node ID", desc="Prefix of all MQTT topics, as in '{nodeId}/systems/{id}'. " diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceDescriptor.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceDescriptor.java similarity index 83% rename from services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceDescriptor.java rename to services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceDescriptor.java index c1a3f2855..8e4c2f4f1 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceDescriptor.java +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/CsApiMqttServiceDescriptor.java @@ -12,7 +12,7 @@ ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.service.consys.mqtt; +package org.sensorhub.impl.service.csapi.mqtt; import org.sensorhub.api.module.IModule; import org.sensorhub.api.module.IModuleProvider; @@ -29,19 +29,19 @@ * @author Alex Robin * @since May 9, 2023 */ -public class ConSysApiMqttServiceDescriptor extends JarModuleProvider implements IModuleProvider +public class CsApiMqttServiceDescriptor extends JarModuleProvider implements IModuleProvider { @Override public Class> getModuleClass() { - return ConSysApiMqttService.class; + return CsApiMqttService.class; } @Override public Class getModuleConfigClass() { - return ConSysApiMqttServiceConfig.class; + return CsApiMqttServiceConfig.class; } } diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/TopicMapper.java similarity index 99% rename from services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java rename to services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/TopicMapper.java index 983090758..599ed20b9 100644 --- a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/TopicMapper.java +++ b/services/sensorhub-service-csapi-mqtt/src/main/java/org/sensorhub/impl/service/csapi/mqtt/TopicMapper.java @@ -12,7 +12,7 @@ ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.service.consys.mqtt; +package org.sensorhub.impl.service.csapi.mqtt; import java.net.URI; import java.net.URISyntaxException; 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-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java b/services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestTopicMapper.java similarity index 99% rename from services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java rename to services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestTopicMapper.java index 03a6c401b..e789914cb 100644 --- a/services/sensorhub-service-consys-mqtt/src/test/java/org/sensorhub/impl/service/consys/mqtt/TestTopicMapper.java +++ b/services/sensorhub-service-csapi-mqtt/src/test/java/org/sensorhub/impl/service/csapi/mqtt/TestTopicMapper.java @@ -12,7 +12,7 @@ ******************************* END LICENSE BLOCK ***************************/ -package org.sensorhub.impl.service.consys.mqtt; +package org.sensorhub.impl.service.csapi.mqtt; import static org.junit.Assert.*; import java.util.List; diff --git a/services/sensorhub-service-consys-mqtt/src/test/resources/empty b/services/sensorhub-service-csapi-mqtt/src/test/resources/empty similarity index 100% rename from services/sensorhub-service-consys-mqtt/src/test/resources/empty rename to services/sensorhub-service-csapi-mqtt/src/test/resources/empty From 537a259062b674254057923e294c30330b6274cd Mon Sep 17 00:00:00 2001 From: Ian Patterson Date: Mon, 9 Nov 2026 16:09:21 -0600 Subject: [PATCH 6/6] chore(consys-mqtt): restore the legacy module from master --- .../build.gradle | 29 ++ .../impl/service/consys/mqtt/Activator.java | 23 ++ .../consys/mqtt/ByteBufferInputStream.java | 79 +++++ .../consys/mqtt/ConSysApiMqttConnector.java | 295 ++++++++++++++++++ .../consys/mqtt/ConSysApiMqttService.java | 80 +++++ .../mqtt/ConSysApiMqttServiceConfig.java | 31 ++ .../mqtt/ConSysApiMqttServiceDescriptor.java | 47 +++ .../org.sensorhub.api.module.IModuleProvider | 1 + .../src/test/resources/empty | 0 9 files changed, 585 insertions(+) create mode 100644 services/sensorhub-service-consys-mqtt/build.gradle create mode 100644 services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/Activator.java create mode 100644 services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ByteBufferInputStream.java create mode 100644 services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java create mode 100644 services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java create mode 100644 services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java create mode 100644 services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceDescriptor.java create mode 100644 services/sensorhub-service-consys-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider create mode 100644 services/sensorhub-service-consys-mqtt/src/test/resources/empty diff --git a/services/sensorhub-service-consys-mqtt/build.gradle b/services/sensorhub-service-consys-mqtt/build.gradle new file mode 100644 index 000000000..fe143ab12 --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/build.gradle @@ -0,0 +1,29 @@ +description = 'Connected Systems API MQTT Extension' +ext.details = 'MQTT Extension for OGC API Connected Systems' + + +dependencies { + implementation 'org.sensorhub:sensorhub-core:' + oshCoreVersion + implementation 'org.sensorhub:sensorhub-service-consys:' + oshCoreVersion + implementation project(':sensorhub-comm-mqtt') +} + +// add info to OSGI manifest +osgi { + manifest { + attributes('Bundle-Vendor': 'Sensia Software LLC') + attributes('Bundle-Activator': 'org.sensorhub.impl.service.consys.mqtt.Activator') + } +} + +// add info to maven pom +ext.pom >>= { + developers { + developer { + id 'alexrobin' + name 'Alex Robin' + organization 'Sensia Software LLC' + organizationUrl 'http://www.sensiasoftware.com' + } + } +} diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/Activator.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/Activator.java new file mode 100644 index 000000000..4fad89510 --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/Activator.java @@ -0,0 +1,23 @@ +/***************************** 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) 2012-2015 Sensia Software LLC. All Rights Reserved. + +******************************* END LICENSE BLOCK ***************************/ + +package org.sensorhub.impl.service.consys.mqtt; + +import org.osgi.framework.BundleActivator; +import org.sensorhub.utils.OshBundleActivator; + + +public class Activator extends OshBundleActivator implements BundleActivator +{ +} diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ByteBufferInputStream.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ByteBufferInputStream.java new file mode 100644 index 000000000..1980d753e --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ByteBufferInputStream.java @@ -0,0 +1,79 @@ +/***************************** 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.consys.mqtt; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; + + +public class ByteBufferInputStream extends InputStream +{ + ByteBuffer buf; + int mark = -1; + + + public ByteBufferInputStream(ByteBuffer buf) + { + this.buf = buf; + } + + + public void setBuffer(ByteBuffer buf) + { + this.buf = buf; + this.mark = -1; + } + + + public int read() throws IOException + { + if (!buf.hasRemaining()) + return -1; + + return buf.get() & 0xFF; + } + + + public int read(byte[] bytes, int off, int len) throws IOException + { + if (!buf.hasRemaining()) + { + return -1; + } + + len = Math.min(len, buf.remaining()); + buf.get(bytes, off, len); + return len; + } + + + @Override + public synchronized void mark(int readlimit) + { + this.mark = buf.position(); + } + + + @Override + public synchronized void reset() throws IOException + { + if (mark >= 0) + { + buf.position(mark); + mark = -1; + } + } +} diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java new file mode 100644 index 000000000..d3f47b48b --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttConnector.java @@ -0,0 +1,295 @@ +/***************************** 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.consys.mqtt; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.ByteBuffer; +import java.security.AccessControlException; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.sensorhub.api.comm.mqtt.IMqttServer; +import org.sensorhub.api.comm.mqtt.InvalidPayloadException; +import org.sensorhub.api.comm.mqtt.IMqttServer.IMqttHandler; +import org.sensorhub.api.comm.mqtt.ImplSpecificException; +import org.sensorhub.api.comm.mqtt.InvalidTopicException; +import org.sensorhub.api.comm.mqtt.MqttException; +import org.sensorhub.api.comm.mqtt.MqttOutputStream; +import org.sensorhub.impl.service.consys.InvalidRequestException; +import org.sensorhub.impl.service.consys.InvalidRequestException.ErrorCode; +import org.sensorhub.impl.service.consys.ConSysApiServlet; +import org.sensorhub.impl.service.consys.resource.RequestContext; +import org.sensorhub.impl.service.consys.resource.ResourceFormat; +import org.sensorhub.impl.service.consys.stream.StreamHandler; +import org.vast.util.Asserts; + + +/** + *

+ * This class handles communication with the embedded MQTT server and transfers + * messages to/from the Connected Systems API servlet for processing. + *

+ * + * @author Alex Robin + * @since May 9, 2023 + */ +public class ConSysApiMqttConnector implements IMqttHandler +{ + ConSysApiServlet servlet; + String endpoint; + Map subscribers = new ConcurrentHashMap<>(); + + + class MqttSubscriber implements StreamHandler + { + IMqttServer server; + String topic; + MqttOutputStream os; + Runnable onStart, onClose; + AtomicInteger numSubscribers = new AtomicInteger(0); + AtomicBoolean started = new AtomicBoolean(); + + MqttSubscriber(IMqttServer server, String topic) + { + this.server = server; + this.topic = topic; + this.os = new MqttOutputStream(server, topic, 1024, false); + } + + @Override + public void sendPacket() throws IOException + { + os.send(); + } + + @Override + public void sendPacket(long correlId) throws IOException + { + os.send(correlId); + } + + @Override + public OutputStream getOutputStream() + { + return os; + } + + @Override + public void setStartCallback(Runnable onStart) + { + this.onStart = Asserts.checkNotNull(onStart, "onStart"); + } + + @Override + public void setCloseCallback(Runnable onClose) + { + this.onClose = Asserts.checkNotNull(onClose, "onClose"); + } + + public void maybeStart() + { + if (onStart != null && started.compareAndSet(false, true)) + onStart.run(); + } + + @Override + public void close() + { + if (onClose != null) + onClose.run(); + } + } + + + public ConSysApiMqttConnector(ConSysApiServlet servlet, String endpoint) + { + this.servlet = servlet; + this.endpoint = endpoint; + } + + + @Override + public void onSubscribe(String userID, String topic, IMqttServer server) throws InvalidTopicException + { + try + { + // register new subscription if needed + var sub = subscribers.compute(topic, (k, v) -> { + // create subscriber if needed + if (v == null) + v = new MqttSubscriber(server, topic); + return v; + }); + + // always evaluate request because we need to check for permissions + // even if subscriber was already created + var ctx = getResourceContext(topic, sub); + servlet.getSecurityHandler().setCurrentUser(userID); + servlet.getRootHandler().doGet(ctx); + + // start stream if all went well + sub.numSubscribers.incrementAndGet(); + sub.maybeStart(); + } + catch (SecurityException | InvalidTopicException e) + { + throw e; + } + catch (InvalidRequestException e) + { + throw new InvalidTopicException(e.getMessage()); + } + catch (Exception e) + { + throw new IllegalStateException("Internal error subscribing to topic " + topic, e); + } + finally + { + servlet.getSecurityHandler().clearCurrentUser(); + } + } + + + @Override + public void onUnsubscribe(String userID, String topic, IMqttServer server) throws InvalidTopicException + { + subscribers.computeIfPresent(topic, (k, sub) -> { + var numSub = sub.numSubscribers.decrementAndGet(); + if (numSub <= 0) + { + servlet.getLogger().debug("No more clients listening on topic {}. Cancelling eventbus subscription.", topic); + sub.close(); + sub = null; + } + else + servlet.getLogger().debug("{} client(s) still listening on topic {}", numSub, topic); + return sub; + }); + } + + + @Override + public void onPublish(String userID, String topic, ByteBuffer payload, ByteBuffer correlData) throws MqttException + { + try + { + var ctx = getResourceContext(topic, payload); + ctx.setRequestContentType(ResourceFormat.AUTO.getMimeType()); + + if (correlData != null) + { + long cmdID = 0; + if (correlData.remaining() == 8) + cmdID = correlData.getLong(); + if (correlData.remaining() == 4) + cmdID = correlData.getInt(); + if (cmdID == 0) + throw new InvalidRequestException(ErrorCode.BAD_PAYLOAD, "Invalid correlation data"); + //var idStr = StandardCharsets.UTF_8.decode(correlData).toString(); + //var cmdID = Long.parseLong(idStr); + ctx.setCorrelationID(cmdID); + } + + servlet.getSecurityHandler().setCurrentUser(userID); + servlet.getRootHandler().doPost(ctx); + } + catch (SecurityException | InvalidTopicException e) + { + throw e; + } + catch (InvalidRequestException e) + { + handleInvalidRequestException(e); + } + catch (Exception e) + { + throw new IllegalStateException("Internal error publishing to topic " + topic, e); + } + finally + { + servlet.getSecurityHandler().clearCurrentUser(); + } + } + + + private RequestContext getResourceContext(String topic, MqttSubscriber streamHandler) throws InvalidTopicException + { + return new RequestContext( + servlet, + getResourceUri(topic), + streamHandler); + } + + + private RequestContext getResourceContext(String topic, ByteBuffer payload) throws InvalidTopicException + { + return new RequestContext( + servlet, + getResourceUri(topic), + new ByteBufferInputStream(payload)); + } + + + private URI getResourceUri(String topic) throws InvalidTopicException + { + try + { + // remove the base URL part + topic = topic.replaceFirst(endpoint, ""); + + // parse URI (this also URL decodes the query string) + return new URI(topic); + } + catch (URISyntaxException e) + { + throw new InvalidTopicException("Invalid SWE API resource URI"); + } + } + + + private void handleInvalidRequestException(InvalidRequestException e) throws MqttException, SecurityException + { + switch (e.getErrorCode()) + { + case NOT_FOUND: + case BAD_REQUEST: + throw new InvalidTopicException(e.getMessage()); + + case BAD_PAYLOAD: + throw new InvalidPayloadException(e.getMessage()); + + case REQUEST_REJECTED: + throw new ImplSpecificException(e.getMessage()); + + case FORBIDDEN: + throw new AccessControlException("Forbidden"); + + default: + throw new IllegalStateException("Internal error", e); + } + } + + + protected void stop() + { + for (var sub: subscribers.values()) + sub.close(); + subscribers.clear(); + } +} diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java new file mode 100644 index 000000000..a88873082 --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttService.java @@ -0,0 +1,80 @@ +/***************************** 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.consys.mqtt; + +import java.util.concurrent.TimeUnit; +import org.sensorhub.api.common.SensorHubException; +import org.sensorhub.api.module.ModuleEvent.ModuleState; +import org.sensorhub.api.service.IServiceModule; +import org.sensorhub.impl.service.consys.ConSysApiService; +import org.sensorhub.impl.service.mqtt.AbstractMqttServiceModule; + + +/** + *

+ * Add-on service to {@link ConSysApiService} to enable MQTT support + *

+ * + * @author Alex Robin + * @since May 9, 2023 + */ +public class ConSysApiMqttService extends AbstractMqttServiceModule implements IServiceModule +{ + protected ConSysApiMqttConnector mqttConnector; + protected String endPoint; + + + @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(); + endPoint = service.getConfiguration().endPoint; + + if (mqttServer != null) + { + mqttConnector = new ConSysApiMqttConnector(servlet, endPoint); + mqttServer.registerHandler(endPoint, mqttConnector); + getLogger().info("CONSYS API MQTT handler registered"); + 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) + { + mqttServer.unregisterHandler(endPoint, mqttConnector); + getLogger().info("CONSYS API MQTT handler unregistered"); + mqttConnector.stop(); + mqttConnector = null; + } + } +} diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java new file mode 100644 index 000000000..4cd1d5512 --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceConfig.java @@ -0,0 +1,31 @@ +/***************************** 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.consys.mqtt; + +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 ConSysApiMqttServiceConfig extends ServiceConfig +{ + +} diff --git a/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceDescriptor.java b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceDescriptor.java new file mode 100644 index 000000000..c1a3f2855 --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/src/main/java/org/sensorhub/impl/service/consys/mqtt/ConSysApiMqttServiceDescriptor.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.consys.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 ConSysApiMqttServiceDescriptor extends JarModuleProvider implements IModuleProvider +{ + @Override + public Class> getModuleClass() + { + return ConSysApiMqttService.class; + } + + + @Override + public Class getModuleConfigClass() + { + return ConSysApiMqttServiceConfig.class; + } + +} diff --git a/services/sensorhub-service-consys-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider b/services/sensorhub-service-consys-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider new file mode 100644 index 000000000..7de6a69a8 --- /dev/null +++ b/services/sensorhub-service-consys-mqtt/src/main/resources/META-INF/services/org.sensorhub.api.module.IModuleProvider @@ -0,0 +1 @@ +org.sensorhub.impl.service.consys.mqtt.ConSysApiMqttServiceDescriptor diff --git a/services/sensorhub-service-consys-mqtt/src/test/resources/empty b/services/sensorhub-service-consys-mqtt/src/test/resources/empty new file mode 100644 index 000000000..e69de29bb