source
ActiveMQ source connector
ActiveMQ Connector integrates Apache Pulsar with Apache ActiveMQ.
Authored by
StreamNative
Support type
streamnative
License
Apache License 2.0

The ActiveMQ source connector receives messages from ActiveMQ clusters and writes messages to Pulsar topics.

Installation

git clone https://github.com/streamnative/pulsar-io-activemq.git
cd pulsar-io-activemq/
mvn clean install -DskipTests
cp target/pulsar-io-activemq-0.0.1.nar $PULSAR_HOME/pulsar-io-activemq-0.0.1.nar

Configuration

The configuration of the ActiveMQ source connector has the following properties.

ActiveMQ source connector configuration

NameTypeRequiredSensitiveDefaultDescription
protocolStringtruefalse"tcp"The ActiveMQ protocol.
hostStringtruefalse" " (empty string)The ActiveMQ host.
portinttruefalse5672The ActiveMQ port.
usernameStringfalsetrue" " (empty string)The username used to authenticate to ActiveMQ.
passwordStringfalsetrue" " (empty string)The password used to authenticate to ActiveMQ.
queueNameStringfalsefalse" " (empty string)The ActiveMQ queue name that messages should be read from or written to.
topicNameStringfalsefalse" " (empty string)The ActiveMQ topic name that messages should be read from or written to.

Configure ActiveMQ source connector

Before using the ActiveMQ source connector, you need to create a configuration file through one of the following methods.

  • JSON

    {
        "tenant": "public",
        "namespace": "default",
        "name": "activemq-source",
        "topicName": "user-op-queue-topic",
        "archive": "connectors/pulsar-io-activemq-2.5.1.nar",
        "parallelism": 1,
        "configs": {
            "protocol": "tcp",
            "host": "localhost",
            "port": "61616",
            "username": "admin",
            "password": "admin",
            "queueName": "user-op-queue"
        }
    }
    
  • YAML

    tenant: "public"
    namespace: "default"
    name: "activemq-source"
    topicName: "user-op-queue-topic"
    archive: "connectors/pulsar-io-activemq-2.5.1.nar"
    parallelism: 1
    
    configs:
        protocol: "tcp"
        host: "localhost"
        port: "61616"
        username: "admin"
        password: "admin"
        queueName: "user-op-queue"
    
  1. Prepare ActiveMQ service.

    docker pull rmohr/activemq
    docker run -p 61616:61616 -p 8161:8161 rmohr/activemq
    
  2. Put the pulsar-io-activemq-2.5.1.nar in the pulsar connectors catalog.

    cp pulsar-io-activemq-2.5.1.nar $PULSAR_HOME/connectors/pulsar-io-activemq-2.5.1.nar
    
  3. Start Pulsar in standalone mode.

    $PULSAR_HOME/bin/pulsar standalone
    
  4. Run ActiveMQ source locally.

    $PULSAR_HOME/bin/pulsar-admin source localrun --source-config-file activemq-source-config.yaml
    
  5. Consume Pulsar messages.

    bin/pulsar-client consume -s "sub-products" public/default/user-op-queue-topic -n 0
    
  6. Send ActiveMQ messages.

    Use the test method sendMessage of the class org.apache.pulsar.ecosystem.io.activemq.ActiveMQDemo to send ActiveMQ messages.

    @Test
    private void sendMessage() throws JMSException {
    
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
    
        @Cleanup
        Connection connection = connectionFactory.createConnection();
        connection.start();
    
        @Cleanup
        Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
    
        Destination destination = session.createQueue("user-op-queue");
    
        @Cleanup
        MessageProducer producer = session.createProducer(destination);
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    
        for (int i = 0; i < 10; i++) {
            String msgContent = "Hello ActiveMQ - " + i;
            ActiveMQTextMessage message = new ActiveMQTextMessage();
            message.setText(msgContent);
            producer.send(message);
        }
    }