Author: ay
Date: Mon Mar 5 23:57:14 2012
New Revision: 1297296
URL: http://svn.apache.org/viewvc?rev=1297296&view=rev
Log:
[CXF-4164] Robust-InOnly processing with WS-RM must delay updating the sequence
until message delivery
Added:
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckBase.java
(with props)
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckPersistenceTest.java
(with props)
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/sync-ack-persistent-server.xml
(with props)
Modified:
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckTest.java
Modified:
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java?rev=1297296&r1=1297295&r2=1297296&view=diff
==============================================================================
---
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java
(original)
+++
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java
Mon Mar 5 23:57:14 2012
@@ -23,6 +23,7 @@ import java.util.logging.Logger;
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.message.Message;
+import org.apache.cxf.message.MessageUtils;
import org.apache.cxf.phase.Phase;
/**
@@ -42,6 +43,12 @@ public class RMDeliveryInterceptor exten
public void handle(Message message) throws SequenceFault, RMException {
LOG.entering(getClass().getName(), "handleMessage");
- getManager().getDestination(message).processingComplete(message);
+ Destination dest = getManager().getDestination(message);
+ final boolean robust =
+
MessageUtils.isTrue(message.getContextualProperty(Message.ROBUST_ONEWAY));
+ if (robust) {
+ dest.acknowledge(message);
+ }
+ dest.processingComplete(message);
}
}
Modified:
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java?rev=1297296&r1=1297295&r2=1297296&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java
(original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java
Mon Mar 5 23:57:14 2012
@@ -25,6 +25,7 @@ import java.util.logging.Logger;
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.message.Message;
+import org.apache.cxf.message.MessageUtils;
import org.apache.cxf.ws.addressing.AddressingProperties;
import org.apache.cxf.ws.addressing.ContextUtils;
import org.apache.cxf.ws.addressing.MAPAggregator;
@@ -150,7 +151,11 @@ public class RMInInterceptor extends Abs
void processSequence(Destination destination, Message message)
throws SequenceFault, RMException {
- destination.acknowledge(message);
+ final boolean robust =
+
MessageUtils.isTrue(message.getContextualProperty(Message.ROBUST_ONEWAY));
+ if (!robust) {
+ destination.acknowledge(message);
+ }
}
void processDeliveryAssurance(RMProperties rmps) {
Added:
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckBase.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckBase.java?rev=1297296&view=auto
==============================================================================
---
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckBase.java
(added)
+++
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckBase.java
Mon Mar 5 23:57:14 2012
@@ -0,0 +1,206 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.cxf.systest.ws.rm;
+
+import java.net.MalformedURLException;
+import java.util.logging.Logger;
+
+import javax.xml.ws.Endpoint;
+
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
+import org.apache.cxf.bus.spring.SpringBusFactory;
+import org.apache.cxf.common.logging.LogUtils;
+import org.apache.cxf.greeter_control.Control;
+import org.apache.cxf.greeter_control.ControlService;
+import org.apache.cxf.greeter_control.Greeter;
+import org.apache.cxf.greeter_control.GreeterService;
+import org.apache.cxf.greeter_control.types.FaultLocation;
+import org.apache.cxf.interceptor.ServiceInvokerInterceptor;
+import org.apache.cxf.phase.Phase;
+import org.apache.cxf.test.TestUtilities;
+import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
+import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
+import org.apache.cxf.ws.rm.RMManager;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Tests the acknowledgement delivery back to the non-decoupled port when
there is some
+ * error at the provider side and how its behavior is affected by the robust
in-only mode setting.
+ */
+public abstract class ServiceInvocationAckBase extends
AbstractBusClientServerTestBase {
+ public static final String PORT = allocatePort(Server.class);
+
+ private static final Logger LOG =
LogUtils.getLogger(ServiceInvocationAckBase.class);
+
+ private static final String CONTROL_PORT_ADDRESS =
+ "http://localhost:"; + PORT + "/SoapContext/ControlPort";
+
+ public static class Server extends AbstractBusTestServerBase {
+
+ protected void run() {
+ SpringBusFactory factory = new SpringBusFactory();
+ Bus bus = factory.createBus();
+ BusFactory.setDefaultBus(bus);
+ setBus(bus);
+
+ ControlImpl implementor = new ControlImpl();
+ implementor.setAddress("http://localhost:"; + PORT +
"/SoapContext/GreeterPort");
+ GreeterImpl greeterImplementor = new GreeterImpl();
+ implementor.setImplementor(greeterImplementor);
+ Endpoint.publish(CONTROL_PORT_ADDRESS, implementor);
+ LOG.fine("Published control endpoint.");
+ }
+
+ public static void main(String[] args) {
+ try {
+ Server s = new Server();
+ s.start();
+ } catch (Exception ex) {
+ ex.printStackTrace();
+ System.exit(-1);
+ } finally {
+ System.out.println("done!");
+ }
+ }
+ }
+
+ private Bus controlBus;
+ private Control control;
+ private Bus greeterBus;
+ private Greeter greeter;
+
+
+ @BeforeClass
+ public static void startServers() throws Exception {
+ TestUtilities.setKeepAliveSystemProperty(false);
+ assertTrue("server did not launch correctly",
launchServer(Server.class, true));
+ }
+
+ @AfterClass
+ public static void cleanup() {
+ TestUtilities.recoverKeepAliveSystemProperty();
+ }
+
+ @After
+ public void tearDown() {
+ if (null != greeter) {
+ assertTrue("Failed to stop greeter.", control.stopGreeter(null));
+ greeterBus.shutdown(true);
+ greeterBus = null;
+ }
+ if (null != control) {
+ assertTrue("Failed to stop greeter", control.stopGreeter(null));
+ controlBus.shutdown(true);
+ }
+ }
+
+ protected void setupGreeter() throws Exception {
+ }
+
+ @Test
+ public void testDefaultInvocationHandling() throws Exception {
+ setupGreeter();
+
+ control.setRobustInOnlyMode(false);
+
+ FaultLocation location = new
org.apache.cxf.greeter_control.types.ObjectFactory()
+ .createFaultLocation();
+ location.setPhase(Phase.INVOKE);
+ location.setBefore(ServiceInvokerInterceptor.class.getName());
+
+ RMManager manager = greeterBus.getExtension(RMManager.class);
+
+ // the message is acked and the invocation takes place
+ greeter.greetMeOneWay("one");
+ Thread.sleep(6000L);
+ assertTrue("RetransmissionQueue must be empty",
manager.getRetransmissionQueue().isEmpty());
+
+ control.setFaultLocation(location);
+
+ // the invocation fails but the message is acked because the delivery
succeeds
+ greeter.greetMeOneWay("two");
+ Thread.sleep(6000L);
+ assertTrue("RetransmissionQueue must be empty",
manager.getRetransmissionQueue().isEmpty());
+ }
+
+ @Test
+ public void testRobustInvocationHandling() throws Exception {
+ setupGreeter();
+
+ control.setRobustInOnlyMode(true);
+
+ FaultLocation location = new
org.apache.cxf.greeter_control.types.ObjectFactory()
+ .createFaultLocation();
+ location.setPhase(Phase.INVOKE);
+ location.setBefore(ServiceInvokerInterceptor.class.getName());
+
+ RMManager manager = greeterBus.getExtension(RMManager.class);
+
+
+ // the message is acked and the invocation takes place
+ greeter.greetMeOneWay("one");
+ Thread.sleep(6000L);
+ assertTrue("RetransmissionQueue must be empty",
manager.getRetransmissionQueue().isEmpty());
+
+ control.setFaultLocation(location);
+
+ // the invocation fails but the message is acked because the delivery
succeeds
+ greeter.greetMeOneWay("two");
+ Thread.sleep(6000L);
+ assertFalse("RetransmissionQueue must not be empty",
manager.getRetransmissionQueue().isEmpty());
+
+ location.setPhase(null);
+ control.setFaultLocation(location);
+
+ // the retransmission succeeds and the invocation succeeds, the
message is acked
+ Thread.sleep(6000L);
+ assertTrue("RetransmissionQueue must be empty",
manager.getRetransmissionQueue().isEmpty());
+
+ }
+
+ protected void setupGreeter(String cfgResource) throws
NumberFormatException, MalformedURLException {
+
+ SpringBusFactory bf = new SpringBusFactory();
+
+ controlBus = bf.createBus();
+ BusFactory.setDefaultBus(controlBus);
+
+ ControlService cs = new ControlService();
+ control = cs.getControlPort();
+ updateAddressPort(control, PORT);
+
+ assertTrue("Failed to start greeter",
control.startGreeter(cfgResource));
+
+ greeterBus = bf.createBus(cfgResource);
+ BusFactory.setDefaultBus(greeterBus);
+ LOG.fine("Initialised greeter bus with configuration: " + cfgResource);
+
+ GreeterService gs = new GreeterService();
+
+ greeter = gs.getGreeterPort();
+ updateAddressPort(greeter, PORT);
+ LOG.fine("Created greeter client.");
+
+ }
+}
Propchange:
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckBase.java
------------------------------------------------------------------------------
svn:executable = *
Added:
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckPersistenceTest.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckPersistenceTest.java?rev=1297296&view=auto
==============================================================================
---
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckPersistenceTest.java
(added)
+++
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckPersistenceTest.java
Mon Mar 5 23:57:14 2012
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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.apache.cxf.systest.ws.rm;
+
+import org.apache.cxf.ws.rm.persistence.jdbc.RMTxStore;
+
+import org.junit.BeforeClass;
+
+/**
+ * Tests the acknowledgement delivery back to the non-decoupled port when
there is some
+ * error at the provider side and how its behavior is affected by the robust
in-only mode setting.
+ */
+public class ServiceInvocationAckPersistenceTest extends
ServiceInvocationAckBase {
+ @BeforeClass
+ public static void cleanUpDerby() throws Exception {
+ String derbyHome = System.getProperty("derby.system.home");
+ if (derbyHome == null) {
+ System.setProperty("derby.system.home", "target/derby");
+ }
+ RMTxStore.deleteDatabaseFiles();
+ derbyHome = System.getProperty("derby.system.home");
+ try {
+ System.setProperty("derby.system.home", derbyHome + "-server");
+ RMTxStore.deleteDatabaseFiles();
+ } finally {
+ if (derbyHome != null) {
+ System.setProperty("derby.system.home", derbyHome);
+ } else {
+ System.clearProperty("derby.system.home");
+ }
+ }
+ }
+
+ protected void setupGreeter() throws Exception {
+
setupGreeter("org/apache/cxf/systest/ws/rm/sync-ack-persistent-server.xml");
+ }
+}
Propchange:
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckPersistenceTest.java
------------------------------------------------------------------------------
svn:executable = *
Modified:
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckTest.java
URL:
http://svn.apache.org/viewvc/cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckTest.java?rev=1297296&r1=1297295&r2=1297296&view=diff
==============================================================================
---
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckTest.java
(original)
+++
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ServiceInvocationAckTest.java
Mon Mar 5 23:57:14 2012
@@ -18,186 +18,12 @@
*/
package org.apache.cxf.systest.ws.rm;
-import java.net.MalformedURLException;
-import java.util.logging.Logger;
-
-import javax.xml.ws.Endpoint;
-
-import org.apache.cxf.Bus;
-import org.apache.cxf.BusFactory;
-import org.apache.cxf.bus.spring.SpringBusFactory;
-import org.apache.cxf.common.logging.LogUtils;
-import org.apache.cxf.greeter_control.Control;
-import org.apache.cxf.greeter_control.ControlService;
-import org.apache.cxf.greeter_control.Greeter;
-import org.apache.cxf.greeter_control.GreeterService;
-import org.apache.cxf.greeter_control.types.FaultLocation;
-import org.apache.cxf.interceptor.ServiceInvokerInterceptor;
-import org.apache.cxf.phase.Phase;
-import org.apache.cxf.test.TestUtilities;
-import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
-import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
-import org.apache.cxf.ws.rm.RMManager;
-
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
/**
* Tests the acknowledgement delivery back to the non-decoupled port when
there is some
* error at the provider side and how its behavior is affected by the robust
in-only mode setting.
*/
-public class ServiceInvocationAckTest extends AbstractBusClientServerTestBase {
- public static final String PORT = allocatePort(Server.class);
-
- private static final Logger LOG =
LogUtils.getLogger(ServiceInvocationAckTest.class);
-
- private static final String CONTROL_PORT_ADDRESS =
- "http://localhost:"; + PORT + "/SoapContext/ControlPort";
-
- public static class Server extends AbstractBusTestServerBase {
-
- protected void run() {
- SpringBusFactory factory = new SpringBusFactory();
- Bus bus = factory.createBus();
- BusFactory.setDefaultBus(bus);
- setBus(bus);
-
- ControlImpl implementor = new ControlImpl();
- implementor.setAddress("http://localhost:"; + PORT +
"/SoapContext/GreeterPort");
- GreeterImpl greeterImplementor = new GreeterImpl();
- implementor.setImplementor(greeterImplementor);
- Endpoint.publish(CONTROL_PORT_ADDRESS, implementor);
- LOG.fine("Published control endpoint.");
- }
-
- public static void main(String[] args) {
- try {
- Server s = new Server();
- s.start();
- } catch (Exception ex) {
- ex.printStackTrace();
- System.exit(-1);
- } finally {
- System.out.println("done!");
- }
- }
- }
-
- private Bus controlBus;
- private Control control;
- private Bus greeterBus;
- private Greeter greeter;
-
-
- @BeforeClass
- public static void startServers() throws Exception {
- TestUtilities.setKeepAliveSystemProperty(false);
- assertTrue("server did not launch correctly",
launchServer(Server.class, true));
- }
-
- @AfterClass
- public static void cleanup() {
- TestUtilities.recoverKeepAliveSystemProperty();
- }
-
- @After
- public void tearDown() {
- if (null != greeter) {
- assertTrue("Failed to stop greeter.", control.stopGreeter(null));
- greeterBus.shutdown(true);
- greeterBus = null;
- }
- if (null != control) {
- assertTrue("Failed to stop greeter", control.stopGreeter(null));
- controlBus.shutdown(true);
- }
- }
-
- @Test
- public void testDefaultInvocationHandling() throws Exception {
+public class ServiceInvocationAckTest extends ServiceInvocationAckBase {
+ protected void setupGreeter() throws Exception {
setupGreeter("org/apache/cxf/systest/ws/rm/sync-ack-server.xml");
-
- control.setRobustInOnlyMode(false);
-
- FaultLocation location = new
org.apache.cxf.greeter_control.types.ObjectFactory()
- .createFaultLocation();
- location.setPhase(Phase.INVOKE);
- location.setBefore(ServiceInvokerInterceptor.class.getName());
-
- RMManager manager = greeterBus.getExtension(RMManager.class);
-
- // the message is acked and the invocation takes place
- greeter.greetMeOneWay("one");
- Thread.sleep(6000L);
- assertTrue("RetransmissionQueue must be empty",
manager.getRetransmissionQueue().isEmpty());
-
- control.setFaultLocation(location);
-
- // the invocation fails but the message is acked because the delivery
succeeds
- greeter.greetMeOneWay("two");
- Thread.sleep(6000L);
- assertTrue("RetransmissionQueue must be empty",
manager.getRetransmissionQueue().isEmpty());
- }
-
- @Test
- public void testRobustInvocationHandling() throws Exception {
- setupGreeter("org/apache/cxf/systest/ws/rm/sync-ack-server.xml");
-
- control.setRobustInOnlyMode(true);
-
- FaultLocation location = new
org.apache.cxf.greeter_control.types.ObjectFactory()
- .createFaultLocation();
- location.setPhase(Phase.INVOKE);
- location.setBefore(ServiceInvokerInterceptor.class.getName());
-
- RMManager manager = greeterBus.getExtension(RMManager.class);
-
-
- // the message is acked and the invocation takes place
- greeter.greetMeOneWay("one");
- Thread.sleep(6000L);
- assertTrue("RetransmissionQueue must be empty",
manager.getRetransmissionQueue().isEmpty());
-
- control.setFaultLocation(location);
-
- // the invocation fails but the message is acked because the delivery
succeeds
- greeter.greetMeOneWay("two");
- Thread.sleep(6000L);
- assertFalse("RetransmissionQueue must not be empty",
manager.getRetransmissionQueue().isEmpty());
-
- location.setPhase(null);
- control.setFaultLocation(location);
-
- // the retransmission succeeds and the invocation succeeds, the
message is acked
- Thread.sleep(6000L);
- assertTrue("RetransmissionQueue must be empty",
manager.getRetransmissionQueue().isEmpty());
-
- }
-
- private void setupGreeter(String cfgResource) throws
NumberFormatException, MalformedURLException {
-
- SpringBusFactory bf = new SpringBusFactory();
-
- controlBus = bf.createBus();
- BusFactory.setDefaultBus(controlBus);
-
- ControlService cs = new ControlService();
- control = cs.getControlPort();
- updateAddressPort(control, PORT);
-
- assertTrue("Failed to start greeter",
control.startGreeter(cfgResource));
-
- greeterBus = bf.createBus(cfgResource);
- BusFactory.setDefaultBus(greeterBus);
- LOG.fine("Initialised greeter bus with configuration: " + cfgResource);
-
- GreeterService gs = new GreeterService();
-
- greeter = gs.getGreeterPort();
- updateAddressPort(greeter, PORT);
- LOG.fine("Created greeter client.");
-
}
}
Added:
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/sync-ack-persistent-server.xml
URL:
http://svn.apache.org/viewvc/cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/sync-ack-persistent-server.xml?rev=1297296&view=auto
==============================================================================
Binary file - no diff available.
Propchange:
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/sync-ack-persistent-server.xml
------------------------------------------------------------------------------
svn:executable = *
Propchange:
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/sync-ack-persistent-server.xml
------------------------------------------------------------------------------
svn:mime-type = application/xml
