Skip to content

Commit 2e9b45f

Browse files
aaronziaaronzi
committed
Fix MQTT eventing and broker tests
Unify MQTT topics, upgrade Moquette, and guard Docker publishing.
1 parent cd90e46 commit 2e9b45f

57 files changed

Lines changed: 1234 additions & 1161 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.github/workflows/docker-milestone-release.yml‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,32 @@ on:
77
env:
88
DOCKER_NAMESPACE: eclipsebasyx
99

10+
permissions: {}
11+
1012
jobs:
13+
guard:
14+
runs-on: ubuntu-latest
15+
outputs:
16+
should-run: ${{ steps.check.outputs.should-run }}
17+
steps:
18+
- id: check
19+
env:
20+
EVENT_REPOSITORY_FORK: ${{ github.event.repository.fork }}
21+
run: |
22+
if [[ "${GITHUB_REPOSITORY}" == "eclipse-basyx/basyx-java-server-sdk" && "${EVENT_REPOSITORY_FORK}" != "true" ]]; then
23+
echo "Running Docker publication in the upstream repository"
24+
echo "should-run=true" >> "$GITHUB_OUTPUT"
25+
else
26+
echo "Skipping Docker publication for fork ${GITHUB_REPOSITORY}"
27+
echo "should-run=false" >> "$GITHUB_OUTPUT"
28+
fi
29+
1130
build-and-push-milestone-release:
31+
needs: guard
32+
if: ${{ needs.guard.outputs.should-run == 'true' }}
1233
runs-on: ubuntu-latest
34+
permissions:
35+
contents: read
1336
strategy:
1437
matrix:
1538
include:

‎.github/workflows/docker-snapshot-release.yml‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,32 @@ on:
2323
env:
2424
DOCKER_NAMESPACE: eclipsebasyx
2525

26+
permissions: {}
27+
2628
jobs:
29+
guard:
30+
runs-on: ubuntu-latest
31+
outputs:
32+
should-run: ${{ steps.check.outputs.should-run }}
33+
steps:
34+
- id: check
35+
env:
36+
EVENT_REPOSITORY_FORK: ${{ github.event.repository.fork }}
37+
run: |
38+
if [[ "${GITHUB_REPOSITORY}" == "eclipse-basyx/basyx-java-server-sdk" && "${EVENT_REPOSITORY_FORK}" != "true" ]]; then
39+
echo "Running Docker publication in the upstream repository"
40+
echo "should-run=true" >> "$GITHUB_OUTPUT"
41+
else
42+
echo "Skipping Docker publication for fork ${GITHUB_REPOSITORY}"
43+
echo "should-run=false" >> "$GITHUB_OUTPUT"
44+
fi
45+
2746
build-and-push-prerelease:
47+
needs: guard
48+
if: ${{ needs.guard.outputs.should-run == 'true' }}
2849
runs-on: ubuntu-latest
50+
permissions:
51+
contents: read
2952
strategy:
3053
matrix:
3154
include:

‎basyx.aasrepository/basyx.aasrepository-feature-mqtt/Readme.md‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,5 +6,9 @@ This feature provides hierarchical MQTT eventing for a multitude of events:
66
| AAS Created | aas-repository/\$repoId/shells/created| Created AAS JSON |
77
| AAS Updated | aas-repository/\$repoId/shells/updated| Updated AAS JSON|
88
| AAS Deleted | aas-repository/\$repoId/shells/deleted| Deleted AAS JSON|
9-
| Submodel Reference Created | aas-repository/\$repoId/shells/submodels/\$submodelId/created| Created AAS JSON |
10-
| Submodel Reference Deleted | aas-repository/\$repoId/shells/submodels/\$submodelId/deleted| Deleted AAS JSON|
9+
| Submodel Reference Created | aas-repository/\$repoId/shells/submodels/\$submodelIdBase64Url/created| Created AAS JSON |
10+
| Submodel Reference Deleted | aas-repository/\$repoId/shells/submodels/\$submodelIdBase64Url/deleted| Deleted AAS JSON|
11+
12+
Identifier segments use unpadded UTF-8 Base64URL encoding. Repository names are not encoded.
13+
14+
> **Breaking topic migration:** Submodel-reference events are published only on the encoded canonical topics. Existing subscribers using raw submodel IDs must update their subscriptions.

‎basyx.aasrepository/basyx.aasrepository-feature-mqtt/pom.xml‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,13 @@
2121
<groupId>org.eclipse.digitaltwin.basyx</groupId>
2222
<artifactId>basyx.mqttcore</artifactId>
2323
</dependency>
24+
<dependency>
25+
<groupId>org.eclipse.digitaltwin.basyx</groupId>
26+
<artifactId>basyx.mqttcore</artifactId>
27+
<version>${revision}</version>
28+
<classifier>tests</classifier>
29+
<scope>test</scope>
30+
</dependency>
2431
<dependency>
2532
<groupId>org.eclipse.digitaltwin.basyx</groupId>
2633
<artifactId>basyx.aasrepository-core</artifactId>
@@ -55,4 +62,4 @@
5562
<scope>test</scope>
5663
</dependency>
5764
</dependencies>
58-
</project>
65+
</project>

‎basyx.aasrepository/basyx.aasrepository-feature-mqtt/src/main/java/org/eclipse/digitaltwin/basyx/aasrepository/feature/mqtt/MqttAasRepository.java‎

Lines changed: 2 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -35,14 +35,12 @@
3535
import org.eclipse.digitaltwin.aas4j.v3.model.Reference;
3636
import org.eclipse.digitaltwin.aas4j.v3.model.SpecificAssetId;
3737
import org.eclipse.digitaltwin.basyx.aasrepository.AasRepository;
38+
import org.eclipse.digitaltwin.basyx.common.mqttcore.MqttEventPublisher;
3839
import org.eclipse.digitaltwin.basyx.core.exceptions.CollidingIdentifierException;
3940
import org.eclipse.digitaltwin.basyx.core.exceptions.ElementDoesNotExistException;
4041
import org.eclipse.digitaltwin.basyx.core.pagination.CursorResult;
4142
import org.eclipse.digitaltwin.basyx.core.pagination.PaginationInfo;
4243
import org.eclipse.paho.client.mqttv3.IMqttClient;
43-
import org.eclipse.paho.client.mqttv3.MqttException;
44-
import org.eclipse.paho.client.mqttv3.MqttMessage;
45-
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
4644
import org.slf4j.Logger;
4745
import org.slf4j.LoggerFactory;
4846

@@ -170,24 +168,7 @@ private String serializePayload(AssetAdministrationShell shell) {
170168
* the actual message
171169
*/
172170
private void sendMqttMessage(String topic, String payload) {
173-
MqttMessage msg = createMqttMessage(payload);
174-
175-
try {
176-
logger.debug("Send MQTT message to " + topic + ": " + payload);
177-
mqttClient.publish(topic, msg);
178-
} catch (MqttPersistenceException e) {
179-
logger.error("Could not persist mqtt message", e);
180-
} catch (MqttException e) {
181-
logger.error("Could not send mqtt message", e);
182-
}
183-
}
184-
185-
private MqttMessage createMqttMessage(String payload) {
186-
if (payload == null) {
187-
return new MqttMessage();
188-
} else {
189-
return new MqttMessage(payload.getBytes());
190-
}
171+
MqttEventPublisher.publish(mqttClient, topic, payload, logger);
191172
}
192173

193174
@Override

‎basyx.aasrepository/basyx.aasrepository-feature-mqtt/src/main/java/org/eclipse/digitaltwin/basyx/aasrepository/feature/mqtt/MqttAasRepositoryConfiguration.java‎

Lines changed: 2 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -26,54 +26,11 @@
2626

2727
package org.eclipse.digitaltwin.basyx.aasrepository.feature.mqtt;
2828

29-
import org.eclipse.paho.client.mqttv3.IMqttClient;
30-
import org.eclipse.paho.client.mqttv3.MqttClient;
31-
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
32-
import org.eclipse.paho.client.mqttv3.MqttException;
33-
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
34-
import org.springframework.beans.factory.annotation.Value;
29+
import org.eclipse.digitaltwin.basyx.common.mqttcore.MqttClientConfiguration;
3530
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
36-
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
37-
import org.springframework.boot.context.properties.ConfigurationProperties;
38-
import org.springframework.context.annotation.Bean;
3931
import org.springframework.context.annotation.Configuration;
4032

4133
@ConditionalOnExpression("#{${" + MqttAasRepositoryFeature.FEATURENAME + ".enabled:false} or ${basyx.feature.mqtt.enabled:false}}")
4234
@Configuration
43-
public class MqttAasRepositoryConfiguration {
44-
45-
@ConditionalOnMissingBean
46-
@Bean
47-
public IMqttClient mqttClient(
48-
@Value("${mqtt.clientId}") String clientId,
49-
@Value("${mqtt.hostname}") String hostname,
50-
@Value("${mqtt.port}") int port,
51-
@Value("${mqtt.protocol:tcp}") String protocol,
52-
MqttConnectOptions mqttConnectOptions)
53-
throws MqttException {
54-
IMqttClient mqttClient = new MqttClient(protocol + "://" + hostname + ":" + port, clientId,
55-
new MemoryPersistence());
56-
57-
mqttClient.connect(mqttConnectOptions);
58-
59-
return mqttClient;
60-
}
61-
62-
@ConditionalOnMissingBean
63-
@Bean
64-
@ConfigurationProperties(prefix = "mqtt")
65-
public MqttConnectOptions mqttConnectOptions(
66-
@Value("${mqtt.username:}") String username,
67-
@Value("${mqtt.password:}") String password) {
68-
MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
69-
mqttConnectOptions.setAutomaticReconnect(true);
70-
71-
if (username.isBlank() || password.isBlank())
72-
return mqttConnectOptions;
73-
74-
mqttConnectOptions.setUserName(username);
75-
mqttConnectOptions.setPassword(password.toCharArray());
76-
77-
return mqttConnectOptions;
78-
}
35+
public class MqttAasRepositoryConfiguration extends MqttClientConfiguration {
7936
}

‎basyx.aasrepository/basyx.aasrepository-feature-mqtt/src/main/java/org/eclipse/digitaltwin/basyx/aasrepository/feature/mqtt/MqttAasRepositoryFeature.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828

2929
import org.eclipse.digitaltwin.basyx.aasrepository.AasRepositoryFactory;
3030
import org.eclipse.digitaltwin.basyx.aasrepository.feature.AasRepositoryFeature;
31-
import org.eclipse.digitaltwin.basyx.common.mqttcore.encoding.URLEncoder;
31+
import org.eclipse.digitaltwin.basyx.common.mqttcore.encoding.Base64URLEncoder;
3232
import org.eclipse.paho.client.mqttv3.IMqttClient;
3333
import org.springframework.beans.factory.annotation.Autowired;
3434
import org.springframework.beans.factory.annotation.Value;
@@ -52,7 +52,7 @@ public MqttAasRepositoryFeature(IMqttClient mqttClient) {
5252

5353
@Override
5454
public AasRepositoryFactory decorate(AasRepositoryFactory aasServiceFactory) {
55-
return new MqttAasRepositoryFactory(aasServiceFactory, mqttClient, new MqttAasRepositoryTopicFactory(new URLEncoder()));
55+
return new MqttAasRepositoryFactory(aasServiceFactory, mqttClient, new MqttAasRepositoryTopicFactory(new Base64URLEncoder()));
5656
}
5757

5858
@Override

‎basyx.aasrepository/basyx.aasrepository-feature-mqtt/src/main/java/org/eclipse/digitaltwin/basyx/aasrepository/feature/mqtt/MqttAasRepositoryTopicFactory.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,7 @@ public String createCreateAASSubmodelReferenceTopic(String repoId, String refere
108108
.add(repoId)
109109
.add(SHELLS)
110110
.add(SUBMODELS)
111-
.add(referenceId)
111+
.add(encodeId(referenceId))
112112
.add(CREATED)
113113
.toString();
114114
}
@@ -126,7 +126,7 @@ public String createDeleteAASSubmodelReferenceTopic(String repoId, String refere
126126
.add(repoId)
127127
.add(SHELLS)
128128
.add(SUBMODELS)
129-
.add(referenceId)
129+
.add(encodeId(referenceId))
130130
.add(DELETED)
131131
.toString();
132132
}

‎basyx.aasrepository/basyx.aasrepository-feature-mqtt/src/test/java/org/eclipse/digitaltwin/basyx/aasrepository/feature/mqtt/MqttAasRepositoryTopicFactoryTest.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
/*******************************************************************************
2+
* Copyright (C) 2026 the Eclipse BaSyx Authors
3+
*
4+
* Permission is hereby granted, free of charge, to any person obtaining
5+
* a copy of this software and associated documentation files (the
6+
* "Software"), to deal in the Software without restriction, including
7+
* without limitation the rights to use, copy, modify, merge, publish,
8+
* distribute, sublicense, and/or sell copies of the Software, and to
9+
* permit persons to whom the Software is furnished to do so, subject to
10+
* the following conditions:
11+
*
12+
* The above copyright notice and this permission notice shall be
13+
* included in all copies or substantial portions of the Software.
14+
*
15+
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
16+
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
17+
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
18+
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
19+
* LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
20+
* OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
21+
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
22+
*
23+
* SPDX-License-Identifier: MIT
24+
******************************************************************************/
25+
package org.eclipse.digitaltwin.basyx.aasrepository.feature.mqtt;
26+
27+
import static org.junit.Assert.assertEquals;
28+
29+
import org.eclipse.digitaltwin.basyx.common.mqttcore.encoding.Base64URLEncoder;
30+
import org.junit.Test;
31+
32+
public class MqttAasRepositoryTopicFactoryTest {
33+
private static final String SUBMODEL_ID = "https://example.org/submodels/ä?x=1";
34+
private static final String ENCODED_SUBMODEL_ID = "aHR0cHM6Ly9leGFtcGxlLm9yZy9zdWJtb2RlbHMvw6Q_eD0x";
35+
36+
private final MqttAasRepositoryTopicFactory topicFactory = new MqttAasRepositoryTopicFactory(new Base64URLEncoder());
37+
38+
@Test
39+
public void encodesSubmodelReferenceIdentifiers() {
40+
assertEquals("aas-repository/custom-repo/shells/submodels/" + ENCODED_SUBMODEL_ID + "/created",
41+
topicFactory.createCreateAASSubmodelReferenceTopic("custom-repo", SUBMODEL_ID));
42+
assertEquals("aas-repository/custom-repo/shells/submodels/" + ENCODED_SUBMODEL_ID + "/deleted",
43+
topicFactory.createDeleteAASSubmodelReferenceTopic("custom-repo", SUBMODEL_ID));
44+
}
45+
}

0 commit comments

Comments
 (0)