This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 1e8919b648 Issue #2653 : Add JMS consumer and producer transforms
(#7860)
1e8919b648 is described below
commit 1e8919b6488281bb842e18b79c8fe7c961a6aacb
Author: vbhanuchander-lang <[email protected]>
AuthorDate: Thu Aug 13 04:33:54 2026 -0400
Issue #2653 : Add JMS consumer and producer transforms (#7860)
* Issue #2653 : Add JMS consumer and producer transforms
Hop can read from and write to Kafka, but has no support for any other
message broker. This adds JMS transforms so RabbitMQ, ActiveMQ "Classic"
and ActiveMQ Artemis can be used without writing middleware to bridge
them into a pipeline.
- IMessageQueueProvider: the broker-independent contract sketched on the
issue -- connect/disconnect, next message, message information, and
acknowledge -- plus send, with MessageQueueRecord carrying key, body,
destination, timestamp and properties.
- JmsProvider implements it over JMS 3.0. The ActiveMQ Artemis client is
bundled so it works with no extra setup; any other JMS provider is
reached by adding its client jar and using JNDI.
- JmsConnection metadata type so credentials are configured once and
shared by both transforms.
- JMS consumer reads a queue or topic into rows, with message selectors,
durable subscriptions and optional output fields. JMS producer sends
rows as text messages and passes them through unchanged.
The consumer uses CLIENT_ACKNOWLEDGE and acknowledges only after the row
has been passed downstream, so a failure mid-pipeline leaves the message
on the broker instead of losing it silently. Consumption stops on a
message limit, a receive timeout, or the pipeline stopping, rather than
blocking a pipeline indefinitely.
The provider interface lives inside the plugin rather than being a Hop
plugin type. Promoting it is the natural next step once a second,
structurally different provider has been written against it and the shape
has been confirmed by more than one caller.
* Issue #2653 : Package the JMS plugin and test it against a real broker
Two gaps in the previous commit.
The plugin had no src/assembly/assembly.xml, so it never produced the zip
that assemblies/plugins consumes and the assembly build could not resolve
hop-transform-jms:zip. Adds the descriptor and the version.xml it filters,
matching the Kafka plugin.
Adds JmsProviderBrokerTest, which runs an embedded ActiveMQ Artemis broker
in-process (no Docker, no external service) and exercises the provider for
real: queue round trip with correlation id and properties, topic
publish/subscribe, a transacted send, variable resolution in the
destination, receive returning null on an empty queue, blank destination
and unreachable broker errors, and idempotent close.
It also proves the acknowledgement claim rather than asserting it: a
message received but not acknowledged is redelivered on reconnect, and
disappears only once acknowledged.
Named ...BrokerTest rather than ...BrokerIT deliberately -- Surefire's
default includes skip *IT classes, so the suffix decided whether the
broker test ran in CI at all.
* Issue #2653 : Move Artemis to org.apache.artemis 2.55.0
Review feedback from hansva: the Artemis client has moved groupId from
org.apache.activemq to org.apache.artemis, and the current release is
2.55.0 rather than 2.42.0.
Applied to both the client and the test-scoped embedded server in
lib/pom.xml and the plugin pom.
Only the Maven coordinates moved, not the Java packages: the reflective
factory lookup still resolves
org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory, and
the embedded broker test still starts a real broker and round-trips a
message on 2.55.0, so no source change was needed. All 18 tests pass.
* spotless
* fix pom
---------
Co-authored-by: Hans Van Akelyen <[email protected]>
---
assemblies/plugins/pom.xml | 6 +
.../images/transforms/icons/jms-connection.svg | 5 +
.../images/transforms/icons/jms-consumer.svg | 5 +
.../images/transforms/icons/jms-producer.svg | 5 +
docs/hop-user-manual/modules/ROOT/nav.adoc | 3 +
.../ROOT/pages/metadata-types/jms-connection.adoc | 91 +++++++
.../modules/ROOT/pages/pipeline/transforms.adoc | 2 +
.../pages/pipeline/transforms/jms-consumer.adoc | 100 +++++++
.../pages/pipeline/transforms/jms-producer.adoc | 62 +++++
lib/pom.xml | 18 ++
plugins/transforms/jms/pom.xml | 86 +++++++
plugins/transforms/jms/src/assembly/assembly.xml | 43 ++++
.../transforms/jms/consumer/JmsConsumer.java | 156 +++++++++++
.../transforms/jms/consumer/JmsConsumerData.java | 36 +++
.../transforms/jms/consumer/JmsConsumerDialog.java | 231 +++++++++++++++++
.../transforms/jms/consumer/JmsConsumerMeta.java | 145 +++++++++++
.../jms/consumer/JmsDestinationType.java | 35 +++
.../transforms/jms/producer/JmsProducer.java | 132 ++++++++++
.../transforms/jms/producer/JmsProducerData.java | 33 +++
.../transforms/jms/producer/JmsProducerDialog.java | 201 +++++++++++++++
.../transforms/jms/producer/JmsProducerMeta.java | 75 ++++++
.../jms/shared/IMessageQueueProvider.java | 67 +++++
.../transforms/jms/shared/JmsConnection.java | 99 +++++++
.../transforms/jms/shared/JmsConnectionEditor.java | 218 ++++++++++++++++
.../transforms/jms/shared/JmsDialogHelper.java | 123 +++++++++
.../transforms/jms/shared/JmsProvider.java | 286 +++++++++++++++++++++
.../transforms/jms/shared/MessageQueueRecord.java | 72 ++++++
.../jms/src/main/resources/dependencies.xml | 20 ++
.../jms/src/main/resources/jms-connection.svg | 5 +
.../jms/src/main/resources/jms-consumer.svg | 5 +
.../jms/src/main/resources/jms-producer.svg | 5 +
.../consumer/messages/messages_en_US.properties | 39 +++
.../producer/messages/messages_en_US.properties | 30 +++
.../jms/shared/messages/messages_en_US.properties | 32 +++
.../transforms/jms/src/main/resources/version.xml | 20 ++
.../jms/consumer/JmsConsumerMetaTest.java | 114 ++++++++
.../jms/shared/JmsProviderBrokerTest.java | 231 +++++++++++++++++
.../transforms/jms/shared/JmsProviderTest.java | 99 +++++++
plugins/transforms/pom.xml | 1 +
39 files changed, 2936 insertions(+)
diff --git a/assemblies/plugins/pom.xml b/assemblies/plugins/pom.xml
index b90c2461e1..ce4e3ff083 100644
--- a/assemblies/plugins/pom.xml
+++ b/assemblies/plugins/pom.xml
@@ -1134,6 +1134,12 @@
<version>${project.version}</version>
<type>zip</type>
</dependency>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-transform-jms</artifactId>
+ <version>${project.version}</version>
+ <type>zip</type>
+ </dependency>
<dependency>
<groupId>org.apache.hop</groupId>
<artifactId>hop-transform-joinrows</artifactId>
diff --git
a/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/jms-connection.svg
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/jms-connection.svg
new file mode 100644
index 0000000000..6fa5853939
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/jms-connection.svg
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="utf-8"?>
+<svg xmlns="http://www.w3.org/2000/svg"; width="24" height="24">
+ <path d="M 2,6 H 22 V 18 H 2 Z M 4,8 v 8 h 4 V 8 Z M 10,8 v 8 h 4 V 8 Z M
16,8 v 8 h 4 V 8 Z"
+ style="fill:#0e3a5a;fill-opacity:1;fill-rule:evenodd"/>
+</svg>
diff --git
a/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/jms-consumer.svg
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/jms-consumer.svg
new file mode 100644
index 0000000000..6fa5853939
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/jms-consumer.svg
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="utf-8"?>
+<svg xmlns="http://www.w3.org/2000/svg"; width="24" height="24">
+ <path d="M 2,6 H 22 V 18 H 2 Z M 4,8 v 8 h 4 V 8 Z M 10,8 v 8 h 4 V 8 Z M
16,8 v 8 h 4 V 8 Z"
+ style="fill:#0e3a5a;fill-opacity:1;fill-rule:evenodd"/>
+</svg>
diff --git
a/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/jms-producer.svg
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/jms-producer.svg
new file mode 100644
index 0000000000..6fa5853939
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/jms-producer.svg
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="utf-8"?>
+<svg xmlns="http://www.w3.org/2000/svg"; width="24" height="24">
+ <path d="M 2,6 H 22 V 18 H 2 Z M 4,8 v 8 h 4 V 8 Z M 10,8 v 8 h 4 V 8 Z M
16,8 v 8 h 4 V 8 Z"
+ style="fill:#0e3a5a;fill-opacity:1;fill-rule:evenodd"/>
+</svg>
diff --git a/docs/hop-user-manual/modules/ROOT/nav.adoc
b/docs/hop-user-manual/modules/ROOT/nav.adoc
index f8a1265bae..5f888f9711 100644
--- a/docs/hop-user-manual/modules/ROOT/nav.adoc
+++ b/docs/hop-user-manual/modules/ROOT/nav.adoc
@@ -184,6 +184,8 @@ under the License.
*** xref:pipeline/transforms/javascript.adoc[JavaScript]
*** xref:pipeline/transforms/jdbcmetadata.adoc[JDBC Metadata]
*** xref:pipeline/transforms/joinrows.adoc[Join Rows]
+*** xref:pipeline/transforms/jms-consumer.adoc[JMS consumer]
+*** xref:pipeline/transforms/jms-producer.adoc[JMS producer]
*** xref:pipeline/transforms/jsoninput.adoc[JSON Input]
*** xref:pipeline/transforms/jsonnormalizeinput.adoc[JSON Normalize Input]
*** xref:pipeline/transforms/jsonoutput.adoc[JSON Output]
@@ -490,6 +492,7 @@ under the License.
** xref:metadata-types/salesforce-connection.adoc[Salesforce Connection]
** xref:metadata-types/sftp-connection.adoc[SFTP Connection]
** xref:metadata-types/splunk-connection.adoc[Splunk Connection]
+** xref:metadata-types/jms-connection.adoc[JMS connection]
** xref:metadata-types/spark-catalog.adoc[Spark Catalog]
** xref:metadata-types/variable-resolver/index.adoc[Variable Resolver]
***
xref:metadata-types/variable-resolver/azure-key-vault-variable-resolver.adoc[Azure
Key Vault variable resolver]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/metadata-types/jms-connection.adoc
b/docs/hop-user-manual/modules/ROOT/pages/metadata-types/jms-connection.adoc
new file mode 100644
index 0000000000..e069bbd096
--- /dev/null
+++ b/docs/hop-user-manual/modules/ROOT/pages/metadata-types/jms-connection.adoc
@@ -0,0 +1,91 @@
+////
+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.
+////
+:documentationPath: /metadata-types/
+:language: en_US
+:description: A JMS connection describes how to reach a JMS broker, and is
shared by the JMS consumer and producer transforms.
+
+= JMS connection
+
+== Description
+
+A *JMS connection* describes how to reach a JMS broker. It is shared by the
xref:pipeline/transforms/jms-consumer.adoc[JMS consumer] and
xref:pipeline/transforms/jms-producer.adoc[JMS producer] transforms, so broker
credentials live in one place.
+
+== Modes
+
+=== DIRECT
+
+Uses the Apache ActiveMQ Artemis client that ships with the JMS transforms.
Supply a broker URL and nothing else needs installing:
+
+[source]
+----
+tcp://localhost:61616
+----
+
+=== JNDI
+
+Looks the connection factory up through JNDI. This is how any other JMS 3.0
provider is used: put its client jar in the JMS transform plugin's `lib`
folder, then name its initial context factory and connection factory here.
+
+For Apache ActiveMQ "Classic":
+
+[options="header",cols="1,2"]
+|===
+|Option |Value
+
+|Initial context factory
+|`org.apache.activemq.jndi.ActiveMQInitialContextFactory`
+
+|Provider URL
+|`tcp://localhost:61616`
+
+|Connection factory name
+|`ConnectionFactory`
+|===
+
+For RabbitMQ, use the initial context factory from its JMS client; for IBM MQ,
the one from its `com.ibm.mq.jakarta.client` distribution.
+
+== Options
+
+[options="header",cols="1,3"]
+|===
+|Option |Description
+
+|Connection name
+|Name this metadata object is referenced by.
+
+|Mode
+|`DIRECT` or `JNDI`.
+
+|Broker URL
+|DIRECT mode only. Supports variables.
+
+|Initial context factory
+|JNDI mode only: the `Context.INITIAL_CONTEXT_FACTORY` class name. Supports
variables.
+
+|Provider URL
+|JNDI mode only: the `Context.PROVIDER_URL`. Supports variables.
+
+|Connection factory name
+|JNDI mode only: the name the connection factory is bound to. Defaults to
`ConnectionFactory`. Supports variables.
+
+|Username / Password
+|Credentials. Leave the username empty for an unauthenticated connection. The
password is stored encrypted like any other Hop password.
+
+|Client id
+|Optional JMS client id. The JMS specification requires one for a durable
topic subscription.
+|===
+
+TIP: Put the broker URL and credentials in xref:projects/index.adoc[project or
environment variables] so the same pipeline runs against a development and a
production broker without editing.
diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
index 3e1a4b1064..1582534598 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
@@ -134,6 +134,8 @@ The pages nested under this topic contain information on
how to use the transfor
* xref:pipeline/transforms/javascript.adoc[JavaScript]
* xref:pipeline/transforms/jdbcmetadata.adoc[JDBC Metadata]
* xref:pipeline/transforms/joinrows.adoc[Join Rows]
+* xref:pipeline/transforms/jms-consumer.adoc[JMS consumer]
+* xref:pipeline/transforms/jms-producer.adoc[JMS producer]
* xref:pipeline/transforms/jsoninput.adoc[JSON Input]
* xref:pipeline/transforms/jsonoutput.adoc[JSON Output]
* xref:pipeline/transforms/kafkaconsumer.adoc[Kafka consumer]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/jms-consumer.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/jms-consumer.adoc
new file mode 100644
index 0000000000..a5c9d21691
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/jms-consumer.adoc
@@ -0,0 +1,100 @@
+////
+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.
+////
+:documentationPath: /pipeline/transforms/
+:language: en_US
+:description: JMS consumer reads messages from a JMS queue or topic and turns
each into a row.
+
+= image:transforms/icons/jms-consumer.svg[JMS consumer transform Icon,
role="image-doc-icon"] JMS consumer
+
+== Description
+
+*JMS consumer* reads messages from a JMS queue or topic and turns each into a
row.
+
+It works with any JMS 3.0 provider. Apache ActiveMQ Artemis works with no
extra setup because its client ships with the transform; other brokers —
ActiveMQ "Classic", RabbitMQ through its JMS client, IBM MQ — are reached by
adding their client jar and using JNDI mode on the
xref:metadata-types/jms-connection.adoc[JMS connection].
+
+== Options
+
+[options="header",cols="1,3"]
+|===
+|Option |Description
+
+|Transform name
+|Unique name of the transform in the pipeline.
+
+|JMS connection
+|The xref:metadata-types/jms-connection.adoc[JMS connection] metadata object
to use.
+
+|Destination type
+|`QUEUE` for point-to-point, `TOPIC` for publish/subscribe.
+
+|Queue / topic name
+|Name of the destination. Supports variables.
+
+|Message selector
+|Optional JMS selector expression, evaluated by the broker so unwanted
messages are never delivered. For example `priority > 5 AND region = 'EU'`.
+
+|Durable subscription name
+|Topics only. Makes the subscription durable, so messages published while the
pipeline is not running are still delivered. The JMS specification requires a
client id on the connection for this.
+
+|Use a JMS transaction
+|Commit each message in a JMS transaction instead of acknowledging it
individually.
+
+|Maximum messages
+|Stop after this many messages. `0` means no limit. Supports variables.
+
+|Receive timeout (ms)
+|How long to wait for a message before concluding the destination is drained —
see <<When the transform stops>>.
+
+|Message body field
+|Field to put the message body in. Leave empty to omit it.
+
+|Correlation id field
+|Field for the JMS correlation id. Leave empty to omit it.
+
+|Destination field
+|Field for the destination the message came from. Leave empty to omit it.
+
+|Message id field
+|Field for the broker-assigned message id. Leave empty to omit it.
+
+|Timestamp field
+|Field for the broker timestamp, as a Date. Leave empty to omit it.
+|===
+
+Every output field is optional: naming a field adds it to the row, leaving it
empty keeps it out. Fields appear in the row in the order listed above.
+
+== When the transform stops
+
+Consumption ends on whichever comes first:
+
+* the *Maximum messages* count is reached,
+* no message arrives within *Receive timeout*, or
+* the pipeline is stopped.
+
+The timeout is what makes this a batch-friendly transform: it drains what is
on the destination and finishes, rather than blocking a pipeline forever. To
poll continuously, schedule the pipeline to run repeatedly rather than raising
the timeout.
+
+== Acknowledgement and message loss
+
+The transform uses `CLIENT_ACKNOWLEDGE`, and acknowledges each message only
*after* its row has been passed downstream. If the pipeline fails partway,
unacknowledged messages stay on the broker and are redelivered.
+
+`AUTO_ACKNOWLEDGE` is deliberately not used: it confirms the message on
receipt, so a failure later in the pipeline would lose it silently.
+
+NOTE: Acknowledging after `putRow` guarantees the row was *handed to* the next
transform, not that the pipeline finished with it. For end-to-end exactly-once
behaviour you still need an idempotent target.
+
+== Message bodies
+
+Only `TextMessage` bodies are read. Any other message type still produces a
row with its identifiers, destination, timestamp and properties, but the body
field is left null rather than guessing at an encoding.
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/jms-producer.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/jms-producer.adoc
new file mode 100644
index 0000000000..f5e494e27e
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/jms-producer.adoc
@@ -0,0 +1,62 @@
+////
+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.
+////
+:documentationPath: /pipeline/transforms/
+:language: en_US
+:description: JMS producer sends each incoming row to a JMS queue or topic as
a text message.
+
+= image:transforms/icons/jms-producer.svg[JMS producer transform Icon,
role="image-doc-icon"] JMS producer
+
+== Description
+
+*JMS producer* sends each incoming row to a JMS queue or topic as a
`TextMessage`.
+
+It works with any JMS 3.0 provider — see
xref:pipeline/transforms/jms-consumer.adoc[JMS consumer] for the broker notes,
which apply equally here.
+
+Rows are passed through unchanged, so the transform can sit in the middle of a
pipeline rather than only at the end.
+
+== Options
+
+[options="header",cols="1,3"]
+|===
+|Option |Description
+
+|Transform name
+|Unique name of the transform in the pipeline.
+
+|JMS connection
+|The xref:metadata-types/jms-connection.adoc[JMS connection] metadata object
to use.
+
+|Destination type
+|`QUEUE` for point-to-point, `TOPIC` for publish/subscribe.
+
+|Queue / topic name
+|Name of the destination. Supports variables.
+
+|Use a JMS transaction
+|Commit each message in a JMS transaction instead of sending it outside one.
+
+|Message body field
+|Incoming field holding the message body. Required.
+
+|Correlation id field
+|Optional incoming field to set as the JMS correlation id, so replies can be
matched back to the row that caused them.
+|===
+
+== Notes
+
+* The body field is read as a string using its Hop value metadata, so a
non-string field is converted with the conversion mask configured on it.
+* A missing body or correlation id field fails the transform on the first row
with the field name it looked for, rather than sending empty messages.
diff --git a/lib/pom.xml b/lib/pom.xml
index d07ad73b23..91cee4ff11 100644
--- a/lib/pom.xml
+++ b/lib/pom.xml
@@ -33,6 +33,8 @@
<angus-activation.version>2.0.3</angus-activation.version>
<angus-mail.version>2.0.5</angus-mail.version>
<annotations.version>26.1.0</annotations.version>
+ <artemis-jakarta-client.version>2.55.0</artemis-jakarta-client.version>
+ <artemis-server.version>2.55.0</artemis-server.version>
<asm.version>9.10.1</asm.version>
<avro.version>1.12.1</avro.version>
<aws-java-sdk.version>2.20.162</aws-java-sdk.version>
@@ -68,6 +70,7 @@
<jackson.version>2.21.5</jackson.version>
<jakarta-activation-api.version>2.1.4</jakarta-activation-api.version>
<jakarta-annotation-api.version>3.0.0</jakarta-annotation-api.version>
+ <jakarta-jms-api.version>3.1.0</jakarta-jms-api.version>
<jakarta-servlet-api.version>6.1.0</jakarta-servlet-api.version>
<jakarta-ws-rs-api.version>4.0.0</jakarta-ws-rs-api.version>
<jakarta-xml-bind-api.version>4.0.5</jakarta-xml-bind-api.version>
@@ -221,6 +224,21 @@
<artifactId>ST4</artifactId>
<version>${ST4.version}</version>
</dependency>
+ <dependency>
+ <groupId>jakarta.jms</groupId>
+ <artifactId>jakarta.jms-api</artifactId>
+ <version>${jakarta-jms-api.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.artemis</groupId>
+ <artifactId>artemis-jakarta-client</artifactId>
+ <version>${artemis-jakarta-client.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.artemis</groupId>
+ <artifactId>artemis-server</artifactId>
+ <version>${artemis-server.version}</version>
+ </dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
diff --git a/plugins/transforms/jms/pom.xml b/plugins/transforms/jms/pom.xml
new file mode 100644
index 0000000000..49df223d6d
--- /dev/null
+++ b/plugins/transforms/jms/pom.xml
@@ -0,0 +1,86 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ ~
+ -->
+<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
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-plugins-transforms</artifactId>
+ <version>2.20.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>hop-transform-jms</artifactId>
+ <packaging>jar</packaging>
+ <name>Hop Plugins Transforms JMS</name>
+ <description>Consume from and produce to JMS destinations (Apache ActiveMQ
Artemis, ActiveMQ "Classic",
+ RabbitMQ via its JMS client, and any other JMS 3.0 provider reachable
through JNDI).</description>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-libs</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <dependencies>
+ <!--
+ The API is what the transform compiles against. Any JMS 3.0 provider
can be used at
+ runtime by dropping its client jar next to this plugin and pointing
the connection at
+ it through JNDI.
+ -->
+ <dependency>
+ <groupId>jakarta.jms</groupId>
+ <artifactId>jakarta.jms-api</artifactId>
+ </dependency>
+ <!--
+ Bundled so Apache ActiveMQ Artemis works with no extra setup, the
same way the Spark
+ plugin ships the lake-format connectors.
+ -->
+ <dependency>
+ <groupId>org.apache.artemis</groupId>
+ <artifactId>artemis-jakarta-client</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-engine</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-ui</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Embedded broker so the provider is exercised against a real
broker in unit tests. -->
+ <dependency>
+ <groupId>org.apache.artemis</groupId>
+ <artifactId>artemis-server</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+</project>
diff --git a/plugins/transforms/jms/src/assembly/assembly.xml
b/plugins/transforms/jms/src/assembly/assembly.xml
new file mode 100644
index 0000000000..7a22fa8e7f
--- /dev/null
+++ b/plugins/transforms/jms/src/assembly/assembly.xml
@@ -0,0 +1,43 @@
+<!--
+ ~ 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.
+ ~
+ -->
+
+<assembly xmlns="http://maven.apache.org/ASSEMBLY/2.2.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/ASSEMBLY/2.2.0
http://maven.apache.org/xsd/assembly-2.2.0.xsd";>
+ <id>hop-transform-jms</id>
+ <formats>
+ <format>zip</format>
+ </formats>
+ <baseDirectory>.</baseDirectory>
+ <files>
+ <file>
+ <source>${project.basedir}/src/main/resources/version.xml</source>
+ <outputDirectory>${hop.plugin.libdir}</outputDirectory>
+ <filtered>true</filtered>
+ </file>
+ <file>
+
<source>${project.basedir}/src/main/resources/dependencies.xml</source>
+ <outputDirectory>${hop.plugin.libdir}</outputDirectory>
+ <filtered>true</filtered>
+ </file>
+ </files>
+
+ <componentDescriptors>
+
<componentDescriptor>${maven.multiModuleProjectDirectory}/assemblies/shared/hop-plugin-libs.xml</componentDescriptor>
+ </componentDescriptors>
+</assembly>
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumer.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumer.java
new file mode 100644
index 0000000000..dbb7ed28ee
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumer.java
@@ -0,0 +1,156 @@
+/*
+ * 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.hop.pipeline.transforms.jms.consumer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hop.core.Const;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.pipeline.Pipeline;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.transform.BaseTransform;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transforms.jms.shared.JmsConnection;
+import org.apache.hop.pipeline.transforms.jms.shared.JmsProvider;
+import org.apache.hop.pipeline.transforms.jms.shared.MessageQueueRecord;
+
+public class JmsConsumer extends BaseTransform<JmsConsumerMeta,
JmsConsumerData> {
+
+ public JmsConsumer(
+ TransformMeta transformMeta,
+ JmsConsumerMeta meta,
+ JmsConsumerData data,
+ int copyNr,
+ PipelineMeta pipelineMeta,
+ Pipeline pipeline) {
+ super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline);
+ }
+
+ @Override
+ public boolean init() {
+ if (!super.init()) {
+ return false;
+ }
+ try {
+ JmsConnection connection =
+ getMetadataProvider()
+ .getSerializer(JmsConnection.class)
+ .load(resolve(meta.getConnectionName()));
+ if (connection == null) {
+ logError(
+ "JMS connection '" + meta.getConnectionName() + "' could not be
found in the metadata");
+ return false;
+ }
+
+ data.maxMessages = Const.toLong(resolve(meta.getMaxMessages()), 0L);
+ data.receiveTimeout = Const.toLong(resolve(meta.getReceiveTimeout()),
5000L);
+ if (data.receiveTimeout <= 0) {
+ logError("The receive timeout must be greater than zero");
+ return false;
+ }
+
+ data.provider =
+ new JmsProvider(
+ connection,
+ meta.getDestination(),
+ meta.isTopic(),
+ meta.isTransacted(),
+ resolve(meta.getMessageSelector()),
+ resolve(meta.getDurableSubscription()),
+ getLogChannel());
+ data.provider.connect(this);
+ return true;
+ } catch (Exception e) {
+ logError("Unable to start the JMS consumer", e);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean processRow() throws HopException {
+ if (first) {
+ first = false;
+ data.outputRowMeta = new RowMeta();
+ meta.getFields(data.outputRowMeta, getTransformName(), null, null, this,
metadataProvider);
+ }
+
+ if (data.maxMessages > 0 && data.messageCount >= data.maxMessages) {
+ setOutputDone();
+ return false;
+ }
+
+ MessageQueueRecord record = data.provider.receive(data.receiveTimeout);
+ if (record == null) {
+ // Nothing waiting within the timeout: treat the destination as drained.
A consumer that
+ // should wait indefinitely belongs behind a scheduler, not a blocked
pipeline.
+ if (isDetailed()) {
+ logDetailed(
+ "No message received within "
+ + data.receiveTimeout
+ + "ms, stopping after "
+ + data.messageCount
+ + " message(s)");
+ }
+ setOutputDone();
+ return false;
+ }
+
+ putRow(data.outputRowMeta, buildRow(record));
+
+ // Acknowledge only once the row is on its way, so a failure upstream of
this point leaves
+ // the message on the broker rather than silently consuming it.
+ data.provider.acknowledge(record);
+ data.messageCount++;
+
+ if (checkFeedback(getLinesRead()) && isBasic()) {
+ logBasic("Read " + getLinesRead() + " messages from the JMS
destination");
+ }
+ return true;
+ }
+
+ /** Builds the output row in the same order {@link
JmsConsumerMeta#getFields} declares it. */
+ private Object[] buildRow(MessageQueueRecord record) {
+ Object[] row = new Object[data.outputRowMeta.size()];
+ int index = 0;
+ if (StringUtils.isNotEmpty(resolve(meta.getBodyField()))) {
+ row[index++] = record.getBody();
+ }
+ if (StringUtils.isNotEmpty(resolve(meta.getKeyField()))) {
+ row[index++] = record.getKey();
+ }
+ if (StringUtils.isNotEmpty(resolve(meta.getDestinationField()))) {
+ row[index++] = record.getDestination();
+ }
+ if (StringUtils.isNotEmpty(resolve(meta.getMessageIdField()))) {
+ row[index++] = record.getMessageId();
+ }
+ if (StringUtils.isNotEmpty(resolve(meta.getTimestampField()))) {
+ row[index] = record.getTimestamp();
+ }
+ return row;
+ }
+
+ @Override
+ public void dispose() {
+ if (data.provider != null) {
+ data.provider.close();
+ data.provider = null;
+ }
+ super.dispose();
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerData.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerData.java
new file mode 100644
index 0000000000..7252c05043
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerData.java
@@ -0,0 +1,36 @@
+/*
+ * 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.hop.pipeline.transforms.jms.consumer;
+
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.pipeline.transform.BaseTransformData;
+import org.apache.hop.pipeline.transform.ITransformData;
+import org.apache.hop.pipeline.transforms.jms.shared.IMessageQueueProvider;
+
+public class JmsConsumerData extends BaseTransformData implements
ITransformData {
+
+ public IRowMeta outputRowMeta;
+ public IMessageQueueProvider provider;
+ public long maxMessages;
+ public long receiveTimeout;
+ public long messageCount;
+
+ public JmsConsumerData() {
+ super();
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerDialog.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerDialog.java
new file mode 100644
index 0000000000..6b60cb9dc0
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerDialog.java
@@ -0,0 +1,231 @@
+/*
+ * 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.hop.pipeline.transforms.jms.consumer;
+
+import org.apache.hop.core.Const;
+import org.apache.hop.core.util.Utils;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.transforms.jms.shared.JmsDialogHelper;
+import org.apache.hop.ui.core.PropsUi;
+import org.apache.hop.ui.core.dialog.BaseDialog;
+import org.apache.hop.ui.core.widget.TextVar;
+import org.apache.hop.ui.pipeline.transform.BaseTransformDialog;
+import org.eclipse.swt.SWT;
+import org.eclipse.swt.custom.CCombo;
+import org.eclipse.swt.layout.FormAttachment;
+import org.eclipse.swt.layout.FormData;
+import org.eclipse.swt.layout.FormLayout;
+import org.eclipse.swt.widgets.Button;
+import org.eclipse.swt.widgets.Control;
+import org.eclipse.swt.widgets.Label;
+import org.eclipse.swt.widgets.Shell;
+import org.eclipse.swt.widgets.Text;
+
+public class JmsConsumerDialog extends BaseTransformDialog {
+ private static final Class<?> PKG = JmsConsumerMeta.class;
+
+ private final JmsConsumerMeta input;
+
+ private CCombo wConnection;
+ private CCombo wDestinationType;
+ private TextVar wDestination;
+ private TextVar wMessageSelector;
+ private TextVar wDurableSubscription;
+ private Button wTransacted;
+ private TextVar wMaxMessages;
+ private TextVar wReceiveTimeout;
+ private TextVar wBodyField;
+ private TextVar wKeyField;
+ private TextVar wDestinationField;
+ private TextVar wMessageIdField;
+ private TextVar wTimestampField;
+
+ public JmsConsumerDialog(
+ Shell parent,
+ IVariables variables,
+ JmsConsumerMeta transformMeta,
+ PipelineMeta pipelineMeta) {
+ super(parent, variables, transformMeta, pipelineMeta);
+ this.input = transformMeta;
+ }
+
+ @Override
+ public String open() {
+ Shell parent = getParent();
+ shell = new Shell(parent, SWT.DIALOG_TRIM | SWT.RESIZE | SWT.MIN |
SWT.MAX);
+ PropsUi.setLook(shell);
+ setShellImage(shell, input);
+
+ changed = input.hasChanged();
+
+ FormLayout formLayout = new FormLayout();
+ formLayout.marginWidth = PropsUi.getFormMargin();
+ formLayout.marginHeight = PropsUi.getFormMargin();
+ shell.setLayout(formLayout);
+ shell.setText(BaseMessages.getString(PKG,
"JmsConsumerDialog.Shell.Title"));
+
+ int middle = props.getMiddlePct();
+ int margin = PropsUi.getMargin();
+
+ wlTransformName = new Label(shell, SWT.RIGHT);
+ wlTransformName.setText(BaseMessages.getString(PKG,
"System.Label.TransformName"));
+ PropsUi.setLook(wlTransformName);
+ fdlTransformName = new FormData();
+ fdlTransformName.left = new FormAttachment(0, 0);
+ fdlTransformName.top = new FormAttachment(0, margin);
+ fdlTransformName.right = new FormAttachment(middle, -margin);
+ wlTransformName.setLayoutData(fdlTransformName);
+ wTransformName = new Text(shell, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
+ wTransformName.setText(transformName);
+ PropsUi.setLook(wTransformName);
+ wTransformName.addModifyListener(e -> input.setChanged());
+ fdTransformName = new FormData();
+ fdTransformName.left = new FormAttachment(middle, 0);
+ fdTransformName.top = new FormAttachment(wlTransformName, 0, SWT.CENTER);
+ fdTransformName.right = new FormAttachment(100, 0);
+ wTransformName.setLayoutData(fdTransformName);
+ Control last = wTransformName;
+
+ wConnection =
+ JmsDialogHelper.labeledCombo(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.Connection", true);
+
wConnection.setItems(JmsDialogHelper.listConnectionNames(metadataProvider));
+ last = wConnection;
+
+ wDestinationType =
+ JmsDialogHelper.labeledCombo(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.DestinationType", true);
+ wDestinationType.setItems(
+ new String[] {JmsDestinationType.QUEUE.name(),
JmsDestinationType.TOPIC.name()});
+ last = wDestinationType;
+
+ wDestination =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.Destination");
+ last = wDestination;
+
+ wMessageSelector =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.MessageSelector");
+ last = wMessageSelector;
+
+ wDurableSubscription =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.DurableSubscription");
+ last = wDurableSubscription;
+
+ wTransacted =
+ JmsDialogHelper.labeledCheckbox(
+ shell, PKG, middle, margin, last, "JmsConsumerDialog.Transacted");
+ last = wTransacted;
+
+ wMaxMessages =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.MaxMessages");
+ last = wMaxMessages;
+
+ wReceiveTimeout =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.ReceiveTimeout");
+ last = wReceiveTimeout;
+
+ wBodyField =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.BodyField");
+ last = wBodyField;
+ wKeyField =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.KeyField");
+ last = wKeyField;
+ wDestinationField =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.DestinationField");
+ last = wDestinationField;
+ wMessageIdField =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.MessageIdField");
+ last = wMessageIdField;
+ wTimestampField =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsConsumerDialog.TimestampField");
+
+ wOk = new Button(shell, SWT.PUSH);
+ wOk.setText(BaseMessages.getString(PKG, "System.Button.OK"));
+ wCancel = new Button(shell, SWT.PUSH);
+ wCancel.setText(BaseMessages.getString(PKG, "System.Button.Cancel"));
+ setButtonPositions(new Button[] {wOk, wCancel}, margin, wTimestampField);
+
+ wOk.addListener(SWT.Selection, e -> ok());
+ wCancel.addListener(SWT.Selection, e -> cancel());
+
+ getData();
+ input.setChanged(changed);
+ BaseDialog.defaultShellHandling(shell, c -> ok(), c -> cancel());
+ return transformName;
+ }
+
+ private void getData() {
+ wTransformName.setText(Const.NVL(transformName, ""));
+ wConnection.setText(Const.NVL(input.getConnectionName(), ""));
+ wDestinationType.setText(
+ Const.NVL(input.getDestinationType(),
JmsDestinationType.QUEUE.name()));
+ wDestination.setText(Const.NVL(input.getDestination(), ""));
+ wMessageSelector.setText(Const.NVL(input.getMessageSelector(), ""));
+ wDurableSubscription.setText(Const.NVL(input.getDurableSubscription(),
""));
+ wTransacted.setSelection(input.isTransacted());
+ wMaxMessages.setText(Const.NVL(input.getMaxMessages(), "0"));
+ wReceiveTimeout.setText(Const.NVL(input.getReceiveTimeout(), "5000"));
+ wBodyField.setText(Const.NVL(input.getBodyField(), ""));
+ wKeyField.setText(Const.NVL(input.getKeyField(), ""));
+ wDestinationField.setText(Const.NVL(input.getDestinationField(), ""));
+ wMessageIdField.setText(Const.NVL(input.getMessageIdField(), ""));
+ wTimestampField.setText(Const.NVL(input.getTimestampField(), ""));
+
+ wTransformName.selectAll();
+ wTransformName.setFocus();
+ }
+
+ private void cancel() {
+ transformName = null;
+ input.setChanged(changed);
+ dispose();
+ }
+
+ private void ok() {
+ if (Utils.isEmpty(wTransformName.getText())) {
+ return;
+ }
+ transformName = wTransformName.getText();
+ input.setConnectionName(wConnection.getText());
+ input.setDestinationType(wDestinationType.getText());
+ input.setDestination(wDestination.getText());
+ input.setMessageSelector(wMessageSelector.getText());
+ input.setDurableSubscription(wDurableSubscription.getText());
+ input.setTransacted(wTransacted.getSelection());
+ input.setMaxMessages(wMaxMessages.getText());
+ input.setReceiveTimeout(wReceiveTimeout.getText());
+ input.setBodyField(wBodyField.getText());
+ input.setKeyField(wKeyField.getText());
+ input.setDestinationField(wDestinationField.getText());
+ input.setMessageIdField(wMessageIdField.getText());
+ input.setTimestampField(wTimestampField.getText());
+ dispose();
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerMeta.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerMeta.java
new file mode 100644
index 0000000000..95e4fd0367
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerMeta.java
@@ -0,0 +1,145 @@
+/*
+ * 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.hop.pipeline.transforms.jms.consumer;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hop.core.annotations.Transform;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.value.ValueMetaDate;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.pipeline.transform.BaseTransformMeta;
+import org.apache.hop.pipeline.transform.TransformMeta;
+
+/**
+ * Reads messages from a JMS queue or topic and turns each into a row.
+ *
+ * <p>Consumption stops on whichever comes first: the configured message
limit, a receive timeout
+ * with no message waiting, or the pipeline being stopped.
+ */
+@Transform(
+ id = "JmsConsumer",
+ name = "i18n::JmsConsumer.Name",
+ description = "i18n::JmsConsumer.Description",
+ image = "jms-consumer.svg",
+ categoryDescription =
"i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Streaming",
+ keywords = "i18n::JmsConsumer.Keyword",
+ documentationUrl = "/pipeline/transforms/jms-consumer.html")
+@Getter
+@Setter
+public class JmsConsumerMeta extends BaseTransformMeta<JmsConsumer,
JmsConsumerData> {
+
+ /** Name of the JMS connection metadata object. */
+ @HopMetadataProperty(key = "connection", injectionKey = "CONNECTION")
+ private String connectionName;
+
+ @HopMetadataProperty(key = "destination", injectionKey = "DESTINATION")
+ private String destination;
+
+ /** QUEUE or TOPIC. */
+ @HopMetadataProperty(key = "destination_type", injectionKey =
"DESTINATION_TYPE")
+ private String destinationType = JmsDestinationType.QUEUE.name();
+
+ /** Optional JMS message selector, e.g. {@code priority > 5}. */
+ @HopMetadataProperty(key = "message_selector", injectionKey =
"MESSAGE_SELECTOR")
+ private String messageSelector;
+
+ /** Durable subscription name. Topics only; requires a client id on the
connection. */
+ @HopMetadataProperty(key = "durable_subscription", injectionKey =
"DURABLE_SUBSCRIPTION")
+ private String durableSubscription;
+
+ /** Commit each message in a JMS transaction rather than acknowledging it
individually. */
+ @HopMetadataProperty(key = "transacted", injectionKey = "TRANSACTED")
+ private boolean transacted;
+
+ /** Stop after this many messages. 0 means no limit. Supports variables. */
+ @HopMetadataProperty(key = "max_messages", injectionKey = "MAX_MESSAGES")
+ private String maxMessages = "0";
+
+ /** How long to wait for a message before concluding the destination is
drained, in ms. */
+ @HopMetadataProperty(key = "receive_timeout", injectionKey =
"RECEIVE_TIMEOUT")
+ private String receiveTimeout = "5000";
+
+ // Output field names. An empty name leaves that field out of the row.
+ @HopMetadataProperty(key = "body_field", injectionKey = "BODY_FIELD")
+ private String bodyField = "message";
+
+ @HopMetadataProperty(key = "key_field", injectionKey = "KEY_FIELD")
+ private String keyField = "";
+
+ @HopMetadataProperty(key = "destination_field", injectionKey =
"DESTINATION_FIELD")
+ private String destinationField = "";
+
+ @HopMetadataProperty(key = "message_id_field", injectionKey =
"MESSAGE_ID_FIELD")
+ private String messageIdField = "";
+
+ @HopMetadataProperty(key = "timestamp_field", injectionKey =
"TIMESTAMP_FIELD")
+ private String timestampField = "";
+
+ public JmsConsumerMeta() {
+ super();
+ }
+
+ @Override
+ public String getDialogClassName() {
+ return JmsConsumerDialog.class.getName();
+ }
+
+ public boolean isTopic() {
+ return JmsDestinationType.TOPIC.name().equalsIgnoreCase(destinationType);
+ }
+
+ @Override
+ public void getFields(
+ IRowMeta inputRowMeta,
+ String name,
+ IRowMeta[] info,
+ TransformMeta nextTransform,
+ IVariables variables,
+ IHopMetadataProvider metadataProvider)
+ throws HopTransformException {
+ inputRowMeta.clear();
+ addStringField(inputRowMeta, variables, bodyField, name);
+ addStringField(inputRowMeta, variables, keyField, name);
+ addStringField(inputRowMeta, variables, destinationField, name);
+ addStringField(inputRowMeta, variables, messageIdField, name);
+
+ String resolvedTimestamp = variables.resolve(timestampField);
+ if (StringUtils.isNotEmpty(resolvedTimestamp)) {
+ ValueMetaDate valueMeta = new ValueMetaDate(resolvedTimestamp);
+ valueMeta.setOrigin(name);
+ inputRowMeta.addValueMeta(valueMeta);
+ }
+ }
+
+ private void addStringField(
+ IRowMeta rowMeta, IVariables variables, String fieldName, String origin)
{
+ String resolved = variables.resolve(fieldName);
+ if (StringUtils.isEmpty(resolved)) {
+ return;
+ }
+ ValueMetaString valueMeta = new ValueMetaString(resolved);
+ valueMeta.setOrigin(origin);
+ rowMeta.addValueMeta(valueMeta);
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsDestinationType.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsDestinationType.java
new file mode 100644
index 0000000000..f1b8bc22ea
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsDestinationType.java
@@ -0,0 +1,35 @@
+/*
+ * 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.hop.pipeline.transforms.jms.consumer;
+
+/** Whether a destination is a point-to-point queue or a publish/subscribe
topic. */
+public enum JmsDestinationType {
+ QUEUE,
+ TOPIC;
+
+ public static JmsDestinationType of(String value) {
+ if (value != null) {
+ for (JmsDestinationType type : values()) {
+ if (type.name().equalsIgnoreCase(value.trim())) {
+ return type;
+ }
+ }
+ }
+ return QUEUE;
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducer.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducer.java
new file mode 100644
index 0000000000..41fc259a43
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducer.java
@@ -0,0 +1,132 @@
+/*
+ * 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.hop.pipeline.transforms.jms.producer;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.pipeline.Pipeline;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.transform.BaseTransform;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transforms.jms.shared.JmsConnection;
+import org.apache.hop.pipeline.transforms.jms.shared.JmsProvider;
+import org.apache.hop.pipeline.transforms.jms.shared.MessageQueueRecord;
+
+public class JmsProducer extends BaseTransform<JmsProducerMeta,
JmsProducerData> {
+
+ public JmsProducer(
+ TransformMeta transformMeta,
+ JmsProducerMeta meta,
+ JmsProducerData data,
+ int copyNr,
+ PipelineMeta pipelineMeta,
+ Pipeline pipeline) {
+ super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline);
+ }
+
+ @Override
+ public boolean init() {
+ if (!super.init()) {
+ return false;
+ }
+ try {
+ JmsConnection connection =
+ getMetadataProvider()
+ .getSerializer(JmsConnection.class)
+ .load(resolve(meta.getConnectionName()));
+ if (connection == null) {
+ logError(
+ "JMS connection '" + meta.getConnectionName() + "' could not be
found in the metadata");
+ return false;
+ }
+ if (StringUtils.isEmpty(meta.getBodyField())) {
+ logError("No field was selected to use as the message body");
+ return false;
+ }
+
+ data.provider =
+ new JmsProvider(
+ connection,
+ meta.getDestination(),
+ meta.isTopic(),
+ meta.isTransacted(),
+ null,
+ null,
+ getLogChannel());
+ data.provider.connect(this);
+ return true;
+ } catch (Exception e) {
+ logError("Unable to start the JMS producer", e);
+ return false;
+ }
+ }
+
+ @Override
+ public boolean processRow() throws HopException {
+ Object[] row = getRow();
+ if (row == null) {
+ setOutputDone();
+ return false;
+ }
+
+ if (first) {
+ first = false;
+ data.bodyFieldIndex =
getInputRowMeta().indexOfValue(resolve(meta.getBodyField()));
+ if (data.bodyFieldIndex < 0) {
+ throw new HopException(
+ "Body field '"
+ + resolve(meta.getBodyField())
+ + "' could not be found in the input row: "
+ + getInputRowMeta().getFieldNames().length
+ + " field(s) available");
+ }
+ String keyField = resolve(meta.getKeyField());
+ if (StringUtils.isNotEmpty(keyField)) {
+ data.keyFieldIndex = getInputRowMeta().indexOfValue(keyField);
+ if (data.keyFieldIndex < 0) {
+ throw new HopException(
+ "Key field '" + keyField + "' could not be found in the input
row");
+ }
+ }
+ }
+
+ MessageQueueRecord record = new MessageQueueRecord();
+ record.setBody(getInputRowMeta().getString(row, data.bodyFieldIndex));
+ if (data.keyFieldIndex >= 0) {
+ record.setKey(getInputRowMeta().getString(row, data.keyFieldIndex));
+ }
+ data.provider.send(record);
+
+ // The row is passed through unchanged so the transform can sit
mid-pipeline.
+ putRow(getInputRowMeta(), row);
+
+ if (checkFeedback(getLinesWritten()) && isBasic()) {
+ logBasic("Sent " + getLinesWritten() + " messages to the JMS
destination");
+ }
+ return true;
+ }
+
+ @Override
+ public void dispose() {
+ if (data.provider != null) {
+ data.provider.close();
+ data.provider = null;
+ }
+ super.dispose();
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducerData.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducerData.java
new file mode 100644
index 0000000000..fb2539a3c8
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducerData.java
@@ -0,0 +1,33 @@
+/*
+ * 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.hop.pipeline.transforms.jms.producer;
+
+import org.apache.hop.pipeline.transform.BaseTransformData;
+import org.apache.hop.pipeline.transform.ITransformData;
+import org.apache.hop.pipeline.transforms.jms.shared.IMessageQueueProvider;
+
+public class JmsProducerData extends BaseTransformData implements
ITransformData {
+
+ public IMessageQueueProvider provider;
+ public int bodyFieldIndex = -1;
+ public int keyFieldIndex = -1;
+
+ public JmsProducerData() {
+ super();
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducerDialog.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducerDialog.java
new file mode 100644
index 0000000000..b8e7538e01
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducerDialog.java
@@ -0,0 +1,201 @@
+/*
+ * 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.hop.pipeline.transforms.jms.producer;
+
+import org.apache.hop.core.Const;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.util.Utils;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.transforms.jms.consumer.JmsDestinationType;
+import org.apache.hop.pipeline.transforms.jms.shared.JmsDialogHelper;
+import org.apache.hop.ui.core.PropsUi;
+import org.apache.hop.ui.core.dialog.BaseDialog;
+import org.apache.hop.ui.core.dialog.ErrorDialog;
+import org.apache.hop.ui.core.widget.TextVar;
+import org.apache.hop.ui.pipeline.transform.BaseTransformDialog;
+import org.eclipse.swt.SWT;
+import org.eclipse.swt.custom.CCombo;
+import org.eclipse.swt.layout.FormAttachment;
+import org.eclipse.swt.layout.FormData;
+import org.eclipse.swt.layout.FormLayout;
+import org.eclipse.swt.widgets.Button;
+import org.eclipse.swt.widgets.Control;
+import org.eclipse.swt.widgets.Label;
+import org.eclipse.swt.widgets.Shell;
+import org.eclipse.swt.widgets.Text;
+
+public class JmsProducerDialog extends BaseTransformDialog {
+ private static final Class<?> PKG = JmsProducerMeta.class;
+
+ private final JmsProducerMeta input;
+
+ private CCombo wConnection;
+ private CCombo wDestinationType;
+ private TextVar wDestination;
+ private Button wTransacted;
+ private CCombo wBodyField;
+ private CCombo wKeyField;
+
+ public JmsProducerDialog(
+ Shell parent,
+ IVariables variables,
+ JmsProducerMeta transformMeta,
+ PipelineMeta pipelineMeta) {
+ super(parent, variables, transformMeta, pipelineMeta);
+ this.input = transformMeta;
+ }
+
+ @Override
+ public String open() {
+ Shell parent = getParent();
+ shell = new Shell(parent, SWT.DIALOG_TRIM | SWT.RESIZE | SWT.MIN |
SWT.MAX);
+ PropsUi.setLook(shell);
+ setShellImage(shell, input);
+
+ changed = input.hasChanged();
+
+ FormLayout formLayout = new FormLayout();
+ formLayout.marginWidth = PropsUi.getFormMargin();
+ formLayout.marginHeight = PropsUi.getFormMargin();
+ shell.setLayout(formLayout);
+ shell.setText(BaseMessages.getString(PKG,
"JmsProducerDialog.Shell.Title"));
+
+ int middle = props.getMiddlePct();
+ int margin = PropsUi.getMargin();
+
+ wlTransformName = new Label(shell, SWT.RIGHT);
+ wlTransformName.setText(BaseMessages.getString(PKG,
"System.Label.TransformName"));
+ PropsUi.setLook(wlTransformName);
+ fdlTransformName = new FormData();
+ fdlTransformName.left = new FormAttachment(0, 0);
+ fdlTransformName.top = new FormAttachment(0, margin);
+ fdlTransformName.right = new FormAttachment(middle, -margin);
+ wlTransformName.setLayoutData(fdlTransformName);
+ wTransformName = new Text(shell, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
+ wTransformName.setText(transformName);
+ PropsUi.setLook(wTransformName);
+ wTransformName.addModifyListener(e -> input.setChanged());
+ fdTransformName = new FormData();
+ fdTransformName.left = new FormAttachment(middle, 0);
+ fdTransformName.top = new FormAttachment(wlTransformName, 0, SWT.CENTER);
+ fdTransformName.right = new FormAttachment(100, 0);
+ wTransformName.setLayoutData(fdTransformName);
+ Control last = wTransformName;
+
+ wConnection =
+ JmsDialogHelper.labeledCombo(
+ shell, PKG, variables, middle, margin, last,
"JmsProducerDialog.Connection", true);
+
wConnection.setItems(JmsDialogHelper.listConnectionNames(metadataProvider));
+ last = wConnection;
+
+ wDestinationType =
+ JmsDialogHelper.labeledCombo(
+ shell, PKG, variables, middle, margin, last,
"JmsProducerDialog.DestinationType", true);
+ wDestinationType.setItems(
+ new String[] {JmsDestinationType.QUEUE.name(),
JmsDestinationType.TOPIC.name()});
+ last = wDestinationType;
+
+ wDestination =
+ JmsDialogHelper.labeledText(
+ shell, PKG, variables, middle, margin, last,
"JmsProducerDialog.Destination");
+ last = wDestination;
+
+ wTransacted =
+ JmsDialogHelper.labeledCheckbox(
+ shell, PKG, middle, margin, last, "JmsProducerDialog.Transacted");
+ last = wTransacted;
+
+ wBodyField =
+ JmsDialogHelper.labeledCombo(
+ shell, PKG, variables, middle, margin, last,
"JmsProducerDialog.BodyField", false);
+ last = wBodyField;
+
+ wKeyField =
+ JmsDialogHelper.labeledCombo(
+ shell, PKG, variables, middle, margin, last,
"JmsProducerDialog.KeyField", false);
+
+ String[] incomingFields = getIncomingFieldNames();
+ wBodyField.setItems(incomingFields);
+ wKeyField.setItems(incomingFields);
+
+ wOk = new Button(shell, SWT.PUSH);
+ wOk.setText(BaseMessages.getString(PKG, "System.Button.OK"));
+ wCancel = new Button(shell, SWT.PUSH);
+ wCancel.setText(BaseMessages.getString(PKG, "System.Button.Cancel"));
+ setButtonPositions(new Button[] {wOk, wCancel}, margin, wKeyField);
+
+ wOk.addListener(SWT.Selection, e -> ok());
+ wCancel.addListener(SWT.Selection, e -> cancel());
+
+ getData();
+ input.setChanged(changed);
+ BaseDialog.defaultShellHandling(shell, c -> ok(), c -> cancel());
+ return transformName;
+ }
+
+ private String[] getIncomingFieldNames() {
+ try {
+ IRowMeta rowMeta = pipelineMeta.getPrevTransformFields(variables,
transformName);
+ return rowMeta.getFieldNames();
+ } catch (Exception e) {
+ // The dialog must still open when the upstream row layout cannot be
resolved.
+ new ErrorDialog(
+ shell,
+ BaseMessages.getString(PKG, "JmsProducerDialog.FieldsError.Title"),
+ BaseMessages.getString(PKG, "JmsProducerDialog.FieldsError.Message"),
+ e);
+ return new String[0];
+ }
+ }
+
+ private void getData() {
+ wTransformName.setText(Const.NVL(transformName, ""));
+ wConnection.setText(Const.NVL(input.getConnectionName(), ""));
+ wDestinationType.setText(
+ Const.NVL(input.getDestinationType(),
JmsDestinationType.QUEUE.name()));
+ wDestination.setText(Const.NVL(input.getDestination(), ""));
+ wTransacted.setSelection(input.isTransacted());
+ wBodyField.setText(Const.NVL(input.getBodyField(), ""));
+ wKeyField.setText(Const.NVL(input.getKeyField(), ""));
+
+ wTransformName.selectAll();
+ wTransformName.setFocus();
+ }
+
+ private void cancel() {
+ transformName = null;
+ input.setChanged(changed);
+ dispose();
+ }
+
+ private void ok() {
+ if (Utils.isEmpty(wTransformName.getText())) {
+ return;
+ }
+ transformName = wTransformName.getText();
+ input.setConnectionName(wConnection.getText());
+ input.setDestinationType(wDestinationType.getText());
+ input.setDestination(wDestination.getText());
+ input.setTransacted(wTransacted.getSelection());
+ input.setBodyField(wBodyField.getText());
+ input.setKeyField(wKeyField.getText());
+ dispose();
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducerMeta.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducerMeta.java
new file mode 100644
index 0000000000..544fc3ffff
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/producer/JmsProducerMeta.java
@@ -0,0 +1,75 @@
+/*
+ * 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.hop.pipeline.transforms.jms.producer;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.hop.core.annotations.Transform;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+import org.apache.hop.pipeline.transform.BaseTransformMeta;
+import org.apache.hop.pipeline.transforms.jms.consumer.JmsDestinationType;
+
+/** Sends each incoming row to a JMS queue or topic as a text message. */
+@Transform(
+ id = "JmsProducer",
+ name = "i18n::JmsProducer.Name",
+ description = "i18n::JmsProducer.Description",
+ image = "jms-producer.svg",
+ categoryDescription =
"i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Streaming",
+ keywords = "i18n::JmsProducer.Keyword",
+ documentationUrl = "/pipeline/transforms/jms-producer.html")
+@Getter
+@Setter
+public class JmsProducerMeta extends BaseTransformMeta<JmsProducer,
JmsProducerData> {
+
+ /** Name of the JMS connection metadata object. */
+ @HopMetadataProperty(key = "connection", injectionKey = "CONNECTION")
+ private String connectionName;
+
+ @HopMetadataProperty(key = "destination", injectionKey = "DESTINATION")
+ private String destination;
+
+ /** QUEUE or TOPIC. */
+ @HopMetadataProperty(key = "destination_type", injectionKey =
"DESTINATION_TYPE")
+ private String destinationType = JmsDestinationType.QUEUE.name();
+
+ /** Commit each message in a JMS transaction instead of sending it outside
one. */
+ @HopMetadataProperty(key = "transacted", injectionKey = "TRANSACTED")
+ private boolean transacted;
+
+ /** Incoming field holding the message body. Required. */
+ @HopMetadataProperty(key = "body_field", injectionKey = "BODY_FIELD")
+ private String bodyField;
+
+ /** Optional incoming field holding the correlation id. */
+ @HopMetadataProperty(key = "key_field", injectionKey = "KEY_FIELD")
+ private String keyField;
+
+ public JmsProducerMeta() {
+ super();
+ }
+
+ @Override
+ public String getDialogClassName() {
+ return JmsProducerDialog.class.getName();
+ }
+
+ public boolean isTopic() {
+ return JmsDestinationType.TOPIC.name().equalsIgnoreCase(destinationType);
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/IMessageQueueProvider.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/IMessageQueueProvider.java
new file mode 100644
index 0000000000..119300dd6c
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/IMessageQueueProvider.java
@@ -0,0 +1,67 @@
+/*
+ * 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.hop.pipeline.transforms.jms.shared;
+
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.variables.IVariables;
+
+/**
+ * A message broker, seen from a Hop transform.
+ *
+ * <p>The operations are the ones a transform actually needs, independent of
broker: connect and
+ * disconnect, get the next message, read information about it, acknowledge
it, and send one. See <a
+ * href="https://github.com/apache/hop/issues/2653";>issue #2653</a>, where
this shape was proposed.
+ *
+ * <p>The interface deliberately lives inside this plugin for now rather than
being a Hop plugin
+ * type. It is exercised here by a single JMS implementation; promoting it to
a plugin type is worth
+ * doing once a second, structurally different provider (Kafka, native AMQP
1.0) has been written
+ * against it and the shape has been confirmed by more than one caller.
+ */
+public interface IMessageQueueProvider extends AutoCloseable {
+
+ /**
+ * Opens the connection and prepares the destination. Called once when the
transform starts.
+ *
+ * @param variables used to resolve any configuration that supports variables
+ */
+ void connect(IVariables variables) throws HopException;
+
+ /**
+ * Returns the next message, or null when {@code timeoutMs} elapses without
one arriving.
+ *
+ * <p>Returning null is normal, not an error: it is how a consumer notices
it should check whether
+ * the pipeline is stopping.
+ *
+ * @param timeoutMs how long to wait, in milliseconds
+ */
+ MessageQueueRecord receive(long timeoutMs) throws HopException;
+
+ /**
+ * Confirms that {@code record} has been processed, so the broker can drop
it.
+ *
+ * <p>A no-op for providers that acknowledge automatically.
+ */
+ void acknowledge(MessageQueueRecord record) throws HopException;
+
+ /** Publishes {@code record}. The destination on the record wins when it is
set. */
+ void send(MessageQueueRecord record) throws HopException;
+
+ /** Closes the connection. Must be safe to call more than once, and must not
throw. */
+ @Override
+ void close();
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsConnection.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsConnection.java
new file mode 100644
index 0000000000..7613470e0d
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsConnection.java
@@ -0,0 +1,99 @@
+/*
+ * 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.hop.pipeline.transforms.jms.shared;
+
+import java.io.Serializable;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.hop.metadata.api.HopMetadata;
+import org.apache.hop.metadata.api.HopMetadataBase;
+import org.apache.hop.metadata.api.HopMetadataCategory;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+import org.apache.hop.metadata.api.IHopMetadata;
+
+/**
+ * A reusable JMS broker connection, shared by the JMS consumer and producer
transforms.
+ *
+ * <p>Two ways to reach a broker:
+ *
+ * <ul>
+ * <li><b>Direct</b> — a broker URL handed to the bundled Apache ActiveMQ
Artemis client. Nothing
+ * else to install.
+ * <li><b>JNDI</b> — an initial context factory and a connection factory
name. This is how any
+ * other JMS 3.0 provider is used: put its client jar beside this plugin
and name its factory
+ * here.
+ * </ul>
+ */
+@Getter
+@Setter
+@HopMetadata(
+ key = "jms-connection",
+ name = "i18n::JmsConnection.Name",
+ description = "i18n::JmsConnection.Description",
+ image = "jms-connection.svg",
+ category = HopMetadataCategory.CONNECTIONS,
+ documentationUrl = "/metadata-types/jms-connection.html")
+public class JmsConnection extends HopMetadataBase implements Serializable,
IHopMetadata {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final String MODE_DIRECT = "DIRECT";
+ public static final String MODE_JNDI = "JNDI";
+
+ /** {@link #MODE_DIRECT} or {@link #MODE_JNDI}. */
+ @HopMetadataProperty(key = "mode")
+ private String mode = MODE_DIRECT;
+
+ /** Broker URL for DIRECT mode, e.g. {@code tcp://localhost:61616}. Supports
variables. */
+ @HopMetadataProperty(key = "broker_url")
+ private String brokerUrl;
+
+ /** JNDI initial context factory class name for JNDI mode. Supports
variables. */
+ @HopMetadataProperty(key = "initial_context_factory")
+ private String initialContextFactory;
+
+ /** JNDI provider URL for JNDI mode. Supports variables. */
+ @HopMetadataProperty(key = "provider_url")
+ private String providerUrl;
+
+ /** Name the connection factory is bound to in JNDI. Supports variables. */
+ @HopMetadataProperty(key = "connection_factory_name")
+ private String connectionFactoryName = "ConnectionFactory";
+
+ @HopMetadataProperty(key = "username")
+ private String username;
+
+ @HopMetadataProperty(key = "password", password = true)
+ private String password;
+
+ /** Optional JMS client id. Required by the spec for a durable topic
subscription. */
+ @HopMetadataProperty(key = "client_id")
+ private String clientId;
+
+ public JmsConnection() {
+ super();
+ }
+
+ public JmsConnection(String name) {
+ super(name);
+ }
+
+ public boolean isJndi() {
+ return MODE_JNDI.equalsIgnoreCase(mode);
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsConnectionEditor.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsConnectionEditor.java
new file mode 100644
index 0000000000..8e9bed38af
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsConnectionEditor.java
@@ -0,0 +1,218 @@
+/*
+ * 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.hop.pipeline.transforms.jms.shared;
+
+import org.apache.hop.core.Const;
+import org.apache.hop.core.gui.plugin.GuiPlugin;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.ui.core.PropsUi;
+import org.apache.hop.ui.core.metadata.MetadataEditor;
+import org.apache.hop.ui.core.metadata.MetadataManager;
+import org.apache.hop.ui.core.widget.PasswordTextVar;
+import org.apache.hop.ui.core.widget.TextVar;
+import org.apache.hop.ui.hopgui.HopGui;
+import org.eclipse.swt.SWT;
+import org.eclipse.swt.custom.CCombo;
+import org.eclipse.swt.layout.FormAttachment;
+import org.eclipse.swt.layout.FormData;
+import org.eclipse.swt.layout.FormLayout;
+import org.eclipse.swt.widgets.Composite;
+import org.eclipse.swt.widgets.Control;
+import org.eclipse.swt.widgets.Label;
+import org.eclipse.swt.widgets.Text;
+
+@GuiPlugin(description = "Editor for JMS connection metadata")
+public class JmsConnectionEditor extends MetadataEditor<JmsConnection> {
+
+ private static final Class<?> PKG = JmsConnection.class;
+
+ private Text wName;
+ private CCombo wMode;
+ private TextVar wBrokerUrl;
+ private TextVar wInitialContextFactory;
+ private TextVar wProviderUrl;
+ private TextVar wConnectionFactoryName;
+ private TextVar wUsername;
+ private TextVar wPassword;
+ private TextVar wClientId;
+
+ public JmsConnectionEditor(
+ HopGui hopGui, MetadataManager<JmsConnection> manager, JmsConnection
metadata) {
+ super(hopGui, manager, metadata);
+ }
+
+ @Override
+ public void createControl(Composite parent) {
+ PropsUi props = PropsUi.getInstance();
+ int middle = props.getMiddlePct();
+ int margin = PropsUi.getMargin() + 2;
+
+ Composite composite = new Composite(parent, SWT.NONE);
+ PropsUi.setLook(composite);
+ FormLayout layout = new FormLayout();
+ layout.marginWidth = 0;
+ layout.marginHeight = 0;
+ composite.setLayout(layout);
+ FormData fdComposite = new FormData();
+ fdComposite.left = new FormAttachment(0, 0);
+ fdComposite.top = new FormAttachment(0, 0);
+ fdComposite.right = new FormAttachment(100, 0);
+ composite.setLayoutData(fdComposite);
+
+ Label wlName = new Label(composite, SWT.RIGHT);
+ wlName.setText(BaseMessages.getString(PKG, "JmsConnectionEditor.Name"));
+ PropsUi.setLook(wlName);
+ FormData fdlName = new FormData();
+ fdlName.left = new FormAttachment(0, 0);
+ fdlName.top = new FormAttachment(0, margin);
+ fdlName.right = new FormAttachment(middle, -margin);
+ wlName.setLayoutData(fdlName);
+ wName = new Text(composite, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
+ PropsUi.setLook(wName);
+ FormData fdName = new FormData();
+ fdName.left = new FormAttachment(middle, 0);
+ fdName.top = new FormAttachment(wlName, 0, SWT.CENTER);
+ fdName.right = new FormAttachment(100, 0);
+ wName.setLayoutData(fdName);
+ Control last = wName;
+
+ wMode =
+ (CCombo) labeled(composite, middle, margin, last,
"JmsConnectionEditor.Mode", Kind.COMBO);
+ wMode.setItems(new String[] {JmsConnection.MODE_DIRECT,
JmsConnection.MODE_JNDI});
+ last = wMode;
+
+ wBrokerUrl = text(composite, middle, margin, last,
"JmsConnectionEditor.BrokerUrl");
+ last = wBrokerUrl;
+ wInitialContextFactory =
+ text(composite, middle, margin, last,
"JmsConnectionEditor.InitialContextFactory");
+ last = wInitialContextFactory;
+ wProviderUrl = text(composite, middle, margin, last,
"JmsConnectionEditor.ProviderUrl");
+ last = wProviderUrl;
+ wConnectionFactoryName =
+ text(composite, middle, margin, last,
"JmsConnectionEditor.ConnectionFactoryName");
+ last = wConnectionFactoryName;
+ wUsername = text(composite, middle, margin, last,
"JmsConnectionEditor.Username");
+ last = wUsername;
+
+ Label wlPassword = new Label(composite, SWT.RIGHT);
+ wlPassword.setText(BaseMessages.getString(PKG,
"JmsConnectionEditor.Password"));
+ PropsUi.setLook(wlPassword);
+ FormData fdlPassword = new FormData();
+ fdlPassword.left = new FormAttachment(0, 0);
+ fdlPassword.top = new FormAttachment(last, margin);
+ fdlPassword.right = new FormAttachment(middle, -margin);
+ wlPassword.setLayoutData(fdlPassword);
+ wPassword =
+ new PasswordTextVar(manager.getVariables(), composite, SWT.SINGLE |
SWT.LEFT | SWT.BORDER);
+ PropsUi.setLook(wPassword);
+ FormData fdPassword = new FormData();
+ fdPassword.left = new FormAttachment(middle, 0);
+ fdPassword.top = new FormAttachment(wlPassword, 0, SWT.CENTER);
+ fdPassword.right = new FormAttachment(100, 0);
+ wPassword.setLayoutData(fdPassword);
+ last = wPassword;
+
+ wClientId = text(composite, middle, margin, last,
"JmsConnectionEditor.ClientId");
+
+ setWidgetsContent();
+ addModifyListeners();
+ }
+
+ private enum Kind {
+ TEXT,
+ COMBO
+ }
+
+ private Control labeled(
+ Composite composite, int middle, int margin, Control last, String
labelKey, Kind kind) {
+ Label label = new Label(composite, SWT.RIGHT);
+ label.setText(BaseMessages.getString(PKG, labelKey));
+ PropsUi.setLook(label);
+ FormData fdLabel = new FormData();
+ fdLabel.left = new FormAttachment(0, 0);
+ fdLabel.top = new FormAttachment(last, margin);
+ fdLabel.right = new FormAttachment(middle, -margin);
+ label.setLayoutData(fdLabel);
+
+ Control control =
+ kind == Kind.COMBO
+ ? new CCombo(composite, SWT.BORDER | SWT.READ_ONLY)
+ : new TextVar(manager.getVariables(), composite, SWT.SINGLE |
SWT.LEFT | SWT.BORDER);
+ PropsUi.setLook(control);
+ FormData fdControl = new FormData();
+ fdControl.left = new FormAttachment(middle, 0);
+ fdControl.top = new FormAttachment(label, 0, SWT.CENTER);
+ fdControl.right = new FormAttachment(100, 0);
+ control.setLayoutData(fdControl);
+
+ String tooltip = BaseMessages.getString(PKG, labelKey + ".Tooltip");
+ if (tooltip != null && !tooltip.startsWith("!")) {
+ control.setToolTipText(tooltip);
+ label.setToolTipText(tooltip);
+ }
+ return control;
+ }
+
+ private TextVar text(Composite composite, int middle, int margin, Control
last, String labelKey) {
+ return (TextVar) labeled(composite, middle, margin, last, labelKey,
Kind.TEXT);
+ }
+
+ private void addModifyListeners() {
+ wName.addListener(SWT.Modify, e -> setChanged());
+ wMode.addListener(SWT.Modify, e -> setChanged());
+ for (TextVar field :
+ new TextVar[] {
+ wBrokerUrl,
+ wInitialContextFactory,
+ wProviderUrl,
+ wConnectionFactoryName,
+ wUsername,
+ wPassword,
+ wClientId
+ }) {
+ field.addListener(SWT.Modify, e -> setChanged());
+ }
+ }
+
+ @Override
+ public void setWidgetsContent() {
+ JmsConnection meta = getMetadata();
+ wName.setText(Const.NVL(meta.getName(), ""));
+ wMode.setText(Const.NVL(meta.getMode(), JmsConnection.MODE_DIRECT));
+ wBrokerUrl.setText(Const.NVL(meta.getBrokerUrl(), ""));
+ wInitialContextFactory.setText(Const.NVL(meta.getInitialContextFactory(),
""));
+ wProviderUrl.setText(Const.NVL(meta.getProviderUrl(), ""));
+ wConnectionFactoryName.setText(Const.NVL(meta.getConnectionFactoryName(),
""));
+ wUsername.setText(Const.NVL(meta.getUsername(), ""));
+ wPassword.setText(Const.NVL(meta.getPassword(), ""));
+ wClientId.setText(Const.NVL(meta.getClientId(), ""));
+ }
+
+ @Override
+ public void getWidgetsContent(JmsConnection meta) {
+ meta.setName(wName.getText());
+ meta.setMode(wMode.getText());
+ meta.setBrokerUrl(wBrokerUrl.getText());
+ meta.setInitialContextFactory(wInitialContextFactory.getText());
+ meta.setProviderUrl(wProviderUrl.getText());
+ meta.setConnectionFactoryName(wConnectionFactoryName.getText());
+ meta.setUsername(wUsername.getText());
+ meta.setPassword(wPassword.getText());
+ meta.setClientId(wClientId.getText());
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsDialogHelper.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsDialogHelper.java
new file mode 100644
index 0000000000..5223457139
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsDialogHelper.java
@@ -0,0 +1,123 @@
+/*
+ * 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.hop.pipeline.transforms.jms.shared;
+
+import java.util.List;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.ui.core.PropsUi;
+import org.apache.hop.ui.core.widget.TextVar;
+import org.eclipse.swt.SWT;
+import org.eclipse.swt.custom.CCombo;
+import org.eclipse.swt.layout.FormAttachment;
+import org.eclipse.swt.layout.FormData;
+import org.eclipse.swt.widgets.Button;
+import org.eclipse.swt.widgets.Composite;
+import org.eclipse.swt.widgets.Control;
+import org.eclipse.swt.widgets.Label;
+
+/** Widget plumbing shared by the JMS consumer and producer dialogs. */
+public final class JmsDialogHelper {
+
+ private JmsDialogHelper() {}
+
+ /** Names of the JMS connections in the metadata, for a connection combo. */
+ public static String[] listConnectionNames(IHopMetadataProvider
metadataProvider) {
+ try {
+ List<String> names =
metadataProvider.getSerializer(JmsConnection.class).listObjectNames();
+ return names.toArray(new String[0]);
+ } catch (HopException e) {
+ // A dialog must still open when the metadata cannot be read; the combo
is simply empty.
+ return new String[0];
+ }
+ }
+
+ public static CCombo labeledCombo(
+ Composite shell,
+ Class<?> pkg,
+ IVariables variables,
+ int middle,
+ int margin,
+ Control last,
+ String labelKey,
+ boolean readOnly) {
+ Label label = label(shell, pkg, middle, margin, last, labelKey);
+ CCombo combo = new CCombo(shell, SWT.BORDER | (readOnly ? SWT.READ_ONLY :
SWT.NONE));
+ PropsUi.setLook(combo);
+ combo.setLayoutData(rightOf(label, middle));
+ applyTooltip(pkg, labelKey, label, combo);
+ return combo;
+ }
+
+ public static TextVar labeledText(
+ Composite shell,
+ Class<?> pkg,
+ IVariables variables,
+ int middle,
+ int margin,
+ Control last,
+ String labelKey) {
+ Label label = label(shell, pkg, middle, margin, last, labelKey);
+ TextVar text = new TextVar(variables, shell, SWT.SINGLE | SWT.LEFT |
SWT.BORDER);
+ PropsUi.setLook(text);
+ text.setLayoutData(rightOf(label, middle));
+ applyTooltip(pkg, labelKey, label, text);
+ return text;
+ }
+
+ public static Button labeledCheckbox(
+ Composite shell, Class<?> pkg, int middle, int margin, Control last,
String labelKey) {
+ Label label = label(shell, pkg, middle, margin, last, labelKey);
+ Button button = new Button(shell, SWT.CHECK);
+ PropsUi.setLook(button);
+ button.setLayoutData(rightOf(label, middle));
+ applyTooltip(pkg, labelKey, label, button);
+ return button;
+ }
+
+ private static Label label(
+ Composite shell, Class<?> pkg, int middle, int margin, Control last,
String labelKey) {
+ Label label = new Label(shell, SWT.RIGHT);
+ label.setText(BaseMessages.getString(pkg, labelKey));
+ PropsUi.setLook(label);
+ FormData fdLabel = new FormData();
+ fdLabel.left = new FormAttachment(0, 0);
+ fdLabel.top = new FormAttachment(last, margin);
+ fdLabel.right = new FormAttachment(middle, -margin);
+ label.setLayoutData(fdLabel);
+ return label;
+ }
+
+ private static FormData rightOf(Label label, int middle) {
+ FormData fd = new FormData();
+ fd.left = new FormAttachment(middle, 0);
+ fd.top = new FormAttachment(label, 0, SWT.CENTER);
+ fd.right = new FormAttachment(100, 0);
+ return fd;
+ }
+
+ private static void applyTooltip(Class<?> pkg, String labelKey, Label label,
Control control) {
+ String tooltip = BaseMessages.getString(pkg, labelKey + ".Tooltip");
+ if (tooltip != null && !tooltip.startsWith("!")) {
+ label.setToolTipText(tooltip);
+ control.setToolTipText(tooltip);
+ }
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsProvider.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsProvider.java
new file mode 100644
index 0000000000..a15a45420b
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/JmsProvider.java
@@ -0,0 +1,286 @@
+/*
+ * 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.hop.pipeline.transforms.jms.shared;
+
+import jakarta.jms.Connection;
+import jakarta.jms.ConnectionFactory;
+import jakarta.jms.Destination;
+import jakarta.jms.JMSException;
+import jakarta.jms.Message;
+import jakarta.jms.MessageConsumer;
+import jakarta.jms.MessageProducer;
+import jakarta.jms.Session;
+import jakarta.jms.TextMessage;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Enumeration;
+import java.util.Hashtable;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.variables.IVariables;
+
+/**
+ * {@link IMessageQueueProvider} backed by JMS 3.0.
+ *
+ * <p>Not thread safe: a JMS {@link Session} may only be used by one thread,
so one provider belongs
+ * to one transform copy.
+ */
+public class JmsProvider implements IMessageQueueProvider {
+
+ private final JmsConnection connectionMeta;
+ private final String destinationName;
+ private final boolean topic;
+ private final boolean transacted;
+ private final String messageSelector;
+ private final String durableSubscriptionName;
+ private final ILogChannel log;
+
+ private Connection connection;
+ private Session session;
+ private MessageConsumer consumer;
+ private MessageProducer producer;
+ private Destination destination;
+
+ public JmsProvider(
+ JmsConnection connectionMeta,
+ String destinationName,
+ boolean topic,
+ boolean transacted,
+ String messageSelector,
+ String durableSubscriptionName,
+ ILogChannel log) {
+ this.connectionMeta = connectionMeta;
+ this.destinationName = destinationName;
+ this.topic = topic;
+ this.transacted = transacted;
+ this.messageSelector = messageSelector;
+ this.durableSubscriptionName = durableSubscriptionName;
+ this.log = log;
+ }
+
+ @Override
+ public void connect(IVariables variables) throws HopException {
+ if (connectionMeta == null) {
+ throw new HopException("No JMS connection was configured");
+ }
+ String resolvedDestination = variables.resolve(destinationName);
+ if (StringUtils.isBlank(resolvedDestination)) {
+ throw new HopException("No JMS destination (queue or topic) was
configured");
+ }
+
+ try {
+ ConnectionFactory factory = createConnectionFactory(variables);
+
+ String user = variables.resolve(connectionMeta.getUsername());
+ String password = variables.resolve(connectionMeta.getPassword());
+ connection =
+ StringUtils.isEmpty(user)
+ ? factory.createConnection()
+ : factory.createConnection(user, password);
+
+ String clientId = variables.resolve(connectionMeta.getClientId());
+ if (StringUtils.isNotEmpty(clientId)) {
+ connection.setClientID(clientId);
+ }
+
+ // AUTO_ACKNOWLEDGE would confirm the message before the pipeline has
processed the row,
+ // so a crash mid-pipeline would lose it. CLIENT_ACKNOWLEDGE lets the
consumer confirm
+ // only once the row is on its way.
+ session =
+ connection.createSession(
+ transacted, transacted ? Session.SESSION_TRANSACTED :
Session.CLIENT_ACKNOWLEDGE);
+
+ destination =
+ topic
+ ? session.createTopic(resolvedDestination)
+ : session.createQueue(resolvedDestination);
+
+ connection.start();
+
+ if (log != null) {
+ log.logBasic(
+ "Connected to JMS "
+ + (topic ? "topic" : "queue")
+ + " '"
+ + resolvedDestination
+ + "'"
+ + (transacted ? " (transacted)" : ""));
+ }
+ } catch (Exception e) {
+ close();
+ throw new HopException(
+ "Unable to connect to the JMS broker for destination '" +
resolvedDestination + "'", e);
+ }
+ }
+
+ private ConnectionFactory createConnectionFactory(IVariables variables)
throws Exception {
+ if (connectionMeta.isJndi()) {
+ Hashtable<String, String> environment = new Hashtable<>();
+ environment.put(
+ Context.INITIAL_CONTEXT_FACTORY,
+ variables.resolve(connectionMeta.getInitialContextFactory()));
+ String providerUrl = variables.resolve(connectionMeta.getProviderUrl());
+ if (StringUtils.isNotEmpty(providerUrl)) {
+ environment.put(Context.PROVIDER_URL, providerUrl);
+ }
+ InitialContext context = new InitialContext(environment);
+ try {
+ String factoryName =
variables.resolve(connectionMeta.getConnectionFactoryName());
+ return (ConnectionFactory) context.lookup(factoryName);
+ } finally {
+ context.close();
+ }
+ }
+ // Direct mode: the bundled Artemis client, reached reflectively so the
plugin still loads
+ // when the connection is JNDI-only and the Artemis jar has been removed.
+ String brokerUrl = variables.resolve(connectionMeta.getBrokerUrl());
+ if (StringUtils.isBlank(brokerUrl)) {
+ throw new HopException(
+ "No broker URL was configured on JMS connection '" +
connectionMeta.getName() + "'");
+ }
+ Class<?> factoryClass =
+
Class.forName("org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory");
+ return (ConnectionFactory)
factoryClass.getConstructor(String.class).newInstance(brokerUrl);
+ }
+
+ @Override
+ public MessageQueueRecord receive(long timeoutMs) throws HopException {
+ try {
+ if (consumer == null) {
+ consumer = createConsumer();
+ }
+ Message message = consumer.receive(timeoutMs);
+ if (message == null) {
+ return null; // timeout: the caller decides whether to keep waiting
+ }
+ return toRecord(message);
+ } catch (JMSException e) {
+ throw new HopException("Error receiving a message from the JMS
destination", e);
+ }
+ }
+
+ private MessageConsumer createConsumer() throws JMSException {
+ String selector = StringUtils.trimToNull(messageSelector);
+ if (topic && StringUtils.isNotEmpty(durableSubscriptionName)) {
+ return session.createDurableSubscriber(
+ (jakarta.jms.Topic) destination, durableSubscriptionName, selector,
false);
+ }
+ return selector == null
+ ? session.createConsumer(destination)
+ : session.createConsumer(destination, selector);
+ }
+
+ /** Maps a JMS message onto the broker-independent record. Visible for
testing. */
+ static MessageQueueRecord toRecord(Message message) throws JMSException {
+ MessageQueueRecord record = new MessageQueueRecord();
+ record.setMessageId(message.getJMSMessageID());
+ record.setKey(message.getJMSCorrelationID());
+ record.setHandle(message);
+ if (message.getJMSDestination() != null) {
+ record.setDestination(message.getJMSDestination().toString());
+ }
+ if (message.getJMSTimestamp() > 0) {
+ record.setTimestamp(new Date(message.getJMSTimestamp()));
+ }
+ if (message instanceof TextMessage textMessage) {
+ record.setBody(textMessage.getText());
+ } else {
+ // Anything that is not a TextMessage still yields its properties and
identifiers; the body
+ // is left null rather than guessing at an encoding.
+ record.setBody(null);
+ }
+
+ Enumeration<?> names = message.getPropertyNames();
+ if (names != null) {
+ for (String name : Collections.list((Enumeration<String>) names)) {
+ Object value = message.getObjectProperty(name);
+ record.property(name, value == null ? null : String.valueOf(value));
+ }
+ }
+ return record;
+ }
+
+ @Override
+ public void acknowledge(MessageQueueRecord record) throws HopException {
+ try {
+ if (transacted) {
+ session.commit();
+ return;
+ }
+ if (record != null && record.getHandle() instanceof Message message) {
+ message.acknowledge();
+ }
+ } catch (JMSException e) {
+ throw new HopException("Error acknowledging a JMS message", e);
+ }
+ }
+
+ @Override
+ public void send(MessageQueueRecord record) throws HopException {
+ try {
+ if (producer == null) {
+ producer = session.createProducer(destination);
+ }
+ TextMessage message = session.createTextMessage(record.getBody());
+ if (StringUtils.isNotEmpty(record.getKey())) {
+ message.setJMSCorrelationID(record.getKey());
+ }
+ for (var property : record.getProperties().entrySet()) {
+ if (property.getValue() != null) {
+ message.setStringProperty(property.getKey(), property.getValue());
+ }
+ }
+ producer.send(message);
+ if (transacted) {
+ session.commit();
+ }
+ } catch (JMSException e) {
+ throw new HopException("Error sending a message to the JMS destination",
e);
+ }
+ }
+
+ @Override
+ public void close() {
+ // Closing the connection closes its sessions, consumers and producers,
but each is closed
+ // explicitly so a failure in one does not leave the others open.
+ closeQuietly(consumer);
+ closeQuietly(producer);
+ closeQuietly(session);
+ closeQuietly(connection);
+ consumer = null;
+ producer = null;
+ session = null;
+ connection = null;
+ }
+
+ private void closeQuietly(AutoCloseable closeable) {
+ if (closeable == null) {
+ return;
+ }
+ try {
+ closeable.close();
+ } catch (Exception e) {
+ if (log != null) {
+ log.logDebug("Ignoring error while closing a JMS resource: " +
e.getMessage());
+ }
+ }
+ }
+}
diff --git
a/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/MessageQueueRecord.java
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/MessageQueueRecord.java
new file mode 100644
index 0000000000..7d7c3d147c
--- /dev/null
+++
b/plugins/transforms/jms/src/main/java/org/apache/hop/pipeline/transforms/jms/shared/MessageQueueRecord.java
@@ -0,0 +1,72 @@
+/*
+ * 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.hop.pipeline.transforms.jms.shared;
+
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * One message, independent of the broker it came from.
+ *
+ * <p>This is the "information about the message" half of the provider
contract: a key, the body,
+ * the destination it belongs to, timestamps, and free-form properties.
Broker-specific concepts
+ * that do not generalise stay behind {@link IMessageQueueProvider}.
+ */
+@Getter
+@Setter
+public class MessageQueueRecord {
+
+ /** Provider-assigned identifier, e.g. the JMS message id. Null when the
provider has none. */
+ private String messageId;
+
+ /**
+ * Application-level key used to relate messages, e.g. the JMS correlation
id. This is the field a
+ * Kafka-style key maps onto.
+ */
+ private String key;
+
+ /** Queue or topic name. */
+ private String destination;
+
+ /** Message body as text. */
+ private String body;
+
+ /** Broker timestamp, null when not supplied. */
+ private Date timestamp;
+
+ /** Provider-specific handle used by {@link
IMessageQueueProvider#acknowledge}. */
+ private Object handle;
+
+ /** Free-form message properties/headers, in encounter order. */
+ private final Map<String, String> properties = new LinkedHashMap<>();
+
+ public MessageQueueRecord() {}
+
+ public MessageQueueRecord(String destination, String body) {
+ this.destination = destination;
+ this.body = body;
+ }
+
+ public MessageQueueRecord property(String name, String value) {
+ properties.put(name, value);
+ return this;
+ }
+}
diff --git a/plugins/transforms/jms/src/main/resources/dependencies.xml
b/plugins/transforms/jms/src/main/resources/dependencies.xml
new file mode 100644
index 0000000000..8250e5ed34
--- /dev/null
+++ b/plugins/transforms/jms/src/main/resources/dependencies.xml
@@ -0,0 +1,20 @@
+<!--
+ ~ 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.
+ ~
+ -->
+
+<dependencies>
+</dependencies>
diff --git a/plugins/transforms/jms/src/main/resources/jms-connection.svg
b/plugins/transforms/jms/src/main/resources/jms-connection.svg
new file mode 100644
index 0000000000..6fa5853939
--- /dev/null
+++ b/plugins/transforms/jms/src/main/resources/jms-connection.svg
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="utf-8"?>
+<svg xmlns="http://www.w3.org/2000/svg"; width="24" height="24">
+ <path d="M 2,6 H 22 V 18 H 2 Z M 4,8 v 8 h 4 V 8 Z M 10,8 v 8 h 4 V 8 Z M
16,8 v 8 h 4 V 8 Z"
+ style="fill:#0e3a5a;fill-opacity:1;fill-rule:evenodd"/>
+</svg>
diff --git a/plugins/transforms/jms/src/main/resources/jms-consumer.svg
b/plugins/transforms/jms/src/main/resources/jms-consumer.svg
new file mode 100644
index 0000000000..6fa5853939
--- /dev/null
+++ b/plugins/transforms/jms/src/main/resources/jms-consumer.svg
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="utf-8"?>
+<svg xmlns="http://www.w3.org/2000/svg"; width="24" height="24">
+ <path d="M 2,6 H 22 V 18 H 2 Z M 4,8 v 8 h 4 V 8 Z M 10,8 v 8 h 4 V 8 Z M
16,8 v 8 h 4 V 8 Z"
+ style="fill:#0e3a5a;fill-opacity:1;fill-rule:evenodd"/>
+</svg>
diff --git a/plugins/transforms/jms/src/main/resources/jms-producer.svg
b/plugins/transforms/jms/src/main/resources/jms-producer.svg
new file mode 100644
index 0000000000..6fa5853939
--- /dev/null
+++ b/plugins/transforms/jms/src/main/resources/jms-producer.svg
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="utf-8"?>
+<svg xmlns="http://www.w3.org/2000/svg"; width="24" height="24">
+ <path d="M 2,6 H 22 V 18 H 2 Z M 4,8 v 8 h 4 V 8 Z M 10,8 v 8 h 4 V 8 Z M
16,8 v 8 h 4 V 8 Z"
+ style="fill:#0e3a5a;fill-opacity:1;fill-rule:evenodd"/>
+</svg>
diff --git
a/plugins/transforms/jms/src/main/resources/org/apache/hop/pipeline/transforms/jms/consumer/messages/messages_en_US.properties
b/plugins/transforms/jms/src/main/resources/org/apache/hop/pipeline/transforms/jms/consumer/messages/messages_en_US.properties
new file mode 100644
index 0000000000..1b512ffb06
--- /dev/null
+++
b/plugins/transforms/jms/src/main/resources/org/apache/hop/pipeline/transforms/jms/consumer/messages/messages_en_US.properties
@@ -0,0 +1,39 @@
+#
+# 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.
+#
+
+JmsConsumer.Name=JMS consumer
+JmsConsumer.Description=Read messages from a JMS queue or topic (ActiveMQ
Artemis, ActiveMQ Classic, RabbitMQ via JMS, or any JMS provider through JNDI)
+JmsConsumer.Keyword=jms,queue,topic,amqp,activemq,artemis,rabbitmq,message,consumer,subscribe
+
+JmsConsumerDialog.Shell.Title=JMS consumer
+JmsConsumerDialog.Connection=JMS connection
+JmsConsumerDialog.DestinationType=Destination type
+JmsConsumerDialog.Destination=Queue / topic name
+JmsConsumerDialog.MessageSelector=Message selector
+JmsConsumerDialog.MessageSelector.Tooltip=Optional JMS selector expression
evaluated by the broker, e.g. priority > 5 AND region = 'EU'
+JmsConsumerDialog.DurableSubscription=Durable subscription name
+JmsConsumerDialog.DurableSubscription.Tooltip=Topics only. Requires a client
id on the JMS connection.
+JmsConsumerDialog.Transacted=Use a JMS transaction
+JmsConsumerDialog.Transacted.Tooltip=Commit each message in a transaction
instead of acknowledging it individually.
+JmsConsumerDialog.MaxMessages=Maximum messages (0 = no limit)
+JmsConsumerDialog.ReceiveTimeout=Receive timeout (ms)
+JmsConsumerDialog.ReceiveTimeout.Tooltip=Stop when no message arrives within
this window. The destination is then treated as drained.
+JmsConsumerDialog.BodyField=Message body field
+JmsConsumerDialog.KeyField=Correlation id field
+JmsConsumerDialog.DestinationField=Destination field
+JmsConsumerDialog.MessageIdField=Message id field
+JmsConsumerDialog.TimestampField=Timestamp field
diff --git
a/plugins/transforms/jms/src/main/resources/org/apache/hop/pipeline/transforms/jms/producer/messages/messages_en_US.properties
b/plugins/transforms/jms/src/main/resources/org/apache/hop/pipeline/transforms/jms/producer/messages/messages_en_US.properties
new file mode 100644
index 0000000000..7ac1e721b0
--- /dev/null
+++
b/plugins/transforms/jms/src/main/resources/org/apache/hop/pipeline/transforms/jms/producer/messages/messages_en_US.properties
@@ -0,0 +1,30 @@
+#
+# 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.
+#
+
+JmsProducer.Name=JMS producer
+JmsProducer.Description=Send rows to a JMS queue or topic as text messages
+JmsProducer.Keyword=jms,queue,topic,amqp,activemq,artemis,rabbitmq,message,producer,publish
+
+JmsProducerDialog.Shell.Title=JMS producer
+JmsProducerDialog.Connection=JMS connection
+JmsProducerDialog.DestinationType=Destination type
+JmsProducerDialog.Destination=Queue / topic name
+JmsProducerDialog.Transacted=Use a JMS transaction
+JmsProducerDialog.BodyField=Message body field
+JmsProducerDialog.KeyField=Correlation id field (optional)
+JmsProducerDialog.FieldsError.Title=Error
+JmsProducerDialog.FieldsError.Message=Unable to read the fields coming into
this transform
diff --git
a/plugins/transforms/jms/src/main/resources/org/apache/hop/pipeline/transforms/jms/shared/messages/messages_en_US.properties
b/plugins/transforms/jms/src/main/resources/org/apache/hop/pipeline/transforms/jms/shared/messages/messages_en_US.properties
new file mode 100644
index 0000000000..2260501a85
--- /dev/null
+++
b/plugins/transforms/jms/src/main/resources/org/apache/hop/pipeline/transforms/jms/shared/messages/messages_en_US.properties
@@ -0,0 +1,32 @@
+#
+# 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.
+#
+
+JmsConnection.Name=JMS connection
+JmsConnection.Description=Connection to a JMS broker, shared by the JMS
consumer and producer transforms
+
+JmsConnectionEditor.Name=Connection name
+JmsConnectionEditor.Mode=Mode
+JmsConnectionEditor.Mode.Tooltip=DIRECT uses the bundled ActiveMQ Artemis
client. JNDI looks the connection factory up, for any other JMS provider.
+JmsConnectionEditor.BrokerUrl=Broker URL (DIRECT)
+JmsConnectionEditor.BrokerUrl.Tooltip=For example tcp\://localhost\:61616
+JmsConnectionEditor.InitialContextFactory=Initial context factory (JNDI)
+JmsConnectionEditor.ProviderUrl=Provider URL (JNDI)
+JmsConnectionEditor.ConnectionFactoryName=Connection factory name (JNDI)
+JmsConnectionEditor.Username=Username
+JmsConnectionEditor.Password=Password
+JmsConnectionEditor.ClientId=Client id
+JmsConnectionEditor.ClientId.Tooltip=Required by the JMS specification for a
durable topic subscription.
diff --git a/plugins/transforms/jms/src/main/resources/version.xml
b/plugins/transforms/jms/src/main/resources/version.xml
new file mode 100644
index 0000000000..36ab20e22e
--- /dev/null
+++ b/plugins/transforms/jms/src/main/resources/version.xml
@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ ~
+ -->
+
+<version>${project.version}</version>
diff --git
a/plugins/transforms/jms/src/test/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerMetaTest.java
b/plugins/transforms/jms/src/test/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerMetaTest.java
new file mode 100644
index 0000000000..c4b97f6da4
--- /dev/null
+++
b/plugins/transforms/jms/src/test/java/org/apache/hop/pipeline/transforms/jms/consumer/JmsConsumerMetaTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.hop.pipeline.transforms.jms.consumer;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.core.variables.Variables;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class JmsConsumerMetaTest {
+
+ @BeforeAll
+ static void init() throws Exception {
+ HopEnvironment.init();
+ }
+
+ @Test
+ void onlyNamedFieldsAppearInTheOutputRow() throws Exception {
+ JmsConsumerMeta meta = new JmsConsumerMeta();
+ meta.setBodyField("message");
+ meta.setKeyField(""); // left out
+ meta.setDestinationField("queue");
+ meta.setMessageIdField("");
+ meta.setTimestampField("sent_at");
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("discarded"));
+ meta.getFields(rowMeta, "jms", null, null, new Variables(), null);
+
+ assertArrayEquals(new String[] {"message", "queue", "sent_at"},
rowMeta.getFieldNames());
+ assertEquals(IValueMeta.TYPE_STRING, rowMeta.getValueMeta(0).getType());
+ assertEquals(IValueMeta.TYPE_STRING, rowMeta.getValueMeta(1).getType());
+ assertEquals(
+ IValueMeta.TYPE_DATE, rowMeta.getValueMeta(2).getType(), "the
timestamp must be a Date");
+ }
+
+ @Test
+ void fieldNamesResolveVariables() throws Exception {
+ JmsConsumerMeta meta = new JmsConsumerMeta();
+ meta.setBodyField("${BODY_FIELD}");
+ meta.setKeyField("");
+ meta.setDestinationField("");
+ meta.setMessageIdField("");
+ meta.setTimestampField("");
+
+ IVariables variables = new Variables();
+ variables.setVariable("BODY_FIELD", "payload");
+
+ IRowMeta rowMeta = new RowMeta();
+ meta.getFields(rowMeta, "jms", null, null, variables, null);
+
+ assertArrayEquals(new String[] {"payload"}, rowMeta.getFieldNames());
+ }
+
+ @Test
+ void outputRowIsEmptyWhenNoFieldIsNamed() throws Exception {
+ JmsConsumerMeta meta = new JmsConsumerMeta();
+ meta.setBodyField("");
+ meta.setKeyField("");
+ meta.setDestinationField("");
+ meta.setMessageIdField("");
+ meta.setTimestampField("");
+
+ IRowMeta rowMeta = new RowMeta();
+ meta.getFields(rowMeta, "jms", null, null, new Variables(), null);
+
+ assertEquals(0, rowMeta.size());
+ }
+
+ @Test
+ void destinationTypeDrivesTopicMode() {
+ JmsConsumerMeta meta = new JmsConsumerMeta();
+ assertFalse(meta.isTopic(), "queue is the default");
+
+ meta.setDestinationType("TOPIC");
+ assertTrue(meta.isTopic());
+
+ meta.setDestinationType("topic");
+ assertTrue(meta.isTopic(), "the comparison must be case insensitive");
+ }
+
+ @Test
+ void destinationTypeParsingFallsBackToQueue() {
+ assertEquals(JmsDestinationType.QUEUE, JmsDestinationType.of(null));
+ assertEquals(JmsDestinationType.QUEUE, JmsDestinationType.of("nonsense"));
+ assertEquals(JmsDestinationType.TOPIC, JmsDestinationType.of(" topic "));
+ assertEquals(JmsDestinationType.QUEUE, JmsDestinationType.of("QUEUE"));
+ }
+}
diff --git
a/plugins/transforms/jms/src/test/java/org/apache/hop/pipeline/transforms/jms/shared/JmsProviderBrokerTest.java
b/plugins/transforms/jms/src/test/java/org/apache/hop/pipeline/transforms/jms/shared/JmsProviderBrokerTest.java
new file mode 100644
index 0000000000..af90ee9bb8
--- /dev/null
+++
b/plugins/transforms/jms/src/test/java/org/apache/hop/pipeline/transforms/jms/shared/JmsProviderBrokerTest.java
@@ -0,0 +1,231 @@
+/*
+ * 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.hop.pipeline.transforms.jms.shared;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
+import org.apache.activemq.artemis.core.server.embedded.EmbeddedActiveMQ;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.core.variables.Variables;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Exercises {@link JmsProvider} against a real embedded Apache ActiveMQ
Artemis broker: connect,
+ * send, receive, acknowledge and redelivery. These are the behaviours that
mocks cannot prove.
+ *
+ * <p>The broker runs in-process on a random-free port, so the test needs no
Docker and no external
+ * service.
+ */
+class JmsProviderBrokerTest {
+
+ private static final int PORT = 61618;
+ private static final String BROKER_URL = "tcp://localhost:" + PORT;
+
+ private static EmbeddedActiveMQ broker;
+ private static Path dataDir;
+
+ @BeforeAll
+ static void startBroker() throws Exception {
+ dataDir = Files.createTempDirectory("hop-jms-it");
+ ConfigurationImpl configuration = new ConfigurationImpl();
+ configuration.setPersistenceEnabled(false);
+ configuration.setSecurityEnabled(false);
+ configuration.setJournalDirectory(dataDir.resolve("journal").toString());
+ configuration.setBindingsDirectory(dataDir.resolve("bindings").toString());
+
configuration.setLargeMessagesDirectory(dataDir.resolve("large").toString());
+ configuration.setPagingDirectory(dataDir.resolve("paging").toString());
+ configuration.addAcceptorConfiguration("tcp", BROKER_URL);
+
+ broker = new EmbeddedActiveMQ();
+ broker.setConfiguration(configuration);
+ broker.start();
+ }
+
+ @AfterAll
+ static void stopBroker() throws Exception {
+ if (broker != null) {
+ broker.stop();
+ }
+ if (dataDir != null) {
+ try (var paths = Files.walk(dataDir)) {
+ paths.sorted(java.util.Comparator.reverseOrder()).forEach(p ->
p.toFile().delete());
+ }
+ }
+ }
+
+ private static JmsConnection connection() {
+ JmsConnection connection = new JmsConnection("embedded");
+ connection.setMode(JmsConnection.MODE_DIRECT);
+ connection.setBrokerUrl(BROKER_URL);
+ return connection;
+ }
+
+ private static JmsProvider provider(String destination, boolean topic,
boolean transacted) {
+ return new JmsProvider(connection(), destination, topic, transacted, null,
null, null);
+ }
+
+ @Test
+ void roundTripsAMessageThroughAQueue() throws Exception {
+ IVariables variables = new Variables();
+ MessageQueueRecord sent = new MessageQueueRecord("ignored", "{\"id\":42}");
+ sent.setKey("corr-42");
+ sent.property("region", "EU");
+
+ try (JmsProvider producer = provider("hop.it.roundtrip", false, false)) {
+ producer.connect(variables);
+ producer.send(sent);
+ }
+
+ try (JmsProvider consumer = provider("hop.it.roundtrip", false, false)) {
+ consumer.connect(variables);
+ MessageQueueRecord received = consumer.receive(5000);
+
+ assertNotNull(received, "the message should have been delivered");
+ assertEquals("{\"id\":42}", received.getBody());
+ assertEquals("corr-42", received.getKey());
+ assertEquals("EU", received.getProperties().get("region"));
+ assertNotNull(received.getMessageId(), "the broker assigns a message
id");
+ assertTrue(received.getDestination().contains("hop.it.roundtrip"));
+ consumer.acknowledge(received);
+ }
+ }
+
+ @Test
+ void receiveReturnsNullOnAnEmptyQueueRatherThanFailing() throws Exception {
+ try (JmsProvider consumer = provider("hop.it.empty", false, false)) {
+ consumer.connect(new Variables());
+ assertNull(consumer.receive(300), "a timeout is normal, not an error");
+ }
+ }
+
+ @Test
+ void unacknowledgedMessageIsRedelivered() throws Exception {
+ IVariables variables = new Variables();
+ try (JmsProvider producer = provider("hop.it.redelivery", false, false)) {
+ producer.connect(variables);
+ producer.send(new MessageQueueRecord(null, "keep-me"));
+ }
+
+ // Receive without acknowledging, then drop the connection: the broker
must give it back.
+ try (JmsProvider consumer = provider("hop.it.redelivery", false, false)) {
+ consumer.connect(variables);
+ assertEquals("keep-me", consumer.receive(5000).getBody());
+ }
+
+ try (JmsProvider consumer = provider("hop.it.redelivery", false, false)) {
+ consumer.connect(variables);
+ MessageQueueRecord redelivered = consumer.receive(5000);
+ assertNotNull(redelivered, "an unacknowledged message must not be lost");
+ assertEquals("keep-me", redelivered.getBody());
+ consumer.acknowledge(redelivered);
+ }
+
+ // Now that it was acknowledged, it is gone.
+ try (JmsProvider consumer = provider("hop.it.redelivery", false, false)) {
+ consumer.connect(variables);
+ assertNull(consumer.receive(500), "an acknowledged message must not be
redelivered");
+ }
+ }
+
+ @Test
+ void publishesAndSubscribesOnATopic() throws Exception {
+ IVariables variables = new Variables();
+ try (JmsProvider subscriber = provider("hop.it.topic", true, false);
+ JmsProvider publisher = provider("hop.it.topic", true, false)) {
+ subscriber.connect(variables);
+ // Force the subscription to exist before publishing; a topic drops
messages with no
+ // subscriber, so this ordering is the point of the test.
+ assertNull(subscriber.receive(200));
+
+ publisher.connect(variables);
+ publisher.send(new MessageQueueRecord(null, "broadcast"));
+
+ MessageQueueRecord received = subscriber.receive(5000);
+ assertNotNull(received, "the subscriber should see the published
message");
+ assertEquals("broadcast", received.getBody());
+ }
+ }
+
+ @Test
+ void transactedSendIsVisibleAfterCommit() throws Exception {
+ IVariables variables = new Variables();
+ try (JmsProvider producer = provider("hop.it.tx", false, true)) {
+ producer.connect(variables);
+ producer.send(new MessageQueueRecord(null, "committed"));
+ }
+ try (JmsProvider consumer = provider("hop.it.tx", false, false)) {
+ consumer.connect(variables);
+ MessageQueueRecord received = consumer.receive(5000);
+ assertNotNull(received);
+ assertEquals("committed", received.getBody());
+ consumer.acknowledge(received);
+ }
+ }
+
+ @Test
+ void variablesAreResolvedInTheDestinationName() throws Exception {
+ IVariables variables = new Variables();
+ variables.setVariable("QUEUE_NAME", "hop.it.from.variable");
+
+ try (JmsProvider producer = provider("${QUEUE_NAME}", false, false)) {
+ producer.connect(variables);
+ producer.send(new MessageQueueRecord(null, "resolved"));
+ }
+ try (JmsProvider consumer = provider("hop.it.from.variable", false,
false)) {
+ consumer.connect(variables);
+ assertEquals("resolved", consumer.receive(5000).getBody());
+ }
+ }
+
+ @Test
+ void aBlankDestinationIsRejectedBeforeConnecting() {
+ try (JmsProvider producer = provider(" ", false, false)) {
+ HopException e = assertThrows(HopException.class, () ->
producer.connect(new Variables()));
+ assertTrue(e.getMessage().contains("destination"), e.getMessage());
+ }
+ }
+
+ @Test
+ void anUnreachableBrokerFailsWithTheDestinationInTheMessage() {
+ JmsConnection unreachable = connection();
+ unreachable.setBrokerUrl("tcp://localhost:1");
+ try (JmsProvider producer =
+ new JmsProvider(unreachable, "hop.it.unreachable", false, false, null,
null, null)) {
+ HopException e = assertThrows(HopException.class, () ->
producer.connect(new Variables()));
+ assertTrue(e.getMessage().contains("hop.it.unreachable"),
e.getMessage());
+ }
+ }
+
+ @Test
+ void closeIsIdempotent() throws Exception {
+ JmsProvider provider = provider("hop.it.close", false, false);
+ provider.connect(new Variables());
+ provider.close();
+ provider.close(); // must not throw
+ }
+}
diff --git
a/plugins/transforms/jms/src/test/java/org/apache/hop/pipeline/transforms/jms/shared/JmsProviderTest.java
b/plugins/transforms/jms/src/test/java/org/apache/hop/pipeline/transforms/jms/shared/JmsProviderTest.java
new file mode 100644
index 0000000000..8900477d7d
--- /dev/null
+++
b/plugins/transforms/jms/src/test/java/org/apache/hop/pipeline/transforms/jms/shared/JmsProviderTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.hop.pipeline.transforms.jms.shared;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import jakarta.jms.BytesMessage;
+import jakarta.jms.Destination;
+import jakarta.jms.Message;
+import jakarta.jms.TextMessage;
+import java.util.Collections;
+import java.util.Enumeration;
+import org.junit.jupiter.api.Test;
+
+/** Covers the JMS-to-Hop mapping, which is the part that can go wrong without
a broker. */
+class JmsProviderTest {
+
+ @Test
+ void mapsTextMessageOntoRecord() throws Exception {
+ Destination destination = mock(Destination.class);
+ when(destination.toString()).thenReturn("orders");
+
+ TextMessage message = mock(TextMessage.class);
+ when(message.getJMSMessageID()).thenReturn("ID:1");
+ when(message.getJMSCorrelationID()).thenReturn("corr-7");
+ when(message.getJMSDestination()).thenReturn(destination);
+ when(message.getJMSTimestamp()).thenReturn(1_700_000_000_000L);
+ when(message.getText()).thenReturn("{\"id\":1}");
+ when(message.getPropertyNames()).thenReturn(enumerationOf("region"));
+ when(message.getObjectProperty("region")).thenReturn("EU");
+
+ MessageQueueRecord record = JmsProvider.toRecord(message);
+
+ assertEquals("ID:1", record.getMessageId());
+ assertEquals("corr-7", record.getKey());
+ assertEquals("orders", record.getDestination());
+ assertEquals("{\"id\":1}", record.getBody());
+ assertEquals(1_700_000_000_000L, record.getTimestamp().getTime());
+ assertEquals("EU", record.getProperties().get("region"));
+ assertSame(message, record.getHandle(), "the handle must allow
acknowledging the message");
+ }
+
+ @Test
+ void leavesBodyNullForNonTextMessagesRatherThanGuessingAnEncoding() throws
Exception {
+ BytesMessage message = mock(BytesMessage.class);
+ when(message.getJMSMessageID()).thenReturn("ID:2");
+ when(message.getPropertyNames()).thenReturn(enumerationOf());
+
+ MessageQueueRecord record = JmsProvider.toRecord(message);
+
+ assertNull(record.getBody());
+ assertEquals("ID:2", record.getMessageId(), "identifiers still come
through");
+ }
+
+ @Test
+ void omitsTimestampWhenTheBrokerDidNotSetOne() throws Exception {
+ Message message = mock(Message.class);
+ when(message.getJMSTimestamp()).thenReturn(0L);
+ when(message.getPropertyNames()).thenReturn(enumerationOf());
+
+ assertNull(JmsProvider.toRecord(message).getTimestamp());
+ }
+
+ @Test
+ void toleratesNullPropertyValues() throws Exception {
+ Message message = mock(Message.class);
+ when(message.getPropertyNames()).thenReturn(enumerationOf("empty"));
+ when(message.getObjectProperty("empty")).thenReturn(null);
+
+ MessageQueueRecord record = JmsProvider.toRecord(message);
+
+ assertTrue(record.getProperties().containsKey("empty"));
+ assertNull(record.getProperties().get("empty"));
+ }
+
+ private static Enumeration<String> enumerationOf(String... names) {
+ return Collections.enumeration(java.util.Arrays.asList(names));
+ }
+}
diff --git a/plugins/transforms/pom.xml b/plugins/transforms/pom.xml
index 70e374551f..c8ad5a2c5f 100644
--- a/plugins/transforms/pom.xml
+++ b/plugins/transforms/pom.xml
@@ -96,6 +96,7 @@
<module>janino</module>
<module>javascript</module>
<module>jdbc-metadata</module>
+ <module>jms</module>
<module>joinrows</module>
<module>json</module>
<module>kafka</module>
