Author: dkulp
Date: Fri Mar 9 18:18:09 2012
New Revision: 1298957
URL: http://svn.apache.org/viewvc?rev=1298957&view=rev
Log:
[CXF-4140] Make registerPublisher compliant with the spec
Modified:
cxf/trunk/services/wsn/wsn-api/src/main/java/org/apache/cxf/wsn/client/Publisher.java
cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractNotificationBroker.java
cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractPublisher.java
cxf/trunk/services/wsn/wsn-core/src/test/java/org/apache/cxf/wsn/WsnBrokerTest.java
Modified:
cxf/trunk/services/wsn/wsn-api/src/main/java/org/apache/cxf/wsn/client/Publisher.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/services/wsn/wsn-api/src/main/java/org/apache/cxf/wsn/client/Publisher.java?rev=1298957&r1=1298956&r2=1298957&view=diff
==============================================================================
---
cxf/trunk/services/wsn/wsn-api/src/main/java/org/apache/cxf/wsn/client/Publisher.java
(original)
+++
cxf/trunk/services/wsn/wsn-api/src/main/java/org/apache/cxf/wsn/client/Publisher.java
Fri Mar 9 18:18:09 2012
@@ -79,15 +79,24 @@ public class Publisher implements Notifi
public Publisher(Callback callback, String address) {
this.callback = callback;
this.address = address;
- this.endpoint = Endpoint.create(this);
- this.endpoint.publish(address);
+ if (callback == null || address == null) {
+ this.endpoint = null;
+ } else {
+ this.endpoint = Endpoint.create(this);
+ this.endpoint.publish(address);
+ }
}
public void stop() {
- this.endpoint.stop();
+ if (endpoint != null) {
+ this.endpoint.stop();
+ }
}
public W3CEndpointReference getEpr() {
+ if (this.endpoint == null) {
+ return null;
+ }
return this.endpoint.getEndpointReference(W3CEndpointReference.class);
}
Modified:
cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractNotificationBroker.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractNotificationBroker.java?rev=1298957&r1=1298956&r2=1298957&view=diff
==============================================================================
---
cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractNotificationBroker.java
(original)
+++
cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractNotificationBroker.java
Fri Mar 9 18:18:09 2012
@@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -92,6 +93,8 @@ public abstract class AbstractNotificati
private AbstractPublisher anonymousPublisher;
private Map<String, AbstractPublisher> publishers;
+
+ private List<AbstractPublisher> nonContactPublishers;
private Map<String, AbstractSubscription> subscriptions;
@@ -100,6 +103,7 @@ public abstract class AbstractNotificati
idGenerator = new IdGenerator();
subscriptions = new ConcurrentHashMap<String, AbstractSubscription>();
publishers = new ConcurrentHashMap<String, AbstractPublisher>();
+ nonContactPublishers = new CopyOnWriteArrayList<AbstractPublisher>();
}
@Override
@@ -332,7 +336,11 @@ public abstract class AbstractNotificati
publisher.create(registerPublisherRequest);
RegisterPublisherResponse response = new
RegisterPublisherResponse();
response.setPublisherRegistrationReference(publisher.getEpr());
-
publishers.put(WSNHelper.getWSAAddress(publisher.getPublisherReference()),
publisher);
+ if (publisher.getPublisherReference() != null) {
+
publishers.put(WSNHelper.getWSAAddress(publisher.getPublisherReference()),
publisher);
+ } else {
+ nonContactPublishers.add(publisher);
+ }
success = true;
return response;
} catch (EndpointRegistrationException e) {
Modified:
cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractPublisher.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractPublisher.java?rev=1298957&r1=1298956&r2=1298957&view=diff
==============================================================================
---
cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractPublisher.java
(original)
+++
cxf/trunk/services/wsn/wsn-core/src/main/java/org/apache/cxf/wsn/AbstractPublisher.java
Fri Mar 9 18:18:09 2012
@@ -112,7 +112,7 @@ public abstract class AbstractPublisher
? registerPublisherRequest.isDemand().booleanValue()
: false;
// Check all parameters
- if (publisherReference == null) {
+ if (publisherReference == null && demand) {
PublisherRegistrationFailedFaultType fault = new
PublisherRegistrationFailedFaultType();
throw new PublisherRegistrationFailedFault("Invalid
PublisherReference: null", fault);
}
Modified:
cxf/trunk/services/wsn/wsn-core/src/test/java/org/apache/cxf/wsn/WsnBrokerTest.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/services/wsn/wsn-core/src/test/java/org/apache/cxf/wsn/WsnBrokerTest.java?rev=1298957&r1=1298956&r2=1298957&view=diff
==============================================================================
---
cxf/trunk/services/wsn/wsn-core/src/test/java/org/apache/cxf/wsn/WsnBrokerTest.java
(original)
+++
cxf/trunk/services/wsn/wsn-core/src/test/java/org/apache/cxf/wsn/WsnBrokerTest.java
Fri Mar 9 18:18:09 2012
@@ -186,7 +186,31 @@ public abstract class WsnBrokerTest exte
publisher.stop();
consumer.stop();
}
+ public void testNullPublisherReference() throws Exception {
+ TestConsumer consumerCallback = new TestConsumer();
+ Consumer consumer = new Consumer(consumerCallback, "http://localhost:";
+ port2 + "/test/consumer");
+
+ Subscription subscription = notificationBroker.subscribe(consumer,
"myTopicNullEPR");
+
+ Publisher publisher = new Publisher(null, null);
+ Registration registration =
notificationBroker.registerPublisher(publisher, "myTopicNullEPR", false);
+ synchronized (consumerCallback.notifications) {
+ notificationBroker.notify(publisher, "myTopicNullEPR",
+ new JAXBElement<String>(new
QName("urn:test:org", "foo"),
+ String.class, "bar"));
+ consumerCallback.notifications.wait(1000000);
+ }
+ assertEquals(1, consumerCallback.notifications.size());
+ NotificationMessageHolderType message =
consumerCallback.notifications.get(0);
+ assertEquals(WSNHelper.getWSAAddress(subscription.getEpr()),
+
WSNHelper.getWSAAddress(message.getSubscriptionReference()));
+
+ subscription.unsubscribe();
+ registration.destroy();
+ publisher.stop();
+ consumer.stop();
+ }
public void testPublisherOnDemand() throws Exception {
TestConsumer consumerCallback = new TestConsumer();
Consumer consumer = new Consumer(consumerCallback, "http://localhost:";
+ port2 + "/test/consumer");
