feat(mq): add activemq adapter passing the contract suite

Phase 3 (1/4): ActiveMQ (Artemis / Classic, JMS 2.0) adapter for
dc3.mq.type=activemq, TCK 11 pass + 1 documented skip.

- live topics map to JMS topics: LOAD_BALANCE rides a shared durable
  subscription named after the consumer group (competing consumers share
  it, offline messages retained), BROADCAST rides a plain per-instance
  consumer — one publish fans out to every subscription like an exchange
- dead-letter destinations are queues; reject(false) republishes to
  dc3.<topic>.dlq and acknowledges, reject(true) recovers the session
- delays use JMS scheduled delivery natively (capability true); batches
  are synthesized by draining the consumer with the shared synchronous
  bounded-retry semantics; exhaustion dead-letters instead of dropping
- JMS property names must be java identifiers, so the dashed standard
  headers ride underscored and are restored on read
- best-effort publisher confirmation (capability false): the synchronous
  persistent send returning is the reported signal
- tck harness against apache/activemq-artemis, or TCK_ARTEMIS_URL for an
  externally managed broker; subscriptionExpiry documented false

dc3-common/dc3-common-mq-activemq/pom.xml

@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2016-present the IoT DC3 original author or authors.
+  ~
+  ~ This program is free software: you can redistribute it and/or modify
+  ~ it under the terms of the GNU Affero General Public License as
+  ~ published by the Free Software Foundation, either version 3 of the
+  ~ License, or (at your option) any later version.
+  ~
+  ~ This program is distributed in the hope that it will be useful,
+  ~ but WITHOUT ANY WARRANTY; without even the implied warranty of
+  ~ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  ~ GNU Affero General Public License for more details.
+  ~
+  ~ You should have received a copy of the GNU Affero General Public License
+  ~ along with this program.  If not, see <https://www.gnu.org/licenses/>.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.github.pnoker</groupId>
+        <artifactId>dc3-common</artifactId>
+        <version>2026.5.22</version>
+    </parent>
+
+    <name>${project.artifactId}</name>
+    <artifactId>dc3-common-mq-activemq</artifactId>
+    <version>2026.5.22</version>
+    <packaging>jar</packaging>
+
+    <description>IoT DC3 ActiveMQ (Artemis / Classic, JMS 2.0) adapter for the broker-neutral messaging port</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.activemq</groupId>
+            <artifactId>artemis-jakarta-client</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.github.pnoker</groupId>
+            <artifactId>dc3-common-mq</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.github.pnoker</groupId>
+            <artifactId>dc3-common-constant</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

dc3-common/dc3-common-mq-activemq/src/main/java/io/github/pnoker/common/mq/activemq/ActiveMqAdapter.java

@@ -0,0 +1,456 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.mq.activemq;
+
+import io.github.pnoker.common.constant.mq.MqTopic;
+import io.github.pnoker.common.constant.mq.OrderingGuarantee;
+import io.github.pnoker.common.constant.mq.SubscriptionMode;
+import io.github.pnoker.common.mq.adapter.BrokerAdapter;
+import io.github.pnoker.common.mq.adapter.BrokerCapabilities;
+import io.github.pnoker.common.mq.adapter.RawBatchListener;
+import io.github.pnoker.common.mq.adapter.RawDeliveryListener;
+import io.github.pnoker.common.mq.adapter.WireConfirmation;
+import io.github.pnoker.common.mq.adapter.WireMqDelivery;
+import io.github.pnoker.common.mq.config.BatchConsumerProperties;
+import io.github.pnoker.common.mq.listener.Acknowledgment;
+import io.github.pnoker.common.mq.listener.MqPoisonException;
+import io.github.pnoker.common.mq.message.WireMqMessage;
+import io.github.pnoker.common.mq.subscription.SubscriptionSpec;
+import jakarta.jms.BytesMessage;
+import jakarta.jms.Connection;
+import jakarta.jms.ConnectionFactory;
+import jakarta.jms.DeliveryMode;
+import jakarta.jms.Destination;
+import jakarta.jms.JMSContext;
+import jakarta.jms.JMSException;
+import jakarta.jms.JMSProducer;
+import jakarta.jms.Message;
+import jakarta.jms.MessageConsumer;
+import jakarta.jms.Session;
+import jakarta.jms.Topic;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * ActiveMQ (Artemis / Classic) implementation of the broker port over JMS 2.0.
+ *
+ * <p>Live topics map to JMS topics: LOAD_BALANCE rides a <b>shared durable
+ * subscription</b> named after the consumer group (competing consumers share it,
+ * offline messages are retained), BROADCAST rides a plain topic consumer per
+ * instance — one publish fans out to every subscription, matching the rabbit
+ * exchange semantics. Dead-letter destinations are queues. Delays use JMS scheduled
+ * delivery natively; rejecting without requeue republishes to the
+ * {@code dc3.<topic>.dlq} queue; batches are synthesized by draining the consumer
+ * within a short window (capability false). JMS has no publisher confirmation — the
+ * synchronous persistent send returning is what the adapter reports (best-effort).
+ *
+ * <p>JMS property names must be valid java identifiers, so the dashed standard
+ * headers ({@code dc3-type}, {@code X-Request-Id}, {@code dc3-correlation-id}) ride
+ * the wire underscored and are restored on read.
+ *
+ * @author pnoker
+ * @since 2026.8.19
+ */
+@Slf4j
+public class ActiveMqAdapter implements BrokerAdapter {
+
+    /**
+     * Default shared-durable subscription name when the spec carries no group.
+     */
+    private static final String DEFAULT_SUBSCRIPTION = "dc3-mq";
+
+    private final ConnectionFactory connectionFactory;
+    private final BatchConsumerProperties retryProperties;
+
+    private final JMSContext publishContext;
+    private final List<Connection> subscriptions = new CopyOnWriteArrayList<>();
+    private final List<BatchPump> pumps = new CopyOnWriteArrayList<>();
+
+    public ActiveMqAdapter(ConnectionFactory connectionFactory, BatchConsumerProperties retryProperties) {
+        this.connectionFactory = connectionFactory;
+        this.retryProperties = retryProperties;
+        this.publishContext = connectionFactory.createContext(JMSContext.CLIENT_ACKNOWLEDGE);
+    }
+
+    @Override
+    public String type() {
+        return "activemq";
+    }
+
+    @Override
+    public BrokerCapabilities capabilities() {
+        return new BrokerCapabilities(true, false, true, true, false, false, false, OrderingGuarantee.NONE);
+    }
+
+    @Override
+    public void publish(WireMqMessage message) {
+        publish(message, null);
+    }
+
+    @Override
+    public void publish(WireMqMessage message, WireConfirmation confirmation) {
+        try {
+            synchronized (publishContext) {
+                JMSProducer producer = publishContext.createProducer();
+                producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+                if (!message.delay().isZero()) {
+                    producer.setDeliveryDelay(message.delay().toMillis());
+                }
+                producer.send(destinationOf(message.topic()), jmsMessage(publishContext, message));
+            }
+            if (Objects.nonNull(confirmation)) {
+                confirmation.onConfirm(message, true, null);
+            }
+        } catch (JMSException e) {
+            if (Objects.nonNull(confirmation)) {
+                confirmation.onConfirm(message, false, e);
+                return;
+            }
+            throw new IllegalStateException("ActiveMQ publish failed, topic=" + message.topic(), e);
+        }
+    }
+
+    @Override
+    public void subscribe(SubscriptionSpec spec, RawDeliveryListener listener) {
+        try {
+            Connection connection = connectionFactory.createConnection();
+            subscriptions.add(connection);
+            Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+            MessageConsumer consumer = consumerOf(spec, session);
+            String destinationLabel = destinationName(spec.topic()) + subscriptionSuffix(spec);
+            consumer.setMessageListener(message -> {
+                Acknowledgment ack = new ActiveMqAcknowledgment(message, session, spec.topic(), this);
+                try {
+                    listener.onDelivery(deliveryOf(message, ack));
+                } catch (MqPoisonException e) {
+                    deadLetter(message, spec.topic());
+                } catch (Exception e) {
+                    log.warn("ActiveMQ delivery failed, recovering session for redelivery, destination={}",
+                            destinationLabel, e);
+                    recover(session);
+                }
+            });
+            connection.start();
+            log.info("ActiveMQ subscription started, topic={}, mode={}, destination={}",
+                    spec.topic(), spec.mode(), destinationLabel);
+        } catch (JMSException e) {
+            throw new IllegalStateException("ActiveMQ subscribe failed, topic=" + spec.topic(), e);
+        }
+    }
+
+    @Override
+    public void subscribeBatch(SubscriptionSpec spec, RawBatchListener listener) {
+        try {
+            Connection connection = connectionFactory.createConnection();
+            subscriptions.add(connection);
+            Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+            MessageConsumer consumer = consumerOf(spec, session);
+            BatchPump pump = new BatchPump(spec, consumer, session, listener);
+            pumps.add(pump);
+            Thread thread = new Thread(pump, "dc3-mq-activemq-batch-" + spec.topic());
+            thread.start();
+            connection.start();
+            log.info("ActiveMQ batch subscription started, topic={}, destination={}",
+                    spec.topic(), destinationName(spec.topic()) + subscriptionSuffix(spec));
+        } catch (JMSException e) {
+            throw new IllegalStateException("ActiveMQ subscribeBatch failed, topic=" + spec.topic(), e);
+        }
+    }
+
+    /**
+     * Stop every subscription connection and batch pump this adapter started. Shared
+     * durable subscriptions survive offline (that is their point) — messages published
+     * while down are retained for the next subscriber.
+     */
+    public void stop() {
+        pumps.forEach(BatchPump::halt);
+        subscriptions.forEach(connection -> {
+            try {
+                connection.close();
+            } catch (JMSException e) {
+                log.debug("ActiveMQ connection close failed", e);
+            }
+        });
+        subscriptions.clear();
+    }
+
+    private void recover(Session session) {
+        try {
+            session.recover();
+        } catch (JMSException e) {
+            log.warn("ActiveMQ session recover failed", e);
+        }
+    }
+
+    private void deadLetter(Message message, MqTopic topic) {
+        try {
+            synchronized (publishContext) {
+                JMSProducer producer = publishContext.createProducer();
+                producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+                producer.send(publishContext.createQueue(deadLetterQueue(topic)), message);
+            }
+            message.acknowledge();
+        } catch (JMSException e) {
+            log.warn("ActiveMQ dead-letter publish failed, topic={}", topic, e);
+        }
+    }
+
+    /**
+     * Synthesized batch consumer: block for the first message, drain up to batchSize
+     * within the receive window, then deliver with the same synchronous bounded-retry
+     * semantics the kafka adapter uses; exhaustion dead-letters and acknowledges.
+     */
+    private final class BatchPump implements Runnable {
+
+        private final SubscriptionSpec spec;
+        private final MessageConsumer consumer;
+        private final Session session;
+        private final RawBatchListener listener;
+        private volatile boolean halted;
+
+        private BatchPump(SubscriptionSpec spec, MessageConsumer consumer, Session session,
+                          RawBatchListener listener) {
+            this.spec = spec;
+            this.consumer = consumer;
+            this.session = session;
+            this.listener = listener;
+        }
+
+        void halt() {
+            halted = true;
+        }
+
+        @Override
+        public void run() {
+            while (!halted) {
+                try {
+                    Message first = consumer.receive(200);
+                    if (Objects.isNull(first)) {
+                        continue;
+                    }
+                    List<Message> batch = new ArrayList<>(List.of(first));
+                    long deadline = System.currentTimeMillis() + retryProperties.getReceiveTimeoutMillis();
+                    while (batch.size() < retryProperties.getBatchSize()
+                            && System.currentTimeMillis() < deadline) {
+                        Message next = consumer.receive(10);
+                        if (Objects.isNull(next)) {
+                            break;
+                        }
+                        batch.add(next);
+                    }
+                    deliverWithRetry(batch);
+                } catch (JMSException e) {
+                    if (!halted) {
+                        log.warn("ActiveMQ batch pump receive failed, topic={}", spec.topic(), e);
+                    }
+                }
+            }
+        }
+
+        private void deliverWithRetry(List<Message> batch) {
+            Message last = batch.get(batch.size() - 1);
+            Acknowledgment ack = new ActiveMqAcknowledgment(last, session, spec.topic(), ActiveMqAdapter.this);
+            int maxAttempts = Math.max(1, retryProperties.getMaxRetries()) + 1;
+            for (int attempt = 1; ; attempt++) {
+                try {
+                    List<WireMqDelivery> deliveries = new ArrayList<>(batch.size());
+                    for (Message message : batch) {
+                        deliveries.add(deliveryOf(message, ack));
+                    }
+                    listener.onBatch(deliveries);
+                    return;
+                } catch (MqPoisonException e) {
+                    log.warn("ActiveMQ poison batch dead-lettered, size={}", batch.size(), e);
+                    batch.forEach(message -> deadLetter(message, spec.topic()));
+                    return;
+                } catch (Exception e) {
+                    if (attempt >= maxAttempts) {
+                        log.error("ActiveMQ batch exhausted retries, dead-lettering, size={}", batch.size(), e);
+                        batch.forEach(message -> deadLetter(message, spec.topic()));
+                        return;
+                    }
+                    sleepBackoff(attempt);
+                }
+            }
+        }
+
+        private void sleepBackoff(int attempt) {
+            long initial = retryProperties.getRetryInitialIntervalMillis();
+            long cap = retryProperties.getRetryMaxIntervalMillis();
+            long exponent = Math.min(Math.max(attempt - 1, 0), 30);
+            long delay = initial >= cap ? cap : Math.min(initial * (1L << exponent), cap);
+            try {
+                Thread.sleep(delay);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+
+    /**
+     * ack acknowledges the session's consumed messages (batch-granular by design);
+     * reject(true) recovers the session for redelivery; reject(false) dead-letters the
+     * delivery's topic then acknowledges everything consumed.
+     */
+    private record ActiveMqAcknowledgment(Message message, Session session, MqTopic topic,
+                                          ActiveMqAdapter adapter) implements Acknowledgment {
+
+        @Override
+        public void ack() {
+            try {
+                message.acknowledge();
+            } catch (JMSException e) {
+                log.warn("ActiveMQ acknowledge failed", e);
+            }
+        }
+
+        @Override
+        public void reject(boolean requeue) {
+            if (requeue) {
+                adapter.recover(session);
+                return;
+            }
+            adapter.deadLetter(message, topic);
+        }
+    }
+
+    /**
+     * Consumer for a subscription: dead-letter destinations are queues; live topics are
+     * JMS topics where LOAD_BALANCE rides a shared durable subscription named after the
+     * consumer group and BROADCAST rides a plain per-instance consumer.
+     */
+    private static MessageConsumer consumerOf(SubscriptionSpec spec, Session session) throws JMSException {
+        if (isDeadLetterTopic(spec.topic())) {
+            return session.createConsumer(session.createQueue(destinationName(spec.topic())));
+        }
+        Topic topic = session.createTopic(destinationName(spec.topic()));
+        if (spec.mode() == SubscriptionMode.BROADCAST) {
+            return session.createConsumer(topic);
+        }
+        String subscription = spec.group().isBlank() ? DEFAULT_SUBSCRIPTION : spec.group();
+        return session.createSharedDurableConsumer(topic, subscription);
+    }
+
+    private static String subscriptionSuffix(SubscriptionSpec spec) {
+        if (spec.mode() != SubscriptionMode.LOAD_BALANCE || spec.group().isBlank()) {
+            return "";
+        }
+        return " (" + spec.group() + ")";
+    }
+
+    private WireMqDelivery deliveryOf(Message message, Acknowledgment acknowledgment) {
+        return new WireMqDelivery(bodyOf(message), headersOf(message), redeliveredOf(message), acknowledgment);
+    }
+
+    private static boolean redeliveredOf(Message message) {
+        try {
+            return message.getJMSRedelivered();
+        } catch (JMSException e) {
+            return false;
+        }
+    }
+
+    private static byte[] bodyOf(Message message) {
+        try {
+            if (message instanceof BytesMessage bytes) {
+                byte[] body = new byte[(int) bytes.getBodyLength()];
+                bytes.readBytes(body);
+                return body;
+            }
+            return new byte[0];
+        } catch (JMSException e) {
+            return new byte[0];
+        }
+    }
+
+    private static Map<String, String> headersOf(Message message) {
+        Map<String, String> headers = new HashMap<>();
+        try {
+            Enumeration<String> names = message.getPropertyNames();
+            while (names.hasMoreElements()) {
+                String name = names.nextElement();
+                Object value = message.getObjectProperty(name);
+                headers.put(portHeaderName(name), Objects.isNull(value) ? null : String.valueOf(value));
+            }
+        } catch (JMSException e) {
+            log.debug("ActiveMQ header read failed", e);
+        }
+        return headers;
+    }
+
+    private Message jmsMessage(JMSContext context, WireMqMessage wire) throws JMSException {
+        BytesMessage message = context.createBytesMessage();
+        message.writeBytes(wire.body());
+        for (Map.Entry<String, String> header : wire.headers().entrySet()) {
+            if (Objects.nonNull(header.getValue())) {
+                message.setStringProperty(jmsHeaderName(header.getKey()), header.getValue());
+            }
+        }
+        return message;
+    }
+
+    /**
+     * JMS property names must be valid java identifiers; the standard envelope headers
+     * use dashes, so they ride the wire underscored.
+     */
+    private static String jmsHeaderName(String name) {
+        return name.replace('-', '_');
+    }
+
+    private static String portHeaderName(String jmsName) {
+        return switch (jmsName) {
+            case "dc3_type" -> "dc3-type";
+            case "X_Request_Id" -> "X-Request-Id";
+            case "dc3_correlation_id" -> "dc3-correlation-id";
+            case "tenant_id" -> "tenant-id";
+            default -> jmsName;
+        };
+    }
+
+    private Destination destinationOf(MqTopic topic) {
+        if (isDeadLetterTopic(topic)) {
+            return publishContext.createQueue(destinationName(topic));
+        }
+        return publishContext.createTopic(destinationName(topic));
+    }
+
+    private static boolean isDeadLetterTopic(MqTopic topic) {
+        return topic == MqTopic.POINT_VALUE_DEAD || topic == MqTopic.POINT_COMMAND_DEAD
+                || topic == MqTopic.COMMAND_DEAD;
+    }
+
+    private static String destinationName(MqTopic topic) {
+        return switch (topic) {
+            case POINT_VALUE_DEAD -> "dc3.point_value.dlq";
+            case POINT_COMMAND_DEAD -> "dc3.point_command.dlq";
+            case COMMAND_DEAD -> "dc3.command.dlq";
+            default -> "dc3." + topic.name().toLowerCase();
+        };
+    }
+
+    private static String deadLetterQueue(MqTopic topic) {
+        return destinationName(topic) + ".dlq";
+    }
+}

dc3-common/dc3-common-mq-activemq/src/main/java/io/github/pnoker/common/mq/activemq/config/ActiveMqAdapterConfiguration.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.mq.activemq.config;
+
+import io.github.pnoker.common.mq.activemq.ActiveMqAdapter;
+import io.github.pnoker.common.mq.config.BatchConsumerProperties;
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+
+/**
+ * Activates the ActiveMQ (Artemis client, JMS 2.0) adapter when
+ * {@code dc3.mq.type=activemq}. Connection URL comes from
+ * {@code dc3.mq.activemq.url} (also covers ActiveMQ Classic brokers).
+ *
+ * @author pnoker
+ * @since 2026.8.19
+ */
+@AutoConfiguration
+@ConditionalOnProperty(prefix = "dc3.mq", name = "type", havingValue = "activemq")
+public class ActiveMqAdapterConfiguration {
+
+    @Bean
+    @ConditionalOnMissingBean(jakarta.jms.ConnectionFactory.class)
+    public ActiveMQConnectionFactory activeMqConnectionFactory(
+            @Value("${dc3.mq.activemq.url:tcp://localhost:61616}") String url,
+            @Value("${dc3.mq.activemq.user:}") String user,
+            @Value("${dc3.mq.activemq.password:}") String password) {
+        return new ActiveMQConnectionFactory(url, user, password);
+    }
+
+    @Bean
+    public ActiveMqAdapter activeMqAdapter(jakarta.jms.ConnectionFactory connectionFactory,
+                                           BatchConsumerProperties batchProperties) {
+        return new ActiveMqAdapter(connectionFactory, batchProperties);
+    }
+}

dc3-common/dc3-common-mq-activemq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1,18 @@
+#
+# Copyright 2016-present the IoT DC3 original author or authors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+
+io.github.pnoker.common.mq.activemq.config.ActiveMqAdapterConfiguration

dc3-common/dc3-common-mq-tck/pom.xml

@@ -66,6 +66,11 @@
             <artifactId>dc3-common-mq-kafka</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.github.pnoker</groupId>
+            <artifactId>dc3-common-mq-activemq</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>

dc3-common/dc3-common-mq-tck/src/test/java/io/github/pnoker/common/mq/tck/ActiveMqContractTest.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.mq.tck;
+
+import io.github.pnoker.common.mq.activemq.ActiveMqAdapter;
+import io.github.pnoker.common.mq.adapter.BrokerAdapter;
+import io.github.pnoker.common.mq.config.BatchConsumerProperties;
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import java.util.Objects;
+
+/**
+ * ActiveMQ (Artemis) harness for the broker-neutral contract suite. By default a
+ * disposable Artemis container; {@code TCK_ARTEMIS_URL} points it at an externally
+ * managed broker instead.
+ *
+ * @author pnoker
+ * @since 2026.8.19
+ */
+class ActiveMqContractTest extends AbstractMqContractTest {
+
+    private static final String EXTERNAL_URL = System.getenv("TCK_ARTEMIS_URL");
+
+    // started manually (not via the extension) so TCK_ARTEMIS_URL fully bypasses it
+    private static final GenericContainer<?> ARTEMIS =
+            new GenericContainer<>(DockerImageName.parse("apache/activemq-artemis:2.38.0-alpine"))
+                    .withExposedPorts(61616)
+                    .withEnv("ARTEMIS_USER", "artemis")
+                    .withEnv("ARTEMIS_PASSWORD", "artemis");
+
+    private static String brokerUrl() {
+        if (Objects.nonNull(EXTERNAL_URL)) {
+            return EXTERNAL_URL;
+        }
+        if (!ARTEMIS.isRunning()) {
+            ARTEMIS.start();
+        }
+        return "tcp://" + ARTEMIS.getHost() + ":" + ARTEMIS.getMappedPort(61616);
+    }
+
+    private ActiveMqAdapter activeMqAdapter;
+
+    @Override
+    protected BrokerAdapter adapter() {
+        if (Objects.isNull(activeMqAdapter)) {
+            BatchConsumerProperties properties = new BatchConsumerProperties();
+            properties.setBatchSize(10);
+            properties.setReceiveTimeoutMillis(100);
+            properties.setMaxRetries(2);
+            properties.setRetryInitialIntervalMillis(100);
+            properties.setRetryMultiplier(2);
+            properties.setRetryMaxIntervalMillis(200);
+            activeMqAdapter = new ActiveMqAdapter(new ActiveMQConnectionFactory(brokerUrl(), "artemis", "artemis"), properties);
+        }
+        return activeMqAdapter;
+    }
+
+    @Override
+    protected void shutdownAdapter() {
+        if (Objects.nonNull(activeMqAdapter)) {
+            activeMqAdapter.stop();
+        }
+    }
+
+    /**
+     * JMS has no per-instance subscription expiry (capability false).
+     */
+    @Test
+    @Override
+    public void perInstanceSubscriptionExpiresAfterInstanceStops() {
+        adapter();
+        Assumptions.assumeTrue(activeMqAdapter.capabilities().subscriptionExpiry(),
+                "activemq declares subscriptionExpiry=false");
+    }
+}

dc3-common/pom.xml

@@ -95,6 +95,7 @@
         <module>dc3-common-mq</module>
         <module>dc3-common-mq-rabbitmq</module>
         <module>dc3-common-mq-kafka</module>
+        <module>dc3-common-mq-activemq</module>
         <module>dc3-common-mq-tck</module>
         <module>dc3-common-postgres</module>
         <module>dc3-common-public</module>
@@ -387,6 +388,11 @@
                 <artifactId>dc3-common-mq-kafka</artifactId>
                 <version>${dc3.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.github.pnoker</groupId>
+                <artifactId>dc3-common-mq-activemq</artifactId>
+                <version>${dc3.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.github.pnoker</groupId>
                 <artifactId>dc3-common-mq-tck</artifactId>

pom.xml

@@ -771,6 +771,11 @@
                 <artifactId>dc3-common-mq-kafka</artifactId>
                 <version>${dc3.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.github.pnoker</groupId>
+                <artifactId>dc3-common-mq-activemq</artifactId>
+                <version>${dc3.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.github.pnoker</groupId>
                 <artifactId>dc3-common-mq-tck</artifactId>