You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicemix.apache.org by gn...@apache.org on 2005/12/21 10:13:33 UTC
svn commit: r358237 - in /incubator/servicemix/trunk/servicemix-wsn2005/src:
main/java/org/servicemix/wsn/ main/java/org/servicemix/wsn/client/
main/java/org/servicemix/wsn/jbi/ main/java/org/servicemix/wsn/jms/
test/java/org/servicemix/wsn/ test/java/...
Author: gnodet
Date: Wed Dec 21 01:13:24 2005
New Revision: 358237
URL: http://svn.apache.org/viewcvs?rev=358237&view=rev
Log:
Add raw notification mode for wsn2005
Modified:
incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/AbstractSubscription.java
incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/client/NotificationBroker.java
incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jbi/JbiSubscription.java
incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jms/JmsSubscription.java
incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/SubscriptionTest.java
incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/component/WSNComponentTest.java
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/AbstractSubscription.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/AbstractSubscription.java?rev=358237&r1=358236&r2=358237&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/AbstractSubscription.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/AbstractSubscription.java Wed Dec 21 01:13:24 2005
@@ -346,9 +346,5 @@
"InitialTerminationTime is not supported",
fault);
}
- if (useRaw) {
- InvalidUseRawValueFaultType fault = new InvalidUseRawValueFaultType();
- throw new InvalidUseRawValueFault("UseRaw mode is not supported", fault);
- }
}
}
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/client/NotificationBroker.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/client/NotificationBroker.java?rev=358237&r1=358236&r2=358237&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/client/NotificationBroker.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/client/NotificationBroker.java Wed Dec 21 01:13:24 2005
@@ -19,7 +19,9 @@
import org.oasis_open.docs.wsn.b_1.QueryExpressionType;
import org.oasis_open.docs.wsn.b_1.Subscribe;
import org.oasis_open.docs.wsn.b_1.SubscribeResponse;
+import org.oasis_open.docs.wsn.b_1.SubscriptionPolicy;
import org.oasis_open.docs.wsn.b_1.TopicExpressionType;
+import org.oasis_open.docs.wsn.b_1.UseRaw;
import org.oasis_open.docs.wsn.br_1.RegisterPublisher;
import org.oasis_open.docs.wsn.br_1.RegisterPublisherResponse;
import org.servicemix.client.DefaultServiceMixClient;
@@ -93,8 +95,20 @@
}
public Subscription subscribe(EndpointReferenceType consumer,
+ String topic) throws JBIException {
+ return subscribe(consumer, topic, null, false);
+ }
+
+ public Subscription subscribe(EndpointReferenceType consumer,
String topic,
String xpath) throws JBIException {
+ return subscribe(consumer, topic, xpath, false);
+ }
+
+ public Subscription subscribe(EndpointReferenceType consumer,
+ String topic,
+ String xpath,
+ boolean raw) throws JBIException {
Subscribe subscribeRequest = new Subscribe();
subscribeRequest.setConsumerReference(consumer);
@@ -109,6 +123,10 @@
xpathExp.setDialect(AbstractSubscription.XPATH1_URI);
xpathExp.getContent().add(xpath);
subscribeRequest.getFilter().getAny().add(new JAXBElement<QueryExpressionType>(AbstractSubscription.QNAME_MESSAGE_CONTENT, QueryExpressionType.class, xpathExp));
+ }
+ if (raw) {
+ subscribeRequest.setSubscriptionPolicy(new Subscribe.SubscriptionPolicy());
+ subscribeRequest.getSubscriptionPolicy().getAny().add(new UseRaw());
}
SubscribeResponse response = (SubscribeResponse) request(subscribeRequest);
return new Subscription(response.getSubscriptionReference(), getClient());
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jbi/JbiSubscription.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jbi/JbiSubscription.java?rev=358237&r1=358236&r2=358237&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jbi/JbiSubscription.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jbi/JbiSubscription.java Wed Dec 21 01:13:24 2005
@@ -8,12 +8,12 @@
import javax.jbi.messaging.NormalizedMessage;
import javax.jbi.servicedesc.ServiceEndpoint;
import javax.xml.namespace.QName;
+import javax.xml.transform.dom.DOMSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.oasis_open.docs.wsn.b_1.Subscribe;
import org.oasis_open.docs.wsn.b_1.SubscribeCreationFailedFaultType;
-import org.servicemix.jbi.jaxp.StringSource;
import org.servicemix.wsn.jaxws.InvalidFilterFault;
import org.servicemix.wsn.jaxws.InvalidMessageContentExpressionFault;
import org.servicemix.wsn.jaxws.InvalidProducerPropertiesExpressionFault;
@@ -24,6 +24,7 @@
import org.servicemix.wsn.jaxws.TopicNotSupportedFault;
import org.servicemix.wsn.jaxws.UnacceptableInitialTerminationTimeFault;
import org.servicemix.wsn.jms.JmsSubscription;
+import org.w3c.dom.Element;
public class JbiSubscription extends JmsSubscription {
@@ -63,14 +64,14 @@
}
@Override
- protected void doNotify(String notify) {
+ protected void doNotify(Element content) {
try {
DeliveryChannel channel = context.getDeliveryChannel();
MessageExchangeFactory factory = channel.createExchangeFactory(endpoint);
InOnly inonly = factory.createInOnlyExchange();
NormalizedMessage msg = inonly.createMessage();
inonly.setInMessage(msg);
- msg.setContent(new StringSource(notify));
+ msg.setContent(new DOMSource(content));
if (!channel.sendSync(inonly)) {
log.warn("Notification was aborted");
}
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jms/JmsSubscription.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jms/JmsSubscription.java?rev=358237&r1=358236&r2=358237&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jms/JmsSubscription.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/servicemix/wsn/jms/JmsSubscription.java Wed Dec 21 01:13:24 2005
@@ -1,6 +1,5 @@
package org.servicemix.wsn.jms;
-import java.io.IOException;
import java.io.StringReader;
import javax.jms.Connection;
@@ -13,7 +12,6 @@
import javax.jms.Topic;
import javax.xml.datatype.XMLGregorianCalendar;
import javax.xml.parsers.DocumentBuilderFactory;
-import javax.xml.parsers.ParserConfigurationException;
import javax.xml.xpath.XPath;
import javax.xml.xpath.XPathConstants;
import javax.xml.xpath.XPathExpression;
@@ -46,7 +44,6 @@
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.xml.sax.InputSource;
-import org.xml.sax.SAXException;
public abstract class JmsSubscription extends AbstractSubscription implements MessageListener {
@@ -149,48 +146,46 @@
this.connection = connection;
}
- public void onMessage(Message message) {
+ public void onMessage(Message jmsMessage) {
try {
- TextMessage text = (TextMessage) message;
- boolean match = doFilter(text.getText());
+ TextMessage text = (TextMessage) jmsMessage;
+ DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
+ factory.setNamespaceAware(true);
+ Document doc = factory.newDocumentBuilder().parse(new InputSource(new StringReader(text.getText())));
+ Element root = doc.getDocumentElement();
+ Element holder = (Element) root.getElementsByTagNameNS("http://docs.oasis-open.org/wsn/b-1", "NotificationMessage").item(0);
+ Element message = (Element) holder.getElementsByTagNameNS("http://docs.oasis-open.org/wsn/b-1", "Message").item(0);
+ Element content = null;
+ for (int i = 0; i < message.getChildNodes().getLength(); i++) {
+ if (message.getChildNodes().item(i) instanceof Element) {
+ content = (Element) message.getChildNodes().item(i);
+ break;
+ }
+ }
+ boolean match = doFilter(content);
if (match) {
- doNotify(text.getText());
+ if (useRaw) {
+ doNotify(content);
+ } else {
+ doNotify(root);
+ }
}
} catch (Exception e) {
log.warn("Error notifying consumer", e);
}
}
- protected boolean doFilter(String notify) {
+ protected boolean doFilter(Element content) {
if (contentFilter != null) {
if (!contentFilter.getDialect().equals(XPATH1_URI)) {
throw new IllegalStateException("Unsupported dialect: " + contentFilter.getDialect());
}
try {
- DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
- factory.setNamespaceAware(true);
- Document doc = factory.newDocumentBuilder().parse(new InputSource(new StringReader(notify)));
- Element root = doc.getDocumentElement();
- Element holder = (Element) root.getElementsByTagNameNS("http://docs.oasis-open.org/wsn/b-1", "NotificationMessage").item(0);
- Element message = (Element) holder.getElementsByTagNameNS("http://docs.oasis-open.org/wsn/b-1", "Message").item(0);
- Element content = null;
- for (int i = 0; i < message.getChildNodes().getLength(); i++) {
- if (message.getChildNodes().item(i) instanceof Element) {
- content = (Element) message.getChildNodes().item(i);
- break;
- }
- }
XPathFactory xpfactory = XPathFactory.newInstance();
XPath xpath = xpfactory.newXPath();
XPathExpression exp = xpath.compile(contentFilter.getContent().get(0).toString());
Boolean ret = (Boolean) exp.evaluate(content, XPathConstants.BOOLEAN);
return ret.booleanValue();
- } catch (SAXException e) {
- log.warn("Could not filter notification", e);
- } catch (IOException e) {
- log.warn("Could not filter notification", e);
- } catch (ParserConfigurationException e) {
- log.warn("Could not filter notification", e);
} catch (XPathExpressionException e) {
log.warn("Could not filter notification", e);
}
@@ -199,6 +194,6 @@
return true;
}
- protected abstract void doNotify(String notify);
+ protected abstract void doNotify(Element content);
}
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/SubscriptionTest.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/SubscriptionTest.java?rev=358237&r1=358236&r2=358237&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/SubscriptionTest.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/SubscriptionTest.java Wed Dec 21 01:13:24 2005
@@ -71,12 +71,7 @@
public void testWithUseRaw() throws Exception {
Subscribe subscribe = getSubscription("subscribe-raw.xml");
- try {
- subscription.validateSubscription(subscribe);
- fail("UseRaw used. Fault was expected.");
- } catch (InvalidUseRawValueFault e) {
- // OK
- }
+ subscription.validateSubscription(subscribe);
}
public void testWithProducerProperties() throws Exception {
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/component/WSNComponentTest.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/component/WSNComponentTest.java?rev=358237&r1=358236&r2=358237&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/component/WSNComponentTest.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/test/java/org/servicemix/wsn/component/WSNComponentTest.java Wed Dec 21 01:13:24 2005
@@ -40,6 +40,7 @@
import org.w3._2005._03.addressing.EndpointReferenceType;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
+import org.w3c.dom.Node;
import org.xml.sax.InputSource;
public class WSNComponentTest extends TestCase {
@@ -101,6 +102,29 @@
Thread.sleep(50);
receiver.getMessageList().assertMessagesReceived(1);
+ NormalizedMessage msg = (NormalizedMessage) receiver.getMessageList().getMessages().get(0);
+ Node node = new SourceTransformer().toDOMNode(msg);
+ assertEquals("Notify", node.getLocalName());
+
+ // Wait for acks to be processed
+ Thread.sleep(50);
+ }
+
+ public void testRawNotify() throws Exception {
+ ReceiverComponent receiver = new ReceiverComponent();
+ jbi.activateComponent(receiver, "receiver");
+
+ EndpointReferenceType consumer = createEPR(ReceiverComponent.SERVICE, ReceiverComponent.ENDPOINT);
+ wsnBroker.subscribe(consumer, "myTopic", null, true);
+
+ wsnBroker.notify("myTopic", parse("<hello>world</hello>"));
+ // Wait for notification
+ Thread.sleep(50);
+
+ receiver.getMessageList().assertMessagesReceived(1);
+ NormalizedMessage msg = (NormalizedMessage) receiver.getMessageList().getMessages().get(0);
+ Node node = new SourceTransformer().toDOMNode(msg);
+ assertEquals("hello", node.getLocalName());
// Wait for acks to be processed
Thread.sleep(50);