​

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

​

ActiveMQ source connector configuration

​

Configure ActiveMQ source connector

• 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);
       }
   }