...
Code Block | ||
---|---|---|
| ||
/* * 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. */ 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{ 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 = "WSO2"; String topicName_2 = "WSO2.MB"; String topicName_3 = "WSO2.*"; String topicName_4 = "WSO2.#"; public static void main(String[] args) throws NamingException, JMSException { SampleHierarchicalTopicsClient hierarchicalTopicsClient = new SampleHierarchicalTopicsClient(); hierarchicalTopicsClient.subscribe(); } public void subscribe() 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 = topicSession.createTopic(topicName_1); Topic topic2 = topicSession.createTopic(topicName_2); Topic topic3 = (Topic) ctx.lookup(topicName_3); Topic topic4 = (Topic) ctx.lookup(topicName_4); TopicSubscriber topicSubscriber1 = topicSession.createSubscriber(topic3); TopicSubscriber topicSubscriber2 = topicSession.createSubscriber(topic4); // Receive messages Message message1; while ((message1 = topicSubscriber1.receive(5000)) != null){ if (message1 instanceof TextMessage) { TextMessage textMessage = (TextMessage) message1; System.out.println("Got Message from subscriber1 => " + textMessage.getText()); } } Message message2; while ((message2 = topicSubscriber2.receive(5000)) != null){ if (message2 instanceof TextMessage) { TextMessage textMessage = (TextMessage) message2; System.out.println("Got Message from subscriber2 => " + textMessage.getText()); } } topicSubscriber1.close(); topicSubscriber2.close(); topicSession.close(); topicConnection.stop(); 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_3,topicName_3); properties.put("topic."+topicName_4,topicName_4); 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(); } } |
Publish a message to each parent and child topic
Code Block | ||
---|---|---|
| ||
/*
* 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.
*/
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 = "WSO2";
String topicName_2 = "WSO2.MB";
public static void main(String[] args) throws NamingException, JMSException {
TopicPublisher topicPublisher = new TopicPublisher();
topicPublisher.publishMessage();
}
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);
javax.jms.TopicPublisher topicPublisher1 = topicSession.createPublisher(topic1);
javax.jms.TopicPublisher topicPublisher2 = topicSession.createPublisher(topic2);
// Create the messages to send
TextMessage textMessage1 = topicSession.createTextMessage("Message for WSO2");
TextMessage textMessage2 = topicSession.createTextMessage("Message for WSO2.MB");
topicPublisher1.publish(textMessage1);
topicPublisher2.publish(textMessage2);
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);
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();
}
} |
According to the above sample topicSubscriber1 is listening on pattern "WSO2.*" where it receives only the messages published to all the child elements under the parent topic '' WSO2 ". As the topicSubscriber2 listens on pattern "WSO2.#" it receives both the messages published for parent topic ' WSO2 ' as well as to its child topics like MB etc. Hence topicSubscriber1 only receives 1 message while the topicSubscriber2 receives 2 messages.