​

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 sink connector configuration

​

Configure ActiveMQ sink connector

• JSON
  {
      "tenant": "public",
      "namespace": "default",
      "name": "activemq-sink",
      "inputs": ["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-pulsar"
      }
  }
  
• YAML
  tenant: "public"
  namespace: "default"
  name: "activemq-sink"
  inputs: 
    - "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-pulsar"
  

​

Usage

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 sink locally.
   $PULSAR_HOME/bin/pulsar-admin sink localrun --sink-config-file activemq-sink-config.yaml
   
5. Send Pulsar messages.
   $PULSAR_HOME/bin/pulsar-client produce public/default/user-op-queue-topic --messages hello -n 10
   
6. Consume ActiveMQ messages. Use the test method receiveMessage of the class org.apache.pulsar.ecosystem.io.activemq.ActiveMQDemo to consume ActiveMQ messages.
   @Test
   private void receiveMessage() throws JMSException, InterruptedException {
   
       ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
   
       @Cleanup
       Connection connection = connectionFactory.createConnection();
       connection.start();
   
       @Cleanup
       Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   
       Destination destination = session.createQueue("user-op-queue-pulsar");
   
       @Cleanup
       MessageConsumer consumer = session.createConsumer(destination);
   
       consumer.setMessageListener(new MessageListener() {
           @Override
           public void onMessage(Message message) {
               if (message instanceof ActiveMQTextMessage) {
                   try {
                       System.out.println("get message ----------------- ");
                       System.out.println("receive: " + ((ActiveMQTextMessage) message).getText());
                   } catch (JMSException e) {
                       e.printStackTrace();
                   }
               }
           }
       });
   }