Este conteúdo não está disponível no idioma selecionado.
2.3. Create a Consumer Client
Overview
This section describes how to create a consumer client of the notification broker. The consumer client subscribes to a particular topic and creates a callback service, which is capable of receiving messages directly from the broker.
Sample consumer client code
Example 2.2, “Consumer Client Code” shows the code for a sample consumer client that subscribes to messages published on the
MyTopic topic.
Example 2.2. Consumer Client Code
// Java
package org.jboss.fuse.example.wsn.consumer;
import javax.xml.bind.JAXBElement;
import javax.xml.namespace.QName;
import org.w3c.dom.Element;
import org.apache.cxf.wsn.client.Consumer;
import org.apache.cxf.wsn.client.NotificationBroker;
import org.apache.cxf.wsn.client.Subscription;
import org.oasis_open.docs.wsn.b_2.NotificationMessageHolderType;
/**
*
*/
public final class Client {
private Client() {
//not constructed
}
/**
* @param args
*/
public static void main(String[] args) throws Exception {
String wsnPort = "8182";
if (args.length > 0) {
wsnPort = args[0];
}
// Start a consumer that will listen for notification messages
// We'll just print the text content out for now.
Consumer consumer = new Consumer(new Consumer.Callback() {
public void notify(NotificationMessageHolderType message) {
Object o = message.getMessage().getAny();
System.out.println(message.getMessage().getAny());
if (o instanceof Element) {
System.out.println(((Element)o).getTextContent());
}
}
}, "http://localhost:9001/MyConsumer");
// Create a subscription for a Topic on the broker
NotificationBroker notificationBroker
= new NotificationBroker("http://localhost:" + wsnPort + "/wsn/NotificationBroker");
Subscription subscription = notificationBroker.subscribe(consumer, "MyTopic");
// Just sleep for a bit to pick up some incoming messages
Thread.sleep(60000);
// Cleanup and exit
subscription.unsubscribe();
consumer.stop();
System.exit(0);
}
}Creating a consumer callback object
In order to receive notification messages from the notification broker, you must create a consumer callback object to receive the messages. The consumer callback object is in fact a Web service which is embedded in your client. The easiest way to create the consumer callback is to use the
org.apache.cxf.wsn.client.Consumer class from the simplified client API, which enables you to define a callback as follows:
// Start a consumer that will listen for notification messages
// We'll just print the text content out for now.
Consumer consumer = new Consumer(new Consumer.Callback() {
public void notify(NotificationMessageHolderType message) {
Object o = message.getMessage().getAny();
System.out.println(message.getMessage().getAny());
if (o instanceof Element) {
System.out.println(((Element)o).getTextContent());
}
}
}, "http://localhost:9001/MyConsumer");
The first argument to the
Consumer constructor is a reference to the consumer callback object, which is defined inline. The second argument specifies the URL of the consumer callback endpoint, which can receive messages from the notification broker.
Subscribing to a topic
To start receiving messages, you must subscribe the consumer to a topic in the notification broker. To create a subscription, invoke the following
subscribe method on the NotificationBroker proxy object:
Subscription subscribe(Referencable consumer, String topic)
The first argument is a reference to a
Consumer object (which is capable of returning a WS-Addressing endpoint reference to the consumer callback through the Referencable.getEpr() method). The second argument is the name of the topic you want to subscribe to.
The return value is a reference to a
Subscription object, which you can use to manage the subscription (for example, pause, resume, or unsubscribe).
Threading in the consumer client
Because the consumer client has an embedded Web service (the consumer callback object), which automatically starts in a background thread, it is necessary to manage threading in this sample client. In particular, after creating the subscription, you need to put the main thread to sleep (by calling
Thread.sleep(60000)), so that the thread context can switch to the background thread, where the callback Web service is running. This makes it possible for the consumer callback to receive some messages.
Steps to create a consumer client
Perform the following steps to create a consumer client:
- Use the
archetype:generategoal to invoke theservicemix-cxf-code-first-osgi-bundlearchetype. Under thewsndirectory, invoke the Maven archetype as follows:mvn archetype:generate \ -DarchetypeGroupId=io.fabric8.archetypes \ -DarchetypeArtifactId=karaf-soap-archetype \ -DarchetypeVersion=1.2.0.redhat-621084 \ -DgroupId=org.jboss.fuse.example \ -DartifactId=wsn-consumer \ -Dversion=1.0-SNAPSHOT \ -Dpackage=org.jboss.fuse.example.wsn.consumer \ -Dfabric8-profile=wsn-consumer-profile
NoteThe backslash characters at the end of each line are effective as line-continuation characters on UNIX and LINUX platforms. If you are entering the command on a Windows platform, however, you must enter the entire command on a single line.You will be prompted to confirm the project settings, with a message similar to this one:Confirm properties configuration: groupId: org.jboss.fuse.example artifactId: wsn-consumer version: 1.0-SNAPSHOT package: org.jboss.fuse.example.wsn.consumer fabric8-profile: wsn-consumer-profile Y: :
Type Return to accept the settings and generate the project. When the command finishes, you should find a new Maven project in thewsn/wsn-consumerdirectory. - Some of the generated project files are not needed for this tutorial. Under the
wsn/wsn-consumerdirectory, delete the following files and directories:src/main/resources/OSGI-INF/blueprint/blueprint.xml src/main/java/org/jboss/fuse/example/wsn/consumer/HelloWorld.java src/main/java/org/jboss/fuse/example/wsn/consumer/HelloWorldImpl.java
- Edit the
pom.xmlfile in thewsn-consumerdirectory, and add the following dependencies, as required by the consumer client:<?xml version="1.0" encoding="UTF-8"?> <project ...> ... <dependencies> ... <!-- Needed for HTTP callback endpoint --> <dependency> <groupId>org.apache.cxf</groupId> <artifactId>cxf-rt-transports-http-jetty</artifactId> </dependency> <!-- Needed for WS-Notification --> <dependency> <groupId>org.apache.cxf.services.wsn</groupId> <artifactId>cxf-services-wsn-api</artifactId> </dependency> </dependencies> ... </project> - Delete the
cxf-java2ws-pluginplug-in configuration from thewsn-consumer/pom.xmlfile. That is, open thepom.xmlfile and delete thecxf-java2ws-pluginplug-in configuration as highlighted in the following example:<?xml version="1.0" encoding="UTF-8"?> <project ...> ... <build> <plugins> ... <!-- DELETE THE FOLLOWING LINES! --> <plugin> <groupId>org.apache.cxf</groupId> <artifactId>cxf-java2ws-plugin</artifactId> <version>${cxf-version}</version> <executions> <execution> <id>process-classes</id> <phase>process-classes</phase> <goals> <goal>java2ws</goal> </goals> <configuration> <className>org.jboss.fuse.example.wsn.consumer.HelloWorld</className> <genWsdl>true</genWsdl> <attachWsdl>false</attachWsdl> <verbose>true</verbose> </configuration> </execution> </executions> </plugin> ... </plugins> </build> ... </project> - Add a
clientprofile to the POM file, which provides an easy way to run the publisher client code. Edit thewsn-consumer/pom.xmlfile and add the newprofileelement, as highlighted in the following example:<?xml version="1.0" encoding="UTF-8"?> <project ...> ... <profiles> ... <profile> <id>client</id> <build> <defaultGoal>test</defaultGoal> <plugins> <plugin> <groupId>org.codehaus.mojo</groupId> <artifactId>exec-maven-plugin</artifactId> <executions> <execution> <phase>test</phase> <goals> <goal>java</goal> </goals> <configuration> <mainClass>org.jboss.fuse.example.wsn.consumer.Client</mainClass> </configuration> </execution> </executions> </plugin> </plugins> </build> </profile> </profiles> ... </project> - Create a new
Client.javafile in thewsn-consumer/src/main/java/org/jboss/fuse/example/wsn/consumer/directory, and add the code from Example 2.2, “Consumer Client Code” to this file.
Test the consumer client
Test the consumer client as follows:
- If the JBoss A-MQ container is not already running (with the notification broker installed), start it up now:
./amq
- Run the publisher client at the command line. Open a new command prompt, and enter the following commands:
cd wsn/wsn-publisher mvn -Pclient
In the command window, you should see some output like the following:... INFO: Creating Service {http://cxf.apache.org/wsn/jaxws}NotificationBrokerService from WSDL: jar:file:/Users/fbolton/.m2/repository/org/apache/cxf/services/wsn/ cxf-services-wsn-api/2.6.0.redhat-60024/cxf-services-wsn-api-2.6.0.redhat-60024.jar !/org/apache/cxf/wsn/wsdl/wsn.wsdlYou now have approximately two minutes before the publisher client times out. - Run the consumer client at the command line. Open a new command prompt and enter the following commands:
cd wsn/wsn-consumer mvn -Pclient
In the command window, you should see some output like the following:... [INFO] --- exec-maven-plugin:1.4.0:java (default) @ wsn-consumer --- [foo: null] Hello World! ...
- To inspect the state of the notification broker, you can connect to the JMX port of the ActiveMQ broker. Start up a JMX console by entering the following command at the command line:
jconsole
In the JConsole: New Connection dialog, select Remote Process and enter the following URL in the accompanying text field:service:jmx:rmi:///jndi/rmi://localhost:1099/karaf-root
In the Username and Password fields, enter one of the user credentials you created at the start of this tutorial. When you are connected to the JMX port, you can inspect the state of the broker by clicking on the MBeans tab and drilling down the object tree in the JConsole.
