Page Comparison

This sample demonstrates how to publish messages to topics and sub topics in a topic hierarchy and to create hierarchical topic subscriptions.

Table of Contents

maxLevel	3
minLevel	3

About the sample

The <MB_HOME>/Samples/HierarchicalTopicsSubscriber/src/org/sample/jms directory has the following classes:

...

Localtabgroup

Localtab

title	SampleHierarchicalTopicsClient.java

Code Block

language	java

/*
*  Copyright (c) 2005-2010, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. 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.sample.jms;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.Properties;
public class SampleHierarchicalTopicsClient extends Thread{
    public static final String QPID_ICF = "org.wso2.andes.jndi.PropertiesFileInitialContextFactory";
    private static final String CF_NAME_PREFIX = "connectionfactory.";
    private static final String CF_NAME = "qpidConnectionfactory";
    String userName = "admin";
    String password = "admin";
    private static String CARBON_CLIENT_ID = "carbon";
    private static String CARBON_VIRTUAL_HOST_NAME = "carbon";
    private static String CARBON_DEFAULT_HOSTNAME = "localhost";
    private static String CARBON_DEFAULT_PORT = "5672";
    String topicName_1 = "Games";
    String topicName_2 = "Games.Cricket";
    String topicName_3 = "Games.Cricket.SL";
    String topicName_4 = "Games.Cricket.India";
    String topicName_5 = "Games.Cricket.India.Delhi";
    String topicName_6 = "Games.Cricket.*";
    String topicName_7 = "Games.Cricket.#";
    @Override
    public void run() {
        try {
package org.sample.jms;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.Properties;
public class SampleHierarchicalTopicsClient extends Thread{
    public static final String QPID_ICF = "org.wso2.andes.jndi.PropertiesFileInitialContextFactory";
    private static final String CF_NAME_PREFIX = "connectionfactory.";
    private static final String CF_NAME = "qpidConnectionfactory";
    String userName = "admin";
    String password = "admin";
    private static String CARBON_CLIENT_ID = "carbon";
    private static String CARBON_VIRTUAL_HOST_NAME = "carbon";
    private static String CARBON_DEFAULT_HOSTNAME = "localhost";
    private static String CARBON_DEFAULT_PORT = "5672";
    String topicName_1 = "Games";
    String topicName_2 = "Games.Cricket";
    String topicName_3 = "Games.Cricket.SL";
    String topicName_4 = "Games.Cricket.India";
    String topicName_5 = "Games.Cricket.India.Delhi";
    String topicName_6 = "Games.Cricket.*";
    String topicName_7 = "Games.Cricket.#";
    @Override
    public void run() {
        try {
            subscribe();
        } catch (NamingException e) {
            e.printStackTrace();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
    public void subscribe() throws NamingException, JMSException {
        InitialContext ctx = init();
        // Lookup connection factory
        TopicConnectionFactory connFactory = subscribe(TopicConnectionFactory);
    ctx.lookup(CF_NAME);
    } catch (NamingException e) {TopicConnection topicConnection = connFactory.createTopicConnection();
         etopicConnection.printStackTracestart();
        TopicSession topicSession }=
catch (JMSException e) {             etopicConnection.printStackTrace(createTopicSession(false, QueueSession.AUTO_ACKNOWLEDGE);
        }Topic topic1 = topicSession.createTopic(topicName_1);
 }     public void subscribe()Topic throwstopic2 NamingException, JMSException {= topicSession.createTopic(topicName_2);
        Topic InitialContext ctxtopic3 = init(topicSession.createTopic(topicName_3);
        //Topic Lookuptopic4 connection factory= topicSession.createTopic(topicName_4);
        TopicConnectionFactoryTopic connFactorytopic5 = (TopicConnectionFactory) ctx.lookup(CF_NAMEtopicSession.createTopic(topicName_5);
        TopicConnectionTopic topicConnectiontopic6 = connFactory.createTopicConnection(Topic);
        topicConnection.start( ctx.lookup(topicName_6);
        TopicSessionTopic topicSessiontopic7 = (Topic) ctx.lookup(topicName_7);
        TopicSubscriber topicSubscriber1 =    topicConnection.createTopicSession(false, QueueSession.AUTO_ACKNOWLEDGEtopicSession.createSubscriber(topic6);
        TopicTopicSubscriber topic1topicSubscriber2 = topicSession.createTopiccreateSubscriber(topicName_1topic7);
        Topic// topic2Receive = topicSession.createTopic(topicName_2);messages
        Message Topic topic3 = topicSession.createTopic(topicName_3);
        Topic topic4 = topicSession.createTopic(topicName_4message1;
	System.out.println(" Receiving messages for " + topicName_6 + " :");
        Topicwhile topic5((message1 = topicSessiontopicSubscriber1.createTopic(topicName_5);
 receive(5000)) != null){
      Topic topic6 = (Topic) ctx.lookup(topicName_6);  if (message1 instanceof TextMessage) {
  Topic  topic7 = (Topic) ctx.lookup(topicName_7);         TopicSubscriberTextMessage topicSubscriber1textMessage = topicSession.createSubscriber(topic6TextMessage) message1;
        TopicSubscriber topicSubscriber2 = topicSession.createSubscriber(topic7);       System.out.println("Got Message from subscriber1 => " + textMessage.getText());
         // Receive messages   }
        }
        Message message1message2;
	System.out.println(" Receiving messages for " + topicName_67 + " :");
        while ((message1message2 = topicSubscriber1topicSubscriber2.receive(5000)) != null){
            if (message1message2 instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message1message2;
                System.out.println("Got Message from subscriber1subscriber2 => " + textMessage.getText());
            }
        }
        Message message2;
	System.out.println(" Receiving messages for " + topicName_7 + " :"topicSubscriber1.close();
        topicSubscriber2.close();
        while ((message2 = topicSubscriber2.receive(5000)) != null){topicSession.close();
        topicConnection.stop();
     if (message2 instanceof TextMessagetopicConnection.close();
{    }
    private        TextMessage textMessage = (TextMessage) message2;
 InitialContext init() throws NamingException {
        Properties properties = new Properties();
 System.out.println("Got Message from subscriber2 => " + textMessageproperties.getText()put(Context.INITIAL_CONTEXT_FACTORY, QPID_ICF);
            }
  properties.put(CF_NAME_PREFIX + CF_NAME, getTCPConnectionURL(userName, password));
     }         topicSubscriber1.close(properties.put("topic."+topicName_6,topicName_6);
        topicSubscriber2properties.close(put("topic."+topicName_7,topicName_7);
        topicSession.close();
        topicConnection.stop();
        topicConnection.close(return new InitialContext(properties);
    }
    private InitialContextString init() throws NamingExceptiongetTCPConnectionURL(String username, String password) {
        Properties properties = new Properties();// amqp://{username}:{password}@carbon/carbon?brokerlist='tcp://{hostname}:{port}'
        return new properties.put(Context.INITIAL_CONTEXT_FACTORY, QPID_ICF);StringBuffer()
        properties.put(CF_NAME_PREFIX + CF_NAME, getTCPConnectionURL(userName, password));
        properties.putappend("topic."+topicName_6,topicName_6);amqp://").append(username).append(":").append(password)
                properties.putappend("topic."+topicName_7,topicName_7);
 @").append(CARBON_CLIENT_ID)
      return new InitialContext(properties);     }     private String getTCPConnectionURL(String username, String password) {.append("/").append(CARBON_VIRTUAL_HOST_NAME)
               // amqp://{username}:{password}@carbon/carbon .append("?brokerlist='tcp://{hostname}:{port}'
        return new StringBuffer()
                .append("amqp://").append(username).append(":").append(password").append(CARBON_DEFAULT_HOSTNAME).append(":").append(CARBON_DEFAULT_PORT).append("'")
                .appendtoString("@").append(CARBON_CLIENT_ID));
    }
}

Localtab

title	TopicPublisher.java

Code Block

language	java

package org.sample.jms;
import javax.jms.JMSException;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.jms.Topic;
import  .append("/").append(CARBON_VIRTUAL_HOST_NAME)
                .append("?brokerlist='tcp://").append(CARBON_DEFAULT_HOSTNAME).append(":").append(CARBON_DEFAULT_PORT).append("'")
                .toString();
    }
}

Localtab

title	TopicPublisher.java

Code Block

language	java

/*
*  Copyright (c) 2005-2010, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. 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.sample.jms;
import javax.jms.JMSException;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.Properties;
public class TopicPublisher {
    public static final String QPID_ICF = "org.wso2.andes.jndi.PropertiesFileInitialContextFactory";
    private static final String CF_NAME_PREFIX = "connectionfactory.";
    private static final String CF_NAME = "qpidConnectionfactory";
    String userName = "admin";
    String password = "admin";
    private static String CARBON_CLIENT_ID = "carbon";
    private static String CARBON_VIRTUAL_HOST_NAME = "carbon";
    private static String CARBON_DEFAULT_HOSTNAME = "localhost";
    private static String CARBON_DEFAULT_PORT = "5672";
    String topicName_1 = "Games";
    String topicName_2 = "Games.Cricket";
    String topicName_3 = "Games.Cricket.SL";
    String topicName_4 = "Games.Cricket.India";
    String topicName_5 = "Games.Cricket.India.Delhi";
    public void publishMessage() throws NamingException, JMSException {javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.Properties;
public class TopicPublisher {
    public static final String QPID_ICF = "org.wso2.andes.jndi.PropertiesFileInitialContextFactory";
    private static final String CF_NAME_PREFIX = "connectionfactory.";
    private static final String CF_NAME = "qpidConnectionfactory";
    String userName = "admin";
    String password = "admin";
    private static String CARBON_CLIENT_ID = "carbon";
    private static String CARBON_VIRTUAL_HOST_NAME = "carbon";
    private static String CARBON_DEFAULT_HOSTNAME = "localhost";
    private static String CARBON_DEFAULT_PORT = "5672";
    String topicName_1 = "Games";
    String topicName_2 = "Games.Cricket";
    String topicName_3 = "Games.Cricket.SL";
    String topicName_4 = "Games.Cricket.India";
    String topicName_5 = "Games.Cricket.India.Delhi";
    public void publishMessage() throws NamingException, JMSException {
        InitialContext ctx = init();
        // Lookup connection factory
        TopicConnectionFactory connFactory = (TopicConnectionFactory) ctx.lookup(CF_NAME);
        TopicConnection topicConnection = connFactory.createTopicConnection();
        topicConnection.start();
        TopicSession topicSession =
                topicConnection.createTopicSession(false, QueueSession.AUTO_ACKNOWLEDGE);
        Topic topic1 = (Topic) ctx.lookup(topicName_1);
        Topic topic2 = (Topic) ctx.lookup(topicName_2);
        Topic topic3 = (Topic) ctx.lookup(topicName_3);
        Topic topic4 = (Topic) ctx.lookup(topicName_4);
        Topic topic5 = (Topic) ctx.lookup(topicName_5);

        javax.jms.TopicPublisher topicPublisher1 = topicSession.createPublisher(topic1);
        javax.jms.TopicPublisher topicPublisher2 = topicSession.createPublisher(topic2);
        javax.jms.TopicPublisher topicPublisher3 = topicSession.createPublisher(topic3);
        javax.jms.TopicPublisher topicPublisher4 = topicSession.createPublisher(topic4);
        InitialContext ctxjavax.jms.TopicPublisher topicPublisher5 = inittopicSession.createPublisher(topic5);
        // LookupCreate connectionthe factorymessages to send
      TopicConnectionFactory  TextMessage connFactorytextMessage1 = (TopicConnectionFactory) ctx.lookup(CF_NAMEtopicSession.createTextMessage("Message for Cricket");
        TopicConnectionTextMessage topicConnectiontextMessage2 = connFactorytopicSession.createTopicConnection(createTextMessage("Message for SL");
        TextMessage topicConnection.start(textMessage3 = topicSession.createTextMessage("Message for India");
        TopicSessionTextMessage topicSessiontextMessage4 =      topicSession.createTextMessage("Message for Delhi");
          topicConnectiontopicPublisher2.createTopicSession(false, QueueSession.AUTO_ACKNOWLEDGEpublish(textMessage1);
        Topic topic1 = (Topic) ctx.lookup(topicName_1topicPublisher3.publish(textMessage2);
        Topic topic2 = (Topic) ctx.lookup(topicName_2topicPublisher4.publish(textMessage3);
        Topic topic3 = (Topic) ctx.lookup(topicName_3topicPublisher5.publish(textMessage4);
        topicSession.close();
        Topic topic4 = (Topic) ctx.lookup(topicName_4topicConnection.close();
    }
   Topic topic5private =InitialContext init(Topic) ctx.lookup(topicName_5);
 throws NamingException {
        javax.jms.TopicPublisher topicPublisher1Properties properties = topicSession.createPublisher(topic1new Properties();
        javaxproperties.jms.TopicPublisher topicPublisher2 = topicSession.createPublisher(topic2put(Context.INITIAL_CONTEXT_FACTORY, QPID_ICF);
        javax.jms.TopicPublisher topicPublisher3 = topicSession.createPublisher(topic3properties.put(CF_NAME_PREFIX + CF_NAME, getTCPConnectionURL(userName, password));
        javax.jms.TopicPublisher topicPublisher4 = topicSession.createPublisher(topic4properties.put("topic."+topicName_1,topicName_1);
        properties.put("topic."+topicName_2,topicName_2);
        javax.jms.TopicPublisher topicPublisher5 = topicSession.createPublisher(topic5properties.put("topic."+topicName_3,topicName_3);
        // Create the messages to send
   properties.put("topic."+topicName_4,topicName_4);
        properties.put("topic."+topicName_5,topicName_5);

     TextMessage textMessage1 = topicSession.createTextMessage("Message for Cricket" return new InitialContext(properties);
    }
   TextMessage textMessage2private =String topicSession.createTextMessage("Message for SL");
getTCPConnectionURL(String username, String password) {
       TextMessage textMessage3 = topicSession.createTextMessage("Message for India");
   // amqp://{username}:{password}@carbon/carbon?brokerlist='tcp://{hostname}:{port}'
     TextMessage textMessage4 = topicSession.createTextMessage("Message for Delhi");
        topicPublisher2.publish(textMessage1);
        topicPublisher3.publish(textMessage2);
        topicPublisher4.publish(textMessage3);
        topicPublisher5.publish(textMessage4);
        topicSession.close();
        topicConnection.close();
    }
    private InitialContext init() throws NamingException {
        Properties properties = new Properties();
        properties.put(Context.INITIAL_CONTEXT_FACTORY, QPID_ICF);
        properties.put(CF_NAME_PREFIX + CF_NAME, getTCPConnectionURL(userName, password));
        properties.put("topic."+topicName_1,topicName_1);
        properties.put("topic."+topicName_2,topicName_2);
        properties.put("topic."+topicName_3,topicName_3);
        properties.put("topic."+topicName_4,topicName_4);
        properties.put("topic."+topicName_5,topicName_5);

        return new InitialContext(properties);
    }
    private String getTCPConnectionURL(String username, String password) {
        // amqp://{username}:{password}@carbon/carbon?brokerlist='tcp://{hostname}:{port}'
        return new StringBuffer()
                .append("amqp://").append(username).append(":").append(password)
                .append("@").append(CARBON_CLIENT_ID)
                .append("/").append(CARBON_VIRTUAL_HOST_NAME)
                .append("?brokerlist='tcp://").append(CARBON_DEFAULT_HOSTNAME).append(":").append(CARBON_DEFAULT_PORT).append("'")
                .toString();
    }
}

Localtab

title	Main.java

Code Block

language	java

/*
*  Copyright (c) 2005-2010, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. 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.
*/
 return new StringBuffer()
                .append("amqp://").append(username).append(":").append(password)
                .append("@").append(CARBON_CLIENT_ID)
                .append("/").append(CARBON_VIRTUAL_HOST_NAME)
                .append("?brokerlist='tcp://").append(CARBON_DEFAULT_HOSTNAME).append(":").append(CARBON_DEFAULT_PORT).append("'")
                .toString();
    }
}

Localtab

title	Main.java

Code Block

language	java

package org.sample.jms;

import javax.jms.JMSException;
import javax.naming.NamingException;
public class Main {
    public static void main(String[] args) throws NamingException, JMSException {
        SampleHierarchicalTopicsClient hierarchicalTopicsClient = new SampleHierarchicalTopicsClient();
        hierarchicalTopicsClient.start();
        TopicPublisher topicPublisher = new TopicPublisher();
        topicPublisher.publishMessage();
    }
}

...

Versions Compared

Old Version 5

New Version 6

Key

About the sample