You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2026/01/29 13:53:48 UTC
(streampipes) branch 4137-updating-mqtt-adapter-can-leave-orphaned-consumer-thread updated (b829126092 -> 85f5e5e39e)
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a change to branch 4137-updating-mqtt-adapter-can-leave-orphaned-consumer-thread
in repository https://gitbox.apache.org/repos/asf/streampipes.git
from b829126092 fix(#4133): Add originType for add timestamp rule migration (#4134)
new 0fb0f273b2 fix(#4137): Improve logic to get single event from mqtt
new bbc91d1da8 fix(#4137): Remove unused method
new 85f5e5e39e fix(#4137): Update MqttConsumer implementation
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../connect/iiot/adapters/oi4/Oi4Adapter.java | 167 +++++--------------
.../connectors/mqtt/adapter/MqttProtocol.java | 98 +++++-------
.../connectors/mqtt/shared/MqttConsumer.java | 177 ++++++++++++---------
.../connectors/mqtt/shared/MqttPublisher.java | 8 +-
.../mqtt/shared/MqttSingleMessageReceiver.java | 129 +++++++++++++++
.../integration/adapters/MQTTPublisherUtils.java | 5 -
6 files changed, 314 insertions(+), 270 deletions(-)
create mode 100644 streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttSingleMessageReceiver.java
(streampipes) 02/03: fix(#4137): Remove unused method
Posted by ze...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch 4137-updating-mqtt-adapter-can-leave-orphaned-consumer-thread
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit bbc91d1da861f402669a82cc4ea3741aa3cbc3c3
Author: Philipp Zehnder <te...@users.noreply.github.com>
AuthorDate: Thu Jan 29 13:58:14 2026 +0100
fix(#4137): Remove unused method
---
.../apache/streampipes/integration/adapters/MQTTPublisherUtils.java | 5 -----
1 file changed, 5 deletions(-)
diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/MQTTPublisherUtils.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/MQTTPublisherUtils.java
index 87ce58e814..c273982f09 100644
--- a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/MQTTPublisherUtils.java
+++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/adapters/MQTTPublisherUtils.java
@@ -57,9 +57,4 @@ public class MQTTPublisherUtils {
publisher.connect();
return publisher;
}
-
- public static void closeConnection(MqttPublisher publisher) {
- publisher.disconnect();
- }
-
}
(streampipes) 01/03: fix(#4137): Improve logic to get single event from mqtt
Posted by ze...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch 4137-updating-mqtt-adapter-can-leave-orphaned-consumer-thread
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 0fb0f273b220c9328726fb91f2b01c780a0f8086
Author: Philipp Zehnder <te...@users.noreply.github.com>
AuthorDate: Thu Jan 29 13:56:37 2026 +0100
fix(#4137): Improve logic to get single event from mqtt
---
.../connect/iiot/adapters/oi4/Oi4Adapter.java | 160 +++++----------------
.../connectors/mqtt/adapter/MqttProtocol.java | 92 +++++-------
.../connectors/mqtt/shared/MqttConsumer.java | 24 +---
.../connectors/mqtt/shared/MqttPublisher.java | 8 +-
.../mqtt/shared/MqttSingleMessageReceiver.java | 129 +++++++++++++++++
5 files changed, 208 insertions(+), 205 deletions(-)
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
index 06e0c76b84..b724b43f58 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
@@ -32,13 +32,11 @@ import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
import org.apache.streampipes.extensions.connectors.mqtt.shared.MqttConfig;
import org.apache.streampipes.extensions.connectors.mqtt.shared.MqttConnectUtils;
import org.apache.streampipes.extensions.connectors.mqtt.shared.MqttConsumer;
+import org.apache.streampipes.extensions.connectors.mqtt.shared.MqttSingleMessageReceiver;
import org.apache.streampipes.extensions.management.connect.adapter.parser.JsonParsers;
import org.apache.streampipes.extensions.management.connect.adapter.parser.json.JsonObjectParser;
-import org.apache.streampipes.messaging.InternalEventProcessor;
-import org.apache.streampipes.model.connect.guess.GuessSchema;
import org.apache.streampipes.model.connect.guess.SampleData;
import org.apache.streampipes.model.extensions.ExtensionAssetType;
-import org.apache.streampipes.model.schema.EventSchema;
import org.apache.streampipes.sdk.StaticProperties;
import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder;
import org.apache.streampipes.sdk.helpers.Alternatives;
@@ -62,12 +60,8 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
-import static org.apache.streampipes.sdk.helpers.EpProperties.timestampProperty;
-
/**
* Adapter to connect to an Open Industry 4.0 (OI4) compatible device.
*
@@ -79,8 +73,6 @@ public class Oi4Adapter implements StreamPipesAdapter {
public static final String ID = "org.apache.streampipes.connect.iiot.adapters.oi4";
private static final Logger LOG = LoggerFactory.getLogger(Oi4Adapter.class);
- private static final long ReceiveSchemaSleepTime = 100;
- private static final long ReceiveSchemaMaxTimeout = 5000;
// Information about the topic structure can be found at page 57 of the above-mentioned development guide
// The app id (missing here) needs to be provided by the user
@@ -95,7 +87,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
public Oi4Adapter() {
mapper = JacksonSerializer.getObjectMapper(Map.of(
- DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, true
+ DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, true
));
}
@@ -107,7 +99,8 @@ public class Oi4Adapter implements StreamPipesAdapter {
.withLocales(Locales.EN)
.withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
.requiredTextParameter(MqttConnectUtils.getBrokerUrlLabel())
- .requiredAlternatives(MqttConnectUtils.getAccessModeLabel(), MqttConnectUtils.getAnonymousAccess(),
+ .requiredAlternatives(
+ MqttConnectUtils.getAccessModeLabel(), MqttConnectUtils.getAnonymousAccess(),
MqttConnectUtils.getUsernameAccess(), MqttConnectUtils.getClientCertAccess()
)
.requiredAlternatives(
@@ -144,7 +137,10 @@ public class Oi4Adapter implements StreamPipesAdapter {
IAdapterRuntimeContext adapterRuntimeContext
) throws AdapterException {
LOG.info("Adapter type {} starting", ID);
- LOG.info("Adapter with id {} starting", extractor.getAdapterDescription().getElementId());
+ LOG.info("Adapter with id {} starting",
+ extractor.getAdapterDescription()
+ .getElementId()
+ );
this.applyConfiguration(extractor.getStaticPropertyExtractor());
@@ -168,7 +164,10 @@ public class Oi4Adapter implements StreamPipesAdapter {
thread.start();
LOG.info("Adapter {} started", ID);
- LOG.info("Adapter with id {} started", extractor.getAdapterDescription().getElementId());
+ LOG.info("Adapter with id {} started",
+ extractor.getAdapterDescription()
+ .getElementId()
+ );
}
private InputStream convertByte(byte[] event) {
@@ -191,7 +190,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
} else {
var selectedSensorsText = extractor.textParameter(OI4AdapterLabels.LABEL_KEY_SENSORS_LIST_INPUT);
selectedSensors = Arrays.stream(selectedSensorsText.split(","))
- .toList();
+ .toList();
}
if (selectedAlternativeSensorDescription.equals(OI4AdapterLabels.LABEL_KEY_SENSOR_TYPE_ALTERNATIVE)) {
@@ -215,48 +214,12 @@ public class Oi4Adapter implements StreamPipesAdapter {
IAdapterParameterExtractor extractor,
IAdapterGuessSchemaContext adapterGuessSchemaContext
) throws AdapterException {
- try {
- this.applyConfiguration(extractor.getStaticPropertyExtractor());
-
- var sampleMessage = getSampleMessage();
- var sampleData = transformToSampleData(sampleMessage);
-
- return sampleData;
- } catch (RuntimeException e) {
- throw new AdapterException(e.getMessage(), e);
- }
- }
-
- /**
- * Updates the timestamp property in the given GuessSchema if it exists as it is not correctly guessed.
- * If the timestamp property exists, it is replaced with a proper timestamp property.
- *
- * @param guessSchema The GuessSchema to update.
- */
- private void updateTimestampPropertyIfExists(GuessSchema guessSchema) {
- var eventProperties = guessSchema.getEventSchema()
- .getEventProperties();
-
- var timestampPropertyOpt = eventProperties.stream()
- .filter(eventProperty ->
- eventProperty.getRuntimeName()
- .equals(OI4AdapterLabels.EVENT_KEY_TIMESTAMP)
- )
- .findFirst();
-
- var newTimestampProperty = timestampProperty(OI4AdapterLabels.EVENT_KEY_TIMESTAMP);
-
- // If the timestamp property exists, replace it with the new timestamp property
- timestampPropertyOpt.ifPresent(prop -> {
- eventProperties.removeIf(eventProperty -> eventProperty.getRuntimeName()
- .equals(OI4AdapterLabels.EVENT_KEY_TIMESTAMP));
- eventProperties.add(newTimestampProperty);
- });
-
- guessSchema.setEventSchema(new EventSchema(eventProperties));
+ this.applyConfiguration(extractor.getStaticPropertyExtractor());
+ var sampleMessage = getSampleMessage();
+ return transformToSampleData(sampleMessage);
}
- private SampleData transformToSampleData(byte[] sampleMessage) {
+ private SampleData transformToSampleData(byte[] sampleMessage) throws AdapterException {
try {
var networkMessage = mapper.readValue(sampleMessage, NetworkMessage.class);
var payload = extractPayload(networkMessage);
@@ -265,68 +228,14 @@ public class Oi4Adapter implements StreamPipesAdapter {
return new JsonParsers(new JsonObjectParser())
.getSampleData(convertByte(plainPayload.getBytes(StandardCharsets.UTF_8)));
} catch (IOException e) {
- LOG.error("Error while reading sample message: {}", sampleMessage);
- throw new RuntimeException(e);
+ throw new AdapterException("Error while reading sample message", e);
}
}
- private byte[] getSampleMessage() throws AdapterException {
- List<byte[]> sampleMessages = new ArrayList<>();
- long timeElapsed = 0;
- AtomicReference<Throwable> exceptionRef = new AtomicReference<>();
- MqttConsumer guessConsumer = getGuessMqttConsumer(sampleMessages);
-
- var thread = new Thread(guessConsumer);
- thread.setUncaughtExceptionHandler((t, e) -> exceptionRef.set(e.getCause()));
- thread.start();
-
- while (sampleMessages.isEmpty() && exceptionRef.get() == null && timeElapsed < ReceiveSchemaMaxTimeout) {
- try {
- TimeUnit.MILLISECONDS.sleep(ReceiveSchemaSleepTime);
- timeElapsed += ReceiveSchemaSleepTime;
- } catch (InterruptedException e) {
- LOG.error("Schema guessing failed during waiting for an incoming event: {}", e.getMessage());
- break;
- }
- }
- guessConsumer.close();
-
- Throwable threadException = exceptionRef.get();
- if (threadException != null) {
- throw new AdapterException(threadException.getMessage(), threadException);
- }
-
- if (!sampleMessages.isEmpty()) {
- return sampleMessages.get(0);
- } else {
- throw new AdapterException("No messages received");
- }
- }
- /**
- * Obtain a specialized MQTT consumer designed to infer the event schema based on provided sampleMessages.
- *
- * @param sampleMessages A list of byte arrays representing MQTT message payloads.
- * @return A customized MqttConsumer instance.
- */
- private MqttConsumer getGuessMqttConsumer(List<byte[]> sampleMessages) {
- // Define a specialized event processor that adds an event to the sampleMessages array
- // only if it meets certain expectations, as verified by extractPayload.
- InternalEventProcessor<byte[]> eventProcessor = event -> {
- InputStream in = convertByte(event);
- NetworkMessage networkMessage;
- try {
- networkMessage = mapper.readValue(in, NetworkMessage.class);
- } catch (IOException e) {
- LOG.error("Error during parsing of incoming MQTT event.");
- throw new RuntimeException(e);
- }
- // Attempt to extract payload from the NetworkMessage
- extractPayload(networkMessage);
- // If successful, add the event to the sampleMessages array
- sampleMessages.add(event);
- };
- return new MqttConsumer(this.mqttConfig, eventProcessor);
+ private byte[] getSampleMessage() throws AdapterException {
+ var receiver = new MqttSingleMessageReceiver(this.mqttConfig, 10);
+ return receiver.receiveSingleMessage();
}
private List<Map<String, Object>> extractPayload(NetworkMessage message) throws ParseException {
@@ -343,7 +252,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
// Verify that the message corresponds to the designated sensor type.
// This validation relies on the assumption that the source information includes the sensor type.
if (dataMessage.source()
- .contains(givenSensorType)) {
+ .contains(givenSensorType)) {
// an empty list of selected sensors means that we want to collect data from all sensors available
if (selectedSensors.isEmpty() || selectedSensors.contains(sensorId)) {
@@ -361,10 +270,10 @@ public class Oi4Adapter implements StreamPipesAdapter {
private List<DataSetMessage> findProcessDataInputMessage(NetworkMessage message) {
return message.messages()
- .stream()
- .filter(msg -> msg.filter()
- .equals(OI4AdapterLabels.MESSAGE_VALUE_FILTER))
- .toList();
+ .stream()
+ .filter(msg -> msg.filter()
+ .equals(OI4AdapterLabels.MESSAGE_VALUE_FILTER))
+ .toList();
}
private Map<String, Object> extractAndEnrichMessagePayload(DataSetMessage dataSetMessage, String sensorId) {
@@ -388,16 +297,19 @@ public class Oi4Adapter implements StreamPipesAdapter {
private Map<String, Object> replaceSpecialChars(Map<String, Object> originalMap) {
return originalMap.entrySet()
- .stream()
- .collect(Collectors.toMap(
- entry -> entry.getKey().replace("-", ""),
- Map.Entry::getValue,
- (oldValue, newValue) -> oldValue)
- );
+ .stream()
+ .collect(Collectors.toMap(
+ entry -> entry.getKey()
+ .replace("-", ""),
+ Map.Entry::getValue,
+ (oldValue, newValue) -> oldValue
+ )
+ );
}
private long parseDate(String timestamp) throws DateTimeParseException {
- return Instant.parse(timestamp).toEpochMilli();
+ return Instant.parse(timestamp)
+ .toEpochMilli();
}
/**
diff --git a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java
index d3816c4f44..97959774c3 100644
--- a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java
+++ b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java
@@ -28,27 +28,18 @@ import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
import org.apache.streampipes.extensions.connectors.mqtt.shared.MqttConfig;
import org.apache.streampipes.extensions.connectors.mqtt.shared.MqttConnectUtils;
import org.apache.streampipes.extensions.connectors.mqtt.shared.MqttConsumer;
+import org.apache.streampipes.extensions.connectors.mqtt.shared.MqttSingleMessageReceiver;
import org.apache.streampipes.extensions.management.connect.adapter.BrokerEventProcessor;
import org.apache.streampipes.extensions.management.connect.adapter.parser.Parsers;
-import org.apache.streampipes.messaging.InternalEventProcessor;
import org.apache.streampipes.model.connect.guess.SampleData;
import org.apache.streampipes.model.extensions.ExtensionAssetType;
import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder;
import org.apache.streampipes.sdk.helpers.Locales;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.io.ByteArrayInputStream;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
public class MqttProtocol implements StreamPipesAdapter {
- private static final Logger LOG = LoggerFactory.getLogger(MqttProtocol.class);
-
public static final String ID = "org.apache.streampipes.connect.iiot.protocol.stream.mqtt";
private MqttConsumer mqttConsumer;
@@ -57,9 +48,6 @@ public class MqttProtocol implements StreamPipesAdapter {
public MqttProtocol() {
}
- public void applyConfiguration(IStaticPropertyExtractor extractor) {
- this.mqttConfig = MqttConnectUtils.getMqttConfig(extractor);
- }
@Override
public IAdapterConfiguration declareConfig() {
@@ -69,18 +57,22 @@ public class MqttProtocol implements StreamPipesAdapter {
.withLocales(Locales.EN)
.withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
.requiredTextParameter(MqttConnectUtils.getBrokerUrlLabel())
- .requiredAlternatives(MqttConnectUtils.getAccessModeLabel(), MqttConnectUtils.getAnonymousAccess(),
- MqttConnectUtils.getUsernameAccess(), MqttConnectUtils.getClientCertAccess())
+ .requiredAlternatives(
+ MqttConnectUtils.getAccessModeLabel(), MqttConnectUtils.getAnonymousAccess(),
+ MqttConnectUtils.getUsernameAccess(), MqttConnectUtils.getClientCertAccess()
+ )
.requiredTextParameter(MqttConnectUtils.getTopicLabel())
.buildConfiguration();
}
@Override
- public void onAdapterStarted(IAdapterParameterExtractor extractor,
- IEventCollector collector,
- IAdapterRuntimeContext adapterRuntimeContext) throws AdapterException {
+ public void onAdapterStarted(
+ IAdapterParameterExtractor extractor,
+ IEventCollector collector,
+ IAdapterRuntimeContext adapterRuntimeContext
+ ) throws AdapterException {
- this.applyConfiguration(extractor.getStaticPropertyExtractor());
+ this.initializeMqttConfig(extractor.getStaticPropertyExtractor());
this.mqttConsumer = new MqttConsumer(
this.mqttConfig,
new BrokerEventProcessor(extractor.selectedParser(), collector)
@@ -91,44 +83,36 @@ public class MqttProtocol implements StreamPipesAdapter {
}
@Override
- public void onAdapterStopped(IAdapterParameterExtractor extractor,
- IAdapterRuntimeContext adapterRuntimeContext) throws AdapterException {
+ public void onAdapterStopped(
+ IAdapterParameterExtractor extractor,
+ IAdapterRuntimeContext adapterRuntimeContext
+ ) {
this.mqttConsumer.close();
}
@Override
- public SampleData onSampleDataRequested(IAdapterParameterExtractor extractor,
- IAdapterGuessSchemaContext adapterGuessSchemaContext) throws AdapterException {
- try {
- AtomicReference<Throwable> exceptionRef = new AtomicReference<>();
- this.applyConfiguration(extractor.getStaticPropertyExtractor());
- List<byte[]> elements = new ArrayList<>();
- InternalEventProcessor<byte[]> eventProcessor = elements::add;
-
-
- MqttConsumer consumer = new MqttConsumer(this.mqttConfig, eventProcessor);
-
- Thread thread = new Thread(consumer);
- thread.setUncaughtExceptionHandler((t, e) -> exceptionRef.set(e.getCause()));
- thread.start();
-
- while (consumer.getMessageCount() < 1 && exceptionRef.get() == null) {
- try {
- TimeUnit.MILLISECONDS.sleep(100);
- } catch (InterruptedException e) {
- break;
- }
- }
- consumer.close();
-
- Throwable threadException = exceptionRef.get();
- if (threadException != null) {
- throw new AdapterException(threadException.getMessage(), threadException);
- }
-
- return extractor.selectedParser().getSampleData(new ByteArrayInputStream(elements.get(0)));
- } catch (Exception e) {
- throw new AdapterException(e.getMessage(), e);
- }
+ public SampleData onSampleDataRequested(
+ IAdapterParameterExtractor extractor,
+ IAdapterGuessSchemaContext adapterGuessSchemaContext
+ ) throws AdapterException {
+
+ this.initializeMqttConfig(extractor.getStaticPropertyExtractor());
+
+ var payload = getSampleEventAsByte();
+
+ return extractor.selectedParser()
+ .getSampleData(new ByteArrayInputStream(payload));
+
}
+
+ private byte[] getSampleEventAsByte() throws AdapterException {
+ var receiver = new MqttSingleMessageReceiver(this.mqttConfig, 10);
+ return receiver.receiveSingleMessage();
+ }
+
+ public void initializeMqttConfig(IStaticPropertyExtractor extractor) {
+ this.mqttConfig = MqttConnectUtils.getMqttConfig(extractor);
+ }
+
}
+
diff --git a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java
index 24b0a6cb93..fd9f960f06 100644
--- a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java
+++ b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java
@@ -30,25 +30,18 @@ import java.util.concurrent.ExecutionException;
public class MqttConsumer extends MqttBase implements Runnable {
+ private static final Logger LOG = LoggerFactory.getLogger(MqttConsumer.class);
+
private final InternalEventProcessor<byte[]> consumer;
private boolean running;
- private int maxElementsToReceive = -1;
- private int messageCount = 0;
private Mqtt3AsyncClient client;
- private static final Logger LOG = LoggerFactory.getLogger(MqttConsumer.class);
-
public MqttConsumer(MqttConfig mqttConfig, InternalEventProcessor<byte[]> consumer) {
super(mqttConfig);
this.consumer = consumer;
}
- public MqttConsumer(MqttConfig mqttConfig, InternalEventProcessor<byte[]> consumer, int maxElementsToReceive) {
- this(mqttConfig, consumer);
- this.maxElementsToReceive = maxElementsToReceive;
- }
-
@Override
public void run() {
this.running = true;
@@ -103,13 +96,6 @@ public class MqttConsumer extends MqttBase implements Runnable {
try {
byte[] payload = publish.getPayloadAsBytes();
consumer.onEvent(payload);
- messageCount++;
-
- if (maxElementsToReceive != -1 && messageCount >= maxElementsToReceive) {
- LOG.info("Max elements ({}) received. Stopping consumer.", maxElementsToReceive);
- this.running = false;
- }
-
} catch (Exception e) {
LOG.error("Error processing MQTT message", e);
}
@@ -128,8 +114,4 @@ public class MqttConsumer extends MqttBase implements Runnable {
}
}
-
- public Integer getMessageCount() {
- return messageCount;
- }
-}
\ No newline at end of file
+}
diff --git a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttPublisher.java b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttPublisher.java
index 002b5fe54d..d11cad2e4a 100644
--- a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttPublisher.java
+++ b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttPublisher.java
@@ -28,12 +28,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.net.URI;
-
-
public class MqttPublisher extends MqttBase {
- private URI uri;
private Mqtt3AsyncClient client;
private static final Logger LOG = LoggerFactory.getLogger(MqttPublisher.class);
@@ -89,7 +85,7 @@ public class MqttPublisher extends MqttBase {
});
} catch (Exception e) {
throw new SpRuntimeException("Could not publish to MQTT broker: "
- + uri.toString() + ", " + e.getMessage(), e);
+ + super.mqttConfig.getUrl() + ", " + e.getMessage(), e);
}
}
@@ -99,7 +95,7 @@ public class MqttPublisher extends MqttBase {
.get(); // block until disconnected
} catch (Exception e) {
throw new SpRuntimeException("Could not disconnect from MQTT broker: "
- + uri.toString() + ", " + e.getMessage(), e);
+ + super.mqttConfig.getUrl() + ", " + e.getMessage(), e);
}
}
diff --git a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttSingleMessageReceiver.java b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttSingleMessageReceiver.java
new file mode 100644
index 0000000000..f00810056f
--- /dev/null
+++ b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttSingleMessageReceiver.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.streampipes.extensions.connectors.mqtt.shared;
+
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
+
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class MqttSingleMessageReceiver extends MqttBase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MqttSingleMessageReceiver.class);
+ private final int timeoutInSeconds;
+
+ public MqttSingleMessageReceiver(MqttConfig mqttConfig, int timeoutInSeconds) {
+ super(mqttConfig);
+ this.timeoutInSeconds = timeoutInSeconds;
+ }
+
+ /**
+ * Receives a single MQTT message and blocks until a message arrives or the timeout elapses.
+ *
+ * @return the payload bytes of the first received message
+ * @throws AdapterException when receiving fails or the thread is interrupted
+ */
+ public byte[] receiveSingleMessage() throws AdapterException {
+ try {
+ return receiveSingleMessageAsync().get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AdapterException("Interrupted while waiting for an MQTT message", e);
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof TimeoutException) {
+ throw new AdapterException(
+ String.format("Timed out after %d seconds while waiting for an MQTT message", timeoutInSeconds),
+ e.getCause()
+ );
+ }
+ throw new AdapterException("Failed to receive MQTT message", e.getCause());
+ } catch (Exception e) {
+ throw new AdapterException("Failed to receive MQTT message", e);
+ }
+ }
+
+ private CompletableFuture<byte[]> receiveSingleMessageAsync() {
+ var result = new CompletableFuture<byte[]>();
+ var client = setupClientOrFail(result);
+ if (client == null) {
+ return result;
+ }
+ registerCleanup(result, client);
+ subscribeForSingleMessage(result, client);
+ applyTimeout(result);
+ return result;
+ }
+
+ private Mqtt3AsyncClient setupClientOrFail(CompletableFuture<byte[]> result) {
+ try {
+ return super.setupMqttClient();
+ } catch (Exception e) {
+ result.completeExceptionally(e);
+ return null;
+ }
+ }
+
+ private void registerCleanup(CompletableFuture<byte[]> result, Mqtt3AsyncClient client) {
+ result.whenComplete((payload, throwable) -> disconnectQuietly(client));
+ }
+
+ private void subscribeForSingleMessage(CompletableFuture<byte[]> result, Mqtt3AsyncClient client) {
+ client.connect()
+ .thenCompose(v -> client.subscribeWith()
+ .topicFilter(mqttConfig.getTopic())
+ .qos(MqttQos.AT_LEAST_ONCE)
+ .callback(publish -> completeIfEmpty(result, publish.getPayloadAsBytes()))
+ .send()
+ )
+ .exceptionally(ex -> {
+ result.completeExceptionally(ex);
+ return null;
+ });
+ }
+
+ private void completeIfEmpty(CompletableFuture<byte[]> result, byte[] payload) {
+ if (!result.isDone()) {
+ result.complete(payload);
+ }
+ }
+
+ private void applyTimeout(CompletableFuture<byte[]> result) {
+ if (this.timeoutInSeconds > 0) {
+ result.orTimeout(this.timeoutInSeconds, TimeUnit.SECONDS);
+ }
+ }
+
+ private void disconnectQuietly(Mqtt3AsyncClient client) {
+ try {
+ client.disconnect().whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ LOG.warn("Failed to disconnect MQTT client after receiving sample", throwable);
+ }
+ });
+ } catch (Exception e) {
+ LOG.warn("Failed to disconnect MQTT client after receiving sample", e);
+ }
+ }
+}
(streampipes) 03/03: fix(#4137): Update MqttConsumer implementation
Posted by ze...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch 4137-updating-mqtt-adapter-can-leave-orphaned-consumer-thread
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 85f5e5e39eaa215dc9fd194aecccbfb18200c382
Author: Philipp Zehnder <te...@users.noreply.github.com>
AuthorDate: Thu Jan 29 14:50:55 2026 +0100
fix(#4137): Update MqttConsumer implementation
---
.../connect/iiot/adapters/oi4/Oi4Adapter.java | 7 +-
.../connectors/mqtt/adapter/MqttProtocol.java | 8 +-
.../connectors/mqtt/shared/MqttConsumer.java | 159 +++++++++++++--------
3 files changed, 110 insertions(+), 64 deletions(-)
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
index b724b43f58..0b280437bd 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
@@ -160,8 +160,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
}
);
- Thread thread = new Thread(this.mqttConsumer);
- thread.start();
+ this.mqttConsumer.start();
LOG.info("Adapter {} started", ID);
LOG.info("Adapter with id {} started",
@@ -206,7 +205,9 @@ public class Oi4Adapter implements StreamPipesAdapter {
IAdapterParameterExtractor extractor,
IAdapterRuntimeContext adapterRuntimeContext
) {
- mqttConsumer.close();
+ if (mqttConsumer != null) {
+ mqttConsumer.stop();
+ }
}
@Override
diff --git a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java
index 97959774c3..36dbf81669 100644
--- a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java
+++ b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/adapter/MqttProtocol.java
@@ -78,8 +78,7 @@ public class MqttProtocol implements StreamPipesAdapter {
new BrokerEventProcessor(extractor.selectedParser(), collector)
);
- Thread thread = new Thread(this.mqttConsumer);
- thread.start();
+ this.mqttConsumer.start();
}
@Override
@@ -87,7 +86,9 @@ public class MqttProtocol implements StreamPipesAdapter {
IAdapterParameterExtractor extractor,
IAdapterRuntimeContext adapterRuntimeContext
) {
- this.mqttConsumer.close();
+ if (this.mqttConsumer != null) {
+ this.mqttConsumer.stop();
+ }
}
@Override
@@ -115,4 +116,3 @@ public class MqttProtocol implements StreamPipesAdapter {
}
}
-
diff --git a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java
index fd9f960f06..1ed4aa2b34 100644
--- a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java
+++ b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/shared/MqttConsumer.java
@@ -17,101 +17,146 @@
*/
package org.apache.streampipes.extensions.connectors.mqtt.shared;
+import org.apache.streampipes.commons.exceptions.connect.AdapterException;
import org.apache.streampipes.messaging.InternalEventProcessor;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
+import com.hivemq.client.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAck;
import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
-public class MqttConsumer extends MqttBase implements Runnable {
+public class MqttConsumer extends MqttBase {
private static final Logger LOG = LoggerFactory.getLogger(MqttConsumer.class);
+ private static final int KEEP_ALIVE_SECONDS = 30;
- private final InternalEventProcessor<byte[]> consumer;
- private boolean running;
-
+ private final InternalEventProcessor<byte[]> eventProcessor;
+ private final AtomicBoolean running;
private Mqtt3AsyncClient client;
public MqttConsumer(MqttConfig mqttConfig, InternalEventProcessor<byte[]> consumer) {
super(mqttConfig);
- this.consumer = consumer;
+ this.eventProcessor = consumer;
+ this.running = new AtomicBoolean(false);
}
- @Override
- public void run() {
- this.running = true;
+ /**
+ * Starts the MQTT consumer and subscribes to the configured topic.
+ *
+ * @throws AdapterException when the connection or subscription fails
+ */
+ public void start() throws AdapterException {
+ if (!this.running.compareAndSet(false, true)) {
+ return;
+ }
try {
- this.client = super.setupMqttClient();
- client.connectWith()
- .keepAlive(30)
- .send()
- .whenComplete((cAck, throwable) -> {
- if (throwable != null) {
- LOG.error("MQTT connection failed", throwable);
- } else {
- LOG.info("MQTT connection established");
- }
- })
- .get();
-
- subscribe(client);
-
- } catch (Exception e) {
- LOG.error("Error in MQTT consumer: ", e);
- throw new RuntimeException("Error when receiving data from MQTT", e);
+ this.client = connectClient();
+ subscribeToTopic(this.client);
+ } catch (InterruptedException e) {
+ this.running.set(false);
+ Thread.currentThread().interrupt();
+ throw new AdapterException("Interrupted while starting MQTT consumer", e);
+ } catch (AdapterException e) {
+ this.running.set(false);
+ throw e;
}
}
- private void subscribe(Mqtt3AsyncClient client) throws Exception {
-
- CountDownLatch subscribed = new CountDownLatch(1);
-
- client.subscribeWith()
- .topicFilter(super.mqttConfig.getTopic())
- .qos(MqttQos.AT_LEAST_ONCE)
- .callback(this::handleMessage)
- .send()
- .whenComplete((subAck, throwable) -> {
- if (throwable != null) {
- LOG.error("MQTT subscribe failed", throwable);
- } else {
- LOG.info("Successfully subscribed to topic {}", super.mqttConfig.getTopic());
- }
- subscribed.countDown();
- });
-
- subscribed.await();
+ /**
+ * Stops the MQTT consumer and disconnects from the broker.
+ */
+ public void stop() {
+ this.running.set(false);
+ if (this.client == null) {
+ return;
+ }
+ disconnectSafely();
}
private void handleMessage(Mqtt3Publish publish) {
- if (!this.running) {
+ if (!this.running.get()) {
return;
}
-
try {
- byte[] payload = publish.getPayloadAsBytes();
- consumer.onEvent(payload);
- } catch (Exception e) {
+ processPayload(publish.getPayloadAsBytes());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn("Interrupted while processing MQTT message", e);
+ } catch (RuntimeException e) {
LOG.error("Error processing MQTT message", e);
}
}
- public void close() {
- this.running = false;
+
+ private Mqtt3AsyncClient connectClient() throws AdapterException, InterruptedException {
try {
+ var mqttClient = super.setupMqttClient();
+ mqttClient.connectWith()
+ .keepAlive(KEEP_ALIVE_SECONDS)
+ .send()
+ .whenComplete(this::logConnect)
+ .get();
+ return mqttClient;
+ } catch (InterruptedException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new AdapterException("Error while connecting to MQTT broker", e);
+ }
+ }
- this.client.disconnect().get();
+ private void logConnect(Mqtt3ConnAck cAck, Throwable throwable) {
+ if (throwable != null) {
+ LOG.error("MQTT connection failed", throwable);
+ } else {
+ LOG.info("MQTT connection established");
+ }
+ }
+ private void subscribeToTopic(Mqtt3AsyncClient client) throws AdapterException, InterruptedException {
+ CountDownLatch subscribed = new CountDownLatch(1);
+ client.subscribeWith()
+ .topicFilter(super.mqttConfig.getTopic())
+ .qos(MqttQos.AT_LEAST_ONCE)
+ .callback(this::handleMessage)
+ .send()
+ .whenComplete((subAck, throwable) -> {
+ logSubscribe(throwable);
+ subscribed.countDown();
+ });
+ try {
+ subscribed.await();
} catch (InterruptedException e) {
- LOG.error("Error disconnecting from MQTT due to thread interruption", e);
- } catch (ExecutionException e) {
- LOG.error("Error disconnecting from MQTT", e.getCause());
+ Thread.currentThread().interrupt();
+ throw e;
}
+ }
+
+ private void logSubscribe(Throwable throwable) {
+ if (throwable != null) {
+ LOG.error("MQTT subscribe failed", throwable);
+ } else {
+ LOG.info("Successfully subscribed to topic {}", super.mqttConfig.getTopic());
+ }
+ }
+ private void processPayload(byte[] payload) throws InterruptedException{
+ eventProcessor.onEvent(payload);
+ }
+
+ private void disconnectSafely() {
+ try {
+ this.client.disconnect().whenComplete((result, throwable) -> {
+ if (throwable != null) {
+ LOG.error("Error disconnecting from MQTT", throwable);
+ }
+ });
+ } catch (RuntimeException e) {
+ LOG.error("Error disconnecting from MQTT", e);
+ }
}
}