You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cxf.apache.org by ds...@apache.org on 2011/07/25 07:56:18 UTC
svn commit: r1150554 [1/2] - in /cxf/trunk:
rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/
rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/
rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/
rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/...
Author: dsosnoski
Date: Mon Jul 25 05:56:11 2011
New Revision: 1150554
URL: http://svn.apache.org/viewvc?rev=1150554&view=rev
Log:
Restructure WS-RM code to support multiple RMEndpoints with different protocol variations for a single logical endpoint, change tests to match new code.
Added:
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ProtocolVariation.java
Modified:
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/AbstractSequence.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Messages.properties
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Proxy.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM10Constants.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMConstants.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RetransmissionCallback.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Source.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/SourceSequence.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/VersionTransformer.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMStore.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/Messages.properties
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInterceptor.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/SoapFaultFactory.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationTest.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/ProxyTest.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMEndpointTest.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMInInterceptorTest.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerConfigurationTest.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/ServantTest.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/SourceSequenceTest.java
cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStoreTest.java
cxf/trunk/systests/ws-specs/src/test/java/org/apache/cxf/systest/ws/rm/ClientPersistenceTest.java
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/AbstractSequence.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/AbstractSequence.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/AbstractSequence.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/AbstractSequence.java Mon Jul 25 05:56:11 2011
@@ -26,10 +26,12 @@ import org.apache.cxf.ws.rm.v200702.Sequ
public abstract class AbstractSequence {
protected final Identifier id;
+ private final ProtocolVariation protocol;
protected SequenceAcknowledgement acknowledgement;
- protected AbstractSequence(Identifier i) {
+ protected AbstractSequence(Identifier i, ProtocolVariation p) {
id = i;
+ protocol = p;
}
/**
@@ -39,6 +41,10 @@ public abstract class AbstractSequence {
return id;
}
+ public ProtocolVariation getProtocol() {
+ return protocol;
+ }
+
public String toString() {
return id.getValue();
}
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java Mon Jul 25 05:56:11 2011
@@ -65,13 +65,17 @@ public class Destination extends Abstrac
}
public void addSequence(DestinationSequence seq, boolean persist) {
- seq.setDestination(this);
- map.put(seq.getIdentifier().getValue(), seq);
- if (persist) {
- RMStore store = getReliableEndpoint().getManager().getStore();
- if (null != store) {
- store.createDestinationSequence(seq);
+ if (seq.getProtocol() == getReliableEndpoint().getProtocol()) {
+ seq.setDestination(this);
+ map.put(seq.getIdentifier().getValue(), seq);
+ if (persist) {
+ RMStore store = getReliableEndpoint().getManager().getStore();
+ if (null != store) {
+ store.createDestinationSequence(seq);
+ }
}
+ } else {
+ LOG.log(Level.SEVERE, "Incompatible protocol version");
}
}
@@ -134,7 +138,7 @@ public class Destination extends Abstrac
}
}
} else {
- RMConstants consts = getReliableEndpoint().getEncoderDecoder().getConstants();
+ RMConstants consts = getReliableEndpoint().getProtocol().getConstants();
SequenceFaultFactory sff = new SequenceFaultFactory(consts);
throw sff.createUnknownSequenceFault(sequenceType.getIdentifier());
}
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java Mon Jul 25 05:56:11 2011
@@ -61,14 +61,14 @@ public class DestinationSequence extends
private long highNumberCompleted;
private List<Continuation> continuations = new LinkedList<Continuation>();
- public DestinationSequence(Identifier i, EndpointReferenceType a, Destination d) {
- this(i, a, 0, null);
+ public DestinationSequence(Identifier i, EndpointReferenceType a, Destination d, ProtocolVariation pv) {
+ this(i, a, 0, null, pv);
destination = d;
}
public DestinationSequence(Identifier i, EndpointReferenceType a,
- long lmn, SequenceAcknowledgement ac) {
- super(i);
+ long lmn, SequenceAcknowledgement ac, ProtocolVariation pv) {
+ super(i, pv);
acksTo = a;
lastMessageNumber = lmn;
acknowledgement = ac;
@@ -113,7 +113,7 @@ public class DestinationSequence extends
long messageNumber = st.getMessageNumber().longValue();
LOG.fine("Acknowledging message: " + messageNumber);
if (0 != lastMessageNumber && messageNumber > lastMessageNumber) {
- RMConstants consts = destination.getReliableEndpoint().getEncoderDecoder().getConstants();
+ RMConstants consts = getProtocol().getConstants();
SequenceFaultFactory sff = new SequenceFaultFactory(consts);
throw sff.createSequenceTerminatedFault(st.getIdentifier(), false);
}
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Messages.properties
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Messages.properties?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Messages.properties (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Messages.properties Mon Jul 25 05:56:11 2011
@@ -36,6 +36,7 @@ INVALID_ACK_EXC = The SequenceAcknowledg
MESSAGE_NR_ROLLOVER_EXC = The maximum value for wsrm:MessageNumber has been exceeded.
SEQUENCE_CLOSED_EXC = The Sequence has been closed.
WSRM_REQUIRED_EXC = WS-ReliableMessaging is required by this endpoint.
+UNSUPPORTED_NAMESPACE = WS-Addressing namespace {0} is not supported with WS-ReliableMessaging namespace {1}.
CREATE_SEQ_REFUSED = The create sequence request has been refused by the RM destination.
SEQ_FAULT_MSG = Interceptor encountered a SequenceFault: {0}.
Added: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ProtocolVariation.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ProtocolVariation.java?rev=1150554&view=auto
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ProtocolVariation.java (added)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ProtocolVariation.java Mon Jul 25 05:56:11 2011
@@ -0,0 +1,70 @@
+/**
+ * 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.ws.rm;
+
+/**
+ * Supported protocol variations.
+ */
+public enum ProtocolVariation
+{
+ RM10WSA200408(EncoderDecoder10Impl.INSTANCE, RM10Constants.INSTANCE),
+ RM10WSA200508(EncoderDecoder10AImpl.INSTANCE, RM10Constants.INSTANCE),
+ RM11WSA200508(EncoderDecoder11Impl.INSTANCE, RM11Constants.INSTANCE);
+
+ private final EncoderDecoder codec;
+ private final RMConstants constants;
+
+ private ProtocolVariation(EncoderDecoder ed, RMConstants rmc) {
+ codec = ed;
+ constants = rmc;
+ }
+
+ public EncoderDecoder getCodec() {
+ return codec;
+ }
+
+ public RMConstants getConstants() {
+ return constants;
+ }
+
+ public String getWSRMNamespace() {
+ return codec.getWSRMNamespace();
+ }
+
+ public String getWSANamespace() {
+ return codec.getWSANamespace();
+ }
+
+ /**
+ * Get the information for a supported version of WS-ReliableMessaging and WS-Addressing.
+ *
+ * @param wsrm WS-RM namespace URI
+ * @param wsa WS-A namespace URI
+ * @return variant (<code>null</code> if not a supported version)
+ */
+ public static ProtocolVariation findVariant(String wsrm, String wsa) {
+ for (ProtocolVariation variant: ProtocolVariation.values()) {
+ if (variant.getWSRMNamespace().equals(wsrm) && variant.getWSANamespace().equals(wsa)) {
+ return variant;
+ }
+ }
+ return null;
+ }
+}
\ No newline at end of file
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Proxy.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Proxy.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Proxy.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Proxy.java Mon Jul 25 05:56:11 2011
@@ -79,26 +79,26 @@ public class Proxy {
return;
}
- RMConstants constants = reliableEndpoint.getEncoderDecoder().getConstants();
+ RMConstants constants = reliableEndpoint.getProtocol().getConstants();
OperationInfo oi = reliableEndpoint.getEndpoint().getEndpointInfo().getService().getInterface()
.getOperation(constants.getSequenceAckOperationName());
invoke(oi, new Object[] {ds}, null);
}
void terminate(SourceSequence ss) throws RMException {
- RMConstants constants = reliableEndpoint.getEncoderDecoder().getConstants();
+ RMConstants constants = reliableEndpoint.getProtocol().getConstants();
OperationInfo oi = reliableEndpoint.getEndpoint().getEndpointInfo().getService().getInterface()
.getOperation(constants.getTerminateSequenceOperationName());
TerminateSequenceType ts = new TerminateSequenceType();
ts.setIdentifier(ss.getIdentifier());
- EncoderDecoder codec = reliableEndpoint.getEncoderDecoder();
+ EncoderDecoder codec = reliableEndpoint.getProtocol().getCodec();
invoke(oi, new Object[] {codec.convertToSend(ts)}, null);
}
void createSequenceResponse(final Object createResponse) throws RMException {
LOG.fine("sending CreateSequenceResponse from client side");
- RMConstants constants = reliableEndpoint.getEncoderDecoder().getConstants();
+ RMConstants constants = reliableEndpoint.getProtocol().getConstants();
final OperationInfo oi = reliableEndpoint.getEndpoint().getEndpointInfo().getService().getInterface()
.getOperation(constants.getCreateSequenceResponseOnewayOperationName());
@@ -147,7 +147,7 @@ public class Proxy {
InterfaceInfo ii = reliableEndpoint.getEndpoint().getEndpointInfo().getService().getInterface();
- EncoderDecoder codec = reliableEndpoint.getEncoderDecoder();
+ EncoderDecoder codec = reliableEndpoint.getProtocol().getCodec();
RMConstants constants = codec.getConstants();
final OperationInfo oi = isServer
? ii.getOperation(constants.getCreateSequenceOnewayOperationName())
@@ -197,7 +197,7 @@ public class Proxy {
return;
}
- RMConstants constants = reliableEndpoint.getEncoderDecoder().getConstants();
+ RMConstants constants = reliableEndpoint.getProtocol().getConstants();
OperationInfo oi = reliableEndpoint.getEndpoint().getEndpointInfo().getService().getInterface()
.getOperation(constants.getCloseSequenceOperationName());
// pass reference to source sequence in invocation context
@@ -229,7 +229,7 @@ public class Proxy {
return;
}
- RMConstants constants = reliableEndpoint.getEncoderDecoder().getConstants();
+ RMConstants constants = reliableEndpoint.getProtocol().getConstants();
OperationInfo oi = reliableEndpoint.getEndpoint().getEndpointInfo().getService().getInterface()
.getOperation(constants.getAckRequestedOperationName());
invoke(oi, new Object[] {}, null);
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM10Constants.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM10Constants.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM10Constants.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM10Constants.java Mon Jul 25 05:56:11 2011
@@ -67,7 +67,7 @@ public final class RM10Constants extends
public static final QName SEQUENCE_ACKNOWLEDGEMENT_QNAME =
new QName(NAMESPACE_URI, "SequenceAcknowledgement");
- public static final QName CLOSE_SEQUENCE_QNAME = new QName(NAMESPACE_URI, "LastMessage");
+ public static final QName CLOSE_SEQUENCE_QNAME = new QName(NAMESPACE_URI, "CloseSequence");
public static final QName ACK_REQ_QNAME = new QName(NAMESPACE_URI, "AckRequested");
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMConstants.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMConstants.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMConstants.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMConstants.java Mon Jul 25 05:56:11 2011
@@ -26,8 +26,6 @@ import javax.xml.namespace.QName;
*/
public abstract class RMConstants {
- // namespaces
-
public static final String NAMESPACE_PREFIX = "wsrm";
// WSDL
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java Mon Jul 25 05:56:11 2011
@@ -84,7 +84,7 @@ public class RMEndpoint {
private RMManager manager;
private Endpoint applicationEndpoint;
private Conduit conduit;
- private EncoderDecoder encoderDecoder;
+ private ProtocolVariation protocol;
private EndpointReferenceType replyTo;
private Source source;
private Destination destination;
@@ -103,16 +103,16 @@ public class RMEndpoint {
*
* @param m
* @param ae
- * @param codec
+ * @param pv
*/
- public RMEndpoint(RMManager m, Endpoint ae, EncoderDecoder codec) {
+ public RMEndpoint(RMManager m, Endpoint ae, ProtocolVariation pv) {
manager = m;
applicationEndpoint = ae;
source = new Source(this);
destination = new Destination(this);
proxy = new Proxy(this);
servant = new Servant(this);
- encoderDecoder = codec;
+ protocol = pv;
}
/**
@@ -136,24 +136,18 @@ public class RMEndpoint {
return endpoint;
}
- /**
- * Get the encoder/decoder used by this endpoint.
- *
- * @return URI
- */
- public EncoderDecoder getEncoderDecoder() {
- return encoderDecoder;
+ public ProtocolVariation getProtocol() {
+ return protocol;
}
/**
- * Set the encoder/decoder used by this endpoint. This should only be used if the encoder/decoder was not
- * set by the constructor, and only used once in that case - once set the encoder/decoder should never be
- * changed.
+ * Set the protocol used by this endpoint. This method is only intended for use in testing; all normal use
+ * uses the constructor to set the value.
*
- * @param codec
+ * @param protocol
*/
- public void setEncoderDecoder(EncoderDecoder codec) {
- encoderDecoder = codec;
+ public void setProtocol(ProtocolVariation protocol) {
+ this.protocol = protocol;
}
/**
@@ -271,7 +265,7 @@ public class RMEndpoint {
void createService() {
ServiceInfo si = new ServiceInfo();
si.setProperty(Schema.class.getName(), getSchema());
- serviceQName = new QName(encoderDecoder.getWSRMNamespace(), SERVICE_NAME);
+ serviceQName = new QName(protocol.getWSRMNamespace(), SERVICE_NAME);
si.setName(serviceQName);
buildInterfaceInfo(si);
@@ -279,7 +273,7 @@ public class RMEndpoint {
service = new WrappedService(applicationEndpoint.getService(), serviceQName, si);
DataBinding dataBinding = null;
- Class create = encoderDecoder.getCreateSequenceType();
+ Class create = protocol.getCodec().getCreateSequenceType();
try {
JAXBContext ctx =
JAXBContext.newInstance(PackageUtils.getPackageName(create), create.getClassLoader());
@@ -324,7 +318,7 @@ public class RMEndpoint {
ei.setAddress(aei.getAddress());
- ei.setName(RMUtils.getConstants(encoderDecoder.getWSRMNamespace()).getPortName());
+ ei.setName(RMUtils.getConstants(protocol.getWSRMNamespace()).getPortName());
ei.setBinding(si.getBinding(bindingQName));
// if addressing was enabled on the application endpoint by means
@@ -380,7 +374,7 @@ public class RMEndpoint {
}
void buildInterfaceInfo(ServiceInfo si) {
- interfaceQName = new QName(encoderDecoder.getWSRMNamespace(), INTERFACE_NAME);
+ interfaceQName = new QName(protocol.getWSRMNamespace(), INTERFACE_NAME);
InterfaceInfo ii = new InterfaceInfo(si, interfaceQName);
buildOperationInfo(ii);
}
@@ -400,7 +394,7 @@ public class RMEndpoint {
OperationInfo operationInfo = null;
MessagePartInfo partInfo = null;
MessageInfo messageInfo = null;
- RMConstants consts = RMUtils.getConstants(encoderDecoder.getWSRMNamespace());
+ RMConstants consts = protocol.getConstants();
operationInfo = ii.addOperation(consts.getCreateSequenceOperationName());
messageInfo = operationInfo.createMessage(consts.getCreateSequenceOperationName(),
MessageInfo.Type.INPUT);
@@ -408,7 +402,7 @@ public class RMEndpoint {
partInfo = messageInfo.addMessagePart(CREATE_PART_NAME);
partInfo.setElementQName(consts.getCreateSequenceOperationName());
partInfo.setElement(true);
- partInfo.setTypeClass(encoderDecoder.getCreateSequenceType());
+ partInfo.setTypeClass(protocol.getCodec().getCreateSequenceType());
messageInfo = operationInfo.createMessage(consts.getCreateSequenceResponseOperationName(),
MessageInfo.Type.OUTPUT);
@@ -416,7 +410,7 @@ public class RMEndpoint {
partInfo = messageInfo.addMessagePart(CREATE_RESPONSE_PART_NAME);
partInfo.setElementQName(consts.getCreateSequenceResponseOperationName());
partInfo.setElement(true);
- partInfo.setTypeClass(encoderDecoder.getCreateSequenceResponseType());
+ partInfo.setTypeClass(protocol.getCodec().getCreateSequenceResponseType());
partInfo.setIndex(0);
operationInfo = ii.addOperation(consts.getCreateSequenceOnewayOperationName());
@@ -426,7 +420,7 @@ public class RMEndpoint {
partInfo = messageInfo.addMessagePart(CREATE_PART_NAME);
partInfo.setElementQName(consts.getCreateSequenceOnewayOperationName());
partInfo.setElement(true);
- partInfo.setTypeClass(encoderDecoder.getCreateSequenceType());
+ partInfo.setTypeClass(protocol.getCodec().getCreateSequenceType());
operationInfo = ii.addOperation(consts.getCreateSequenceResponseOnewayOperationName());
messageInfo = operationInfo.createMessage(consts.getCreateSequenceResponseOnewayOperationName(),
@@ -435,7 +429,7 @@ public class RMEndpoint {
partInfo = messageInfo.addMessagePart(CREATE_RESPONSE_PART_NAME);
partInfo.setElementQName(consts.getCreateSequenceResponseOnewayOperationName());
partInfo.setElement(true);
- partInfo.setTypeClass(encoderDecoder.getCreateSequenceResponseType());
+ partInfo.setTypeClass(protocol.getCodec().getCreateSequenceResponseType());
}
void buildTerminateSequenceOperationInfo(InterfaceInfo ii) {
@@ -444,7 +438,7 @@ public class RMEndpoint {
MessagePartInfo partInfo = null;
MessageInfo messageInfo = null;
- RMConstants consts = RMUtils.getConstants(encoderDecoder.getWSRMNamespace());
+ RMConstants consts = protocol.getConstants();
operationInfo = ii.addOperation(consts.getTerminateSequenceOperationName());
messageInfo = operationInfo.createMessage(consts.getTerminateSequenceOperationName(),
MessageInfo.Type.INPUT);
@@ -452,7 +446,7 @@ public class RMEndpoint {
partInfo = messageInfo.addMessagePart(TERMINATE_PART_NAME);
partInfo.setElementQName(consts.getTerminateSequenceOperationName());
partInfo.setElement(true);
- partInfo.setTypeClass(encoderDecoder.getTerminateSequenceType());
+ partInfo.setTypeClass(protocol.getCodec().getTerminateSequenceType());
}
void buildSequenceAckOperationInfo(InterfaceInfo ii) {
@@ -460,7 +454,7 @@ public class RMEndpoint {
OperationInfo operationInfo = null;
MessageInfo messageInfo = null;
- RMConstants consts = RMUtils.getConstants(encoderDecoder.getWSRMNamespace());
+ RMConstants consts = protocol.getConstants();
operationInfo = ii.addOperation(consts.getSequenceAckOperationName());
messageInfo = operationInfo.createMessage(consts.getSequenceAckOperationName(),
MessageInfo.Type.INPUT);
@@ -472,7 +466,7 @@ public class RMEndpoint {
OperationInfo operationInfo = null;
MessageInfo messageInfo = null;
- RMConstants consts = RMUtils.getConstants(encoderDecoder.getWSRMNamespace());
+ RMConstants consts = protocol.getConstants();
operationInfo = ii.addOperation(consts.getCloseSequenceOperationName());
messageInfo = operationInfo.createMessage(consts.getCloseSequenceOperationName(),
MessageInfo.Type.INPUT);
@@ -484,7 +478,7 @@ public class RMEndpoint {
OperationInfo operationInfo = null;
MessageInfo messageInfo = null;
- RMConstants consts = RMUtils.getConstants(encoderDecoder.getWSRMNamespace());
+ RMConstants consts = protocol.getConstants();
operationInfo = ii.addOperation(consts.getAckRequestedOperationName());
messageInfo = operationInfo.createMessage(consts.getAckRequestedOperationName(),
MessageInfo.Type.INPUT);
@@ -501,11 +495,11 @@ public class RMEndpoint {
SoapVersion sv = sbi.getSoapVersion();
String bindingId = sbi.getBindingId();
SoapBindingInfo bi = new SoapBindingInfo(si, bindingId, sv);
- bindingQName = new QName(encoderDecoder.getWSRMNamespace(), BINDING_NAME);
+ bindingQName = new QName(protocol.getWSRMNamespace(), BINDING_NAME);
bi.setName(bindingQName);
BindingOperationInfo boi = null;
- RMConstants consts = RMUtils.getConstants(encoderDecoder.getWSRMNamespace());
+ RMConstants consts = protocol.getConstants();
boi = bi.buildOperation(consts.getCreateSequenceOperationName(),
consts.getCreateSequenceOperationName().getLocalPart(), null);
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=1150554&r1=1150553&r2=1150554&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 Jul 25 05:56:11 2011
@@ -81,7 +81,7 @@ public class RMInInterceptor extends Abs
boolean isServer = RMContextUtils.isServerSide(message);
LOG.fine("isServerSide: " + isServer);
- boolean isApplicationMessage = !RMContextUtils.isRMProtocolMessage(action);
+ boolean isApplicationMessage = !RMContextUtils.isRMProtocolMessage(action);
LOG.fine("isApplicationMessage: " + isApplicationMessage);
// for application AND out of band messages
@@ -135,7 +135,7 @@ public class RMInInterceptor extends Abs
if (null != ss) {
ss.setAcknowledged(ack);
} else {
- RMConstants consts = rme.getEncoderDecoder().getConstants();
+ RMConstants consts = rme.getProtocol().getConstants();
SequenceFaultFactory sff = new SequenceFaultFactory(consts);
throw sff.createUnknownSequenceFault(id);
}
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java Mon Jul 25 05:56:11 2011
@@ -55,7 +55,6 @@ import org.apache.cxf.ws.addressing.Addr
import org.apache.cxf.ws.addressing.AddressingPropertiesImpl;
import org.apache.cxf.ws.addressing.EndpointReferenceType;
import org.apache.cxf.ws.addressing.MAPAggregator;
-import org.apache.cxf.ws.addressing.Names;
import org.apache.cxf.ws.addressing.RelatesToType;
import org.apache.cxf.ws.addressing.VersionTransformer.Names200408;
import org.apache.cxf.ws.rm.manager.DeliveryAssuranceType;
@@ -95,7 +94,7 @@ public class RMManager implements Server
private RMStore store;
private SequenceIdentifierGenerator idGenerator;
private RetransmissionQueue retransmissionQueue;
- private Map<Endpoint, RMEndpoint> reliableEndpoints = new HashMap<Endpoint, RMEndpoint>();
+ private Map<ProtocolVariation,Map<Endpoint, RMEndpoint>> endpointMaps;
private AtomicReference<Timer> timer = new AtomicReference<Timer>();
private RMAssertion rmAssertion;
private DeliveryAssuranceType deliveryAssurance;
@@ -104,10 +103,16 @@ public class RMManager implements Server
private String rmNamespace = RM10Constants.NAMESPACE_URI;
private String rmAddressingNamespace = Names200408.WSA_NAMESPACE_NAME;
+ public RMManager() {
+ setEndpointMaps(new HashMap<ProtocolVariation,Map<Endpoint, RMEndpoint>>());
+ }
+
// ServerLifeCycleListener
public void startServer(Server server) {
- recoverReliableEndpoint(server.getEndpoint(), (Conduit)null);
+ for (ProtocolVariation protocol : ProtocolVariation.values()) {
+ recoverReliableEndpoint(server.getEndpoint(), (Conduit)null, protocol);
+ }
}
public void stopServer(Server server) {
@@ -122,13 +127,14 @@ public class RMManager implements Server
}
String id = RMUtils.getEndpointIdentifier(client.getEndpoint());
- Collection<SourceSequence> sss = store.getSourceSequences(id);
+ ProtocolVariation protocol = ProtocolVariation.findVariant(getRMNamespace(),
+ getRMAddressingNamespace());
+ Collection<SourceSequence> sss = store.getSourceSequences(id, protocol);
if (null == sss || 0 == sss.size()) {
return;
}
LOG.log(Level.FINE, "Number of source sequences: {0}", sss.size());
-
- recoverReliableEndpoint(client.getEndpoint(), client.getConduit());
+ recoverReliableEndpoint(client.getEndpoint(), client.getConduit(), protocol);
}
public void clientDestroyed(Client client) {
@@ -285,7 +291,7 @@ public class RMManager implements Server
// The real stuff ...
- public synchronized RMEndpoint getReliableEndpoint(Message message) {
+ public synchronized RMEndpoint getReliableEndpoint(Message message) throws RMException {
Endpoint endpoint = message.getExchange().get(Endpoint.class);
QName name = endpoint.getEndpointInfo().getName();
if (LOG.isLoggable(Level.FINE)) {
@@ -295,29 +301,19 @@ public class RMManager implements Server
WrappedEndpoint wrappedEndpoint = (WrappedEndpoint)endpoint;
endpoint = wrappedEndpoint.getWrappedEndpoint();
}
- RMEndpoint rme = reliableEndpoints.get(endpoint);
+ String rmUri = getRMNamespace(message);
+ String addrUri = getAddressingNamespace(message);
+ ProtocolVariation protocol = ProtocolVariation.findVariant(rmUri, addrUri);
+ Map<Endpoint, RMEndpoint> endpointMap = endpointMaps.get(protocol);
+ if (endpointMap == null) {
+ org.apache.cxf.common.i18n.Message msg = new org.apache.cxf.common.i18n.Message(
+ "UNSUPPORTED_NAMESPACE", LOG, addrUri, rmUri);
+ LOG.log(Level.INFO, msg.toString());
+ throw new RMException(msg);
+ }
+ RMEndpoint rme = endpointMap.get(endpoint);
if (null == rme) {
- RMProperties rmps = RMContextUtils.retrieveRMProperties(message, false);
- String rmUri = null;
- if (rmps != null) {
- rmUri = rmps.getNamespaceURI();
- }
- if (rmUri == null) {
- rmUri = getRMNamespace();
- }
- String addrUri = null;
- if (RM10Constants.NAMESPACE_URI.equals(rmUri)) {
- AddressingPropertiesImpl maps = RMContextUtils.retrieveMAPs(message, false, false);
- if (maps != null) {
- addrUri = maps.getNamespaceURI();
- }
- if (addrUri == null) {
- addrUri = getRMAddressingNamespace();
- }
- } else {
- addrUri = Names.WSA_ADDRESS_NAME;
- }
- rme = createReliableEndpoint(endpoint, VersionTransformer.getEncoderDecoder(rmUri, addrUri));
+ rme = createReliableEndpoint(endpoint, protocol);
org.apache.cxf.transport.Destination destination = message.getExchange().getDestination();
EndpointReferenceType replyTo = null;
if (null != destination) {
@@ -330,13 +326,51 @@ public class RMManager implements Server
.getProperty(MAPAggregator.DECOUPLED_DESTINATION,
org.apache.cxf.transport.Destination.class);
rme.initialise(message.getExchange().getConduit(message), replyTo, dest);
- reliableEndpoints.put(endpoint, rme);
+ endpointMap.put(endpoint, rme);
LOG.fine("Created new RMEndpoint.");
}
return rme;
}
- public Destination getDestination(Message message) {
+ /**
+ * Get the WS-Addressing namespace being used for a message. If the WS-Addressing namespace has not been
+ * set, this returns the default configured for this manager.
+ *
+ * @param message
+ * @return namespace URI
+ */
+ String getAddressingNamespace(Message message) {
+ AddressingPropertiesImpl maps = RMContextUtils.retrieveMAPs(message, false, false);
+ String addrUri = null;
+ if (maps != null) {
+ addrUri = maps.getNamespaceURI();
+ }
+ if (addrUri == null) {
+ addrUri = getRMAddressingNamespace();
+ }
+ return addrUri;
+ }
+
+ /**
+ * Get the WS-RM namespace being used for a message. If the WS-RM namespace has not been set, this returns
+ * the default configured for this manager.
+ *
+ * @param message
+ * @return namespace URI
+ */
+ String getRMNamespace(Message message) {
+ RMProperties rmps = RMContextUtils.retrieveRMProperties(message, false);
+ String rmUri = null;
+ if (rmps != null) {
+ rmUri = rmps.getNamespaceURI();
+ }
+ if (rmUri == null) {
+ rmUri = getRMNamespace();
+ }
+ return rmUri;
+ }
+
+ public Destination getDestination(Message message) throws RMException {
RMEndpoint rme = getReliableEndpoint(message);
if (null != rme) {
return rme.getDestination();
@@ -344,7 +378,7 @@ public class RMManager implements Server
return null;
}
- public Source getSource(Message message) {
+ public Source getSource(Message message) throws RMException {
RMEndpoint rme = getReliableEndpoint(message);
if (null != rme) {
return rme.getSource();
@@ -409,11 +443,16 @@ public class RMManager implements Server
@PreDestroy
public void shutdown() {
// shutdown remaining endpoints
-
- LOG.log(Level.FINE, "Shutting down RMManager with {0} remaining endpoints.",
- reliableEndpoints.size());
- for (RMEndpoint rme : reliableEndpoints.values()) {
- rme.shutdown();
+ for (ProtocolVariation protocol : ProtocolVariation.values()) {
+ Map<Endpoint, RMEndpoint> map = endpointMaps.get(protocol);
+ if (map.size() > 0) {
+ LOG.log(Level.FINE,
+ "Shutting down RMManager with {0} remaining endpoints for protocol variation {1}.",
+ new Object[] { new Integer(map.size()), protocol });
+ for (RMEndpoint rme : map.values()) {
+ rme.shutdown();
+ }
+ }
}
// remove references to timer tasks cancelled above to make them
@@ -426,12 +465,18 @@ public class RMManager implements Server
}
synchronized void shutdownReliableEndpoint(Endpoint e) {
- RMEndpoint rme = reliableEndpoints.get(e);
- if (null == rme) {
- // not interested
+ RMEndpoint rme = null;
+ for (ProtocolVariation protocol : ProtocolVariation.values()) {
+ Map<Endpoint, RMEndpoint> map = endpointMaps.get(protocol);
+ rme = map.get(e);
+ if (rme != null) {
+ break;
+ }
+ }
+ if (rme == null) {
+ // not found
return;
}
-
rme.shutdown();
// remove references to timer tasks cancelled above to make them
@@ -441,18 +486,20 @@ public class RMManager implements Server
t.purge();
}
- reliableEndpoints.remove(e);
+ for (ProtocolVariation protocol : ProtocolVariation.values()) {
+ endpointMaps.get(protocol).remove(e);
+ }
}
- void recoverReliableEndpoint(Endpoint endpoint, Conduit conduit) {
+ void recoverReliableEndpoint(Endpoint endpoint, Conduit conduit, ProtocolVariation protocol) {
if (null == store || null == retransmissionQueue) {
return;
}
String id = RMUtils.getEndpointIdentifier(endpoint);
- Collection<SourceSequence> sss = store.getSourceSequences(id);
- Collection<DestinationSequence> dss = store.getDestinationSequences(id);
+ Collection<SourceSequence> sss = store.getSourceSequences(id, protocol);
+ Collection<DestinationSequence> dss = store.getDestinationSequences(id, protocol);
if ((null == sss || 0 == sss.size()) && (null == dss || 0 == dss.size())) {
return;
}
@@ -461,10 +508,9 @@ public class RMManager implements Server
LOG.log(Level.FINE, "Recovering {0} endpoint with id: {1}",
new Object[] {null == conduit ? "client" : "server", id});
- EncoderDecoder codec = VersionTransformer.getEncoderDecoder(rmNamespace, rmAddressingNamespace);
- RMEndpoint rme = createReliableEndpoint(endpoint, codec);
+ RMEndpoint rme = createReliableEndpoint(endpoint, protocol);
rme.initialise(conduit, null, null);
- reliableEndpoints.put(endpoint, rme);
+ endpointMaps.get(protocol).put(endpoint, rme);
SourceSequence css = null;
for (SourceSequence ss : sss) {
@@ -530,8 +576,8 @@ public class RMManager implements Server
}
- RMEndpoint createReliableEndpoint(Endpoint endpoint, EncoderDecoder codec) {
- return new RMEndpoint(this, endpoint, codec);
+ RMEndpoint createReliableEndpoint(Endpoint endpoint, ProtocolVariation protocol) {
+ return new RMEndpoint(this, endpoint, protocol);
}
public void init(Bus b) {
@@ -590,15 +636,6 @@ public class RMManager implements Server
}
}
-
- Map<Endpoint, RMEndpoint> getReliableEndpointsMap() {
- return reliableEndpoints;
- }
-
- void setReliableEndpointsMap(Map<Endpoint, RMEndpoint> map) {
- reliableEndpoints = map;
- }
-
class DefaultSequenceIdentifierGenerator implements SequenceIdentifierGenerator {
public Identifier generateSequenceIdentifier() {
@@ -608,4 +645,15 @@ public class RMManager implements Server
return sid;
}
}
+
+ Map<ProtocolVariation, Map<Endpoint, RMEndpoint>> getEndpointMaps() {
+ return endpointMaps;
+ }
+
+ void setEndpointMaps(Map<ProtocolVariation, Map<Endpoint, RMEndpoint>> endpointMaps) {
+ endpointMaps.put(ProtocolVariation.RM10WSA200408, new HashMap<Endpoint, RMEndpoint>());
+ endpointMaps.put(ProtocolVariation.RM10WSA200508, new HashMap<Endpoint, RMEndpoint>());
+ endpointMaps.put(ProtocolVariation.RM11WSA200508, new HashMap<Endpoint, RMEndpoint>());
+ this.endpointMaps = endpointMaps;
+ }
}
\ No newline at end of file
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RetransmissionCallback.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RetransmissionCallback.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RetransmissionCallback.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RetransmissionCallback.java Mon Jul 25 05:56:11 2011
@@ -57,20 +57,24 @@ public class RetransmissionCallback impl
RMStore store = manager.getStore();
if (null != store) {
- Source s = manager.getSource(message);
- RMProperties rmps = RMContextUtils.retrieveRMProperties(message, true);
- Identifier sid = rmps.getSequence().getIdentifier();
- SourceSequence ss = s.getSequence(sid);
- RMMessage msg = new RMMessage();
- msg.setMessageNumber(rmps.getSequence().getMessageNumber());
- if (!MessageUtils.isRequestor(message)) {
- AddressingProperties maps = RMContextUtils.retrieveMAPs(message, false, true);
- if (null != maps && null != maps.getTo()) {
- msg.setTo(maps.getTo().getValue());
+ try {
+ Source s = manager.getSource(message);
+ RMProperties rmps = RMContextUtils.retrieveRMProperties(message, true);
+ Identifier sid = rmps.getSequence().getIdentifier();
+ SourceSequence ss = s.getSequence(sid);
+ RMMessage msg = new RMMessage();
+ msg.setMessageNumber(rmps.getSequence().getMessageNumber());
+ if (!MessageUtils.isRequestor(message)) {
+ AddressingProperties maps = RMContextUtils.retrieveMAPs(message, false, true);
+ if (null != maps && null != maps.getTo()) {
+ msg.setTo(maps.getTo().getValue());
+ }
}
- }
- msg.setContent(saved);
- store.persistOutgoing(ss, msg);
+ msg.setContent(saved);
+ store.persistOutgoing(ss, msg);
+ } catch (RMException e) {
+ // ignore
+ }
}
}
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java Mon Jul 25 05:56:11 2011
@@ -77,7 +77,7 @@ public class Servant implements Invoker
}
} else if (RM10Constants.INSTANCE.getCreateSequenceResponseOnewayOperationName().equals(oi.getName())
|| RM11Constants.INSTANCE.getCreateSequenceResponseOnewayOperationName().equals(oi.getName())) {
- EncoderDecoder codec = reliableEndpoint.getEncoderDecoder();
+ EncoderDecoder codec = reliableEndpoint.getProtocol().getCodec();
CreateSequenceResponseType createResponse =
codec.convertReceivedCreateSequenceResponse(getParameter(exchange.getInMessage()));
createSequenceResponse(createResponse);
@@ -99,7 +99,7 @@ public class Servant implements Invoker
RMContextUtils.storeMAPs(maps, outMessage, false, false);
}
- EncoderDecoder codec = reliableEndpoint.getEncoderDecoder();
+ EncoderDecoder codec = reliableEndpoint.getProtocol().getCodec();
CreateSequenceType create = codec.convertReceivedCreateSequence(getParameter(message));
Destination destination = reliableEndpoint.getDestination();
@@ -134,9 +134,8 @@ public class Servant implements Invoker
LOG.fine("Accepting inbound sequence offer");
// AddressingProperties maps = RMContextUtils.retrieveMAPs(message, false, false);
accept.setAcksTo(RMUtils.createReference(maps.getTo().getValue()));
- SourceSequence seq = new SourceSequence(offer.getIdentifier(),
- null,
- createResponse.getIdentifier());
+ SourceSequence seq = new SourceSequence(offer.getIdentifier(), null,
+ createResponse.getIdentifier(), reliableEndpoint.getProtocol());
seq.setExpires(offer.getExpires());
seq.setTarget(create.getAcksTo());
source.addSequence(seq);
@@ -155,7 +154,7 @@ public class Servant implements Invoker
}
DestinationSequence seq = new DestinationSequence(createResponse.getIdentifier(),
- create.getAcksTo(), destination);
+ create.getAcksTo(), destination, reliableEndpoint.getProtocol());
seq.setCorrelationID(maps.getMessageID().getValue());
destination.addSequence(seq);
LOG.fine("returning " + createResponse);
@@ -165,7 +164,8 @@ public class Servant implements Invoker
public void createSequenceResponse(CreateSequenceResponseType createResponse) {
LOG.fine("Creating sequence response");
- SourceSequence seq = new SourceSequence(createResponse.getIdentifier());
+ SourceSequence seq = new SourceSequence(createResponse.getIdentifier(),
+ reliableEndpoint.getProtocol());
seq.setExpires(createResponse.getExpires());
Source source = reliableEndpoint.getSource();
source.addSequence(seq);
@@ -186,8 +186,8 @@ public class Servant implements Invoker
Destination dest = reliableEndpoint.getDestination();
String address = accept.getAcksTo().getAddress().getValue();
if (!RMUtils.getAddressingConstants().getNoneURI().equals(address)) {
- DestinationSequence ds =
- new DestinationSequence(offeredId, accept.getAcksTo(), dest);
+ DestinationSequence ds = new DestinationSequence(offeredId, accept.getAcksTo(), dest,
+ reliableEndpoint.getProtocol());
dest.addSequence(ds);
}
}
@@ -196,7 +196,7 @@ public class Servant implements Invoker
public void terminateSequence(Message message) {
LOG.fine("Terminating sequence");
- EncoderDecoder codec = reliableEndpoint.getEncoderDecoder();
+ EncoderDecoder codec = reliableEndpoint.getProtocol().getCodec();
TerminateSequenceType terminate = codec.convertReceivedTerminateSequence(getParameter(message));
// check if the terminated sequence was created in response to a a createSequence
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Source.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Source.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Source.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Source.java Mon Jul 25 05:56:11 2011
@@ -26,12 +26,17 @@ import java.util.Map;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.helpers.CastUtils;
import org.apache.cxf.ws.rm.persistence.RMStore;
import org.apache.cxf.ws.rm.v200702.Identifier;
public class Source extends AbstractEndpoint {
+
+ private static final Logger LOG = LogUtils.getL7dLogger(Source.class);
private static final String REQUESTOR_SEQUENCE_ID = "";
@@ -63,13 +68,17 @@ public class Source extends AbstractEndp
}
public void addSequence(SourceSequence seq, boolean persist) {
- seq.setSource(this);
- map.put(seq.getIdentifier().getValue(), seq);
- if (persist) {
- RMStore store = getReliableEndpoint().getManager().getStore();
- if (null != store) {
- store.createSourceSequence(seq);
+ if (seq.getProtocol() == getReliableEndpoint().getProtocol()) {
+ seq.setSource(this);
+ map.put(seq.getIdentifier().getValue(), seq);
+ if (persist) {
+ RMStore store = getReliableEndpoint().getManager().getStore();
+ if (null != store) {
+ store.createSourceSequence(seq);
+ }
}
+ } else {
+ LOG.log(Level.SEVERE, "Incompatible protocol version");
}
}
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/SourceSequence.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/SourceSequence.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/SourceSequence.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/SourceSequence.java Mon Jul 25 05:56:11 2011
@@ -47,16 +47,16 @@ public class SourceSequence extends Abst
private Identifier offeringId;
private EndpointReferenceType target;
- public SourceSequence(Identifier i) {
- this(i, null, null);
+ public SourceSequence(Identifier i, ProtocolVariation pv) {
+ this(i, null, null, pv);
}
- public SourceSequence(Identifier i, Date e, Identifier oi) {
- this(i, e, oi, 0, false);
+ public SourceSequence(Identifier i, Date e, Identifier oi, ProtocolVariation pv) {
+ this(i, e, oi, 0, false, pv);
}
- public SourceSequence(Identifier i, Date e, Identifier oi, long cmn, boolean lm) {
- super(i);
+ public SourceSequence(Identifier i, Date e, Identifier oi, long cmn, boolean lm, ProtocolVariation pv) {
+ super(i, pv);
expires = e;
offeringId = oi;
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/VersionTransformer.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/VersionTransformer.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/VersionTransformer.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/VersionTransformer.java Mon Jul 25 05:56:11 2011
@@ -72,28 +72,6 @@ public final class VersionTransformer {
}
/**
- * Get the encoder/decoder for a supported version of WS-ReliableMessaging and WS-Addressing.
- *
- * @param wsrm WS-RM namespace URI
- * @param wsa WS-A namespace URI
- * @return instance (<code>null</code> if not a supported version)
- */
- public static EncoderDecoder getEncoderDecoder(String wsrm, String wsa) {
- if (EncoderDecoder10Impl.INSTANCE.getWSRMNamespace().equals(wsrm)
- && EncoderDecoder10Impl.INSTANCE.getWSANamespace().equals(wsa)) {
- return EncoderDecoder10Impl.INSTANCE;
- } else if (EncoderDecoder10AImpl.INSTANCE.getWSRMNamespace().equals(wsrm)
- && EncoderDecoder10AImpl.INSTANCE.getWSANamespace().equals(wsa)) {
- return EncoderDecoder10AImpl.INSTANCE;
- } else if (EncoderDecoder11Impl.INSTANCE.getWSRMNamespace().equals(wsrm)
- && EncoderDecoder11Impl.INSTANCE.getWSANamespace().equals(wsa)) {
- return EncoderDecoder11Impl.INSTANCE;
- } else {
- return null;
- }
- }
-
- /**
* Convert CreateSequenceType to 200502 version with 200508 WS-Addressing namespace.
*
* @param internal (non-<code>null</code>)
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMStore.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMStore.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMStore.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMStore.java Mon Jul 25 05:56:11 2011
@@ -22,6 +22,7 @@ package org.apache.cxf.ws.rm.persistence
import java.util.Collection;
import org.apache.cxf.ws.rm.DestinationSequence;
+import org.apache.cxf.ws.rm.ProtocolVariation;
import org.apache.cxf.ws.rm.SourceSequence;
import org.apache.cxf.ws.rm.v200702.Identifier;
@@ -41,20 +42,6 @@ public interface RMStore {
* @param seq the sequence
*/
void createDestinationSequence(DestinationSequence seq);
-
- /**
- * Retrieve the source sequence with the specified identifier from persistent store.
- * @param seq the sequence
- * @return the sequence if present; otherwise null
- */
- SourceSequence getSourceSequence(Identifier seq);
-
- /**
- * Retrieve the destination sequence with the specified identifier from persistent store.
- * @param seq the sequence
- * @return the sequence if present; otherwise null
- */
- DestinationSequence getDestinationSequence(Identifier seq);
/**
* Remove the source sequence with the specified identifier from persistent store.
@@ -75,7 +62,8 @@ public interface RMStore {
* @param endpointIdentifier the identifier for the source
* @return the collection of sequences
*/
- Collection<SourceSequence> getSourceSequences(String endpointIdentifier);
+ Collection<SourceSequence> getSourceSequences(String endpointIdentifier,
+ ProtocolVariation protocol);
/**
* Retrieves all sequences managed by the identified RM destination endpoint
@@ -84,7 +72,8 @@ public interface RMStore {
* @param endpointIdentifier the identifier for the destination
* @return the collection of sequences
*/
- Collection<DestinationSequence> getDestinationSequences(String endpointIdentifier);
+ Collection<DestinationSequence> getDestinationSequences(String endpointIdentifier,
+ ProtocolVariation protocol);
/**
* Retrieves the outbound/inbound messages stored for the source/destination sequence with
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java Mon Jul 25 05:56:11 2011
@@ -44,6 +44,7 @@ import org.apache.cxf.common.i18n.Messag
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.ws.addressing.EndpointReferenceType;
import org.apache.cxf.ws.rm.DestinationSequence;
+import org.apache.cxf.ws.rm.ProtocolVariation;
import org.apache.cxf.ws.rm.RMUtils;
import org.apache.cxf.ws.rm.SourceSequence;
import org.apache.cxf.ws.rm.persistence.PersistenceUtils;
@@ -57,14 +58,6 @@ public class RMTxStore implements RMStor
public static final String DEFAULT_DATABASE_NAME = "rmdb";
- private static final String CREATE_ENDPOINTS_TABLE_STMT =
- "CREATE TABLE CXF_ENDPOINTS "
- + "(QNAME_URI VARCHAR(1024), "
- + "QNAME_LOCAL VARCHAR(1024), "
- + "ENDPOINT_ID VARCHAR(1024), "
- + "WSRM_URI VARCHAR(256) NOT NULL, "
- + "WSA_URI VARCHAR(256) NOT NULL, "
- + "PRIMARY KEY (QNAME_URI, QNAME_LOCAL))";
private static final String CREATE_DEST_SEQUENCES_TABLE_STMT =
"CREATE TABLE CXF_RM_DEST_SEQUENCES "
+ "(SEQ_ID VARCHAR(256) NOT NULL, "
@@ -108,12 +101,6 @@ public class RMTxStore implements RMStor
= "INSERT INTO {0} VALUES(?, ?, ?, ?)";
private static final String DELETE_MESSAGE_STMT_STR =
"DELETE FROM {0} WHERE SEQ_ID = ? AND MSG_NO = ?";
- private static final String SELECT_DEST_SEQUENCE_STMT_STR =
- "SELECT ACKS_TO, LAST_MSG_NO, ACKNOWLEDGED FROM CXF_RM_DEST_SEQUENCES "
- + "WHERE SEQ_ID = ?";
- private static final String SELECT_SRC_SEQUENCE_STMT_STR =
- "SELECT CUR_MSG_NO, LAST_MSG, EXPIRY, OFFERING_SEQ_ID FROM CXF_RM_SRC_SEQUENCES "
- + "WHERE SEQ_ID = ?";
private static final String SELECT_DEST_SEQUENCES_STMT_STR =
"SELECT SEQ_ID, ACKS_TO, LAST_MSG_NO, ACKNOWLEDGED FROM CXF_RM_DEST_SEQUENCES "
+ "WHERE ENDPOINT_ID = ?";
@@ -139,8 +126,6 @@ public class RMTxStore implements RMStor
private PreparedStatement updateSrcSequenceStmt;
private PreparedStatement selectDestSequencesStmt;
private PreparedStatement selectSrcSequencesStmt;
- private PreparedStatement selectDestSequenceStmt;
- private PreparedStatement selectSrcSequenceStmt;
private PreparedStatement createInboundMessageStmt;
private PreparedStatement createOutboundMessageStmt;
private PreparedStatement deleteInboundMessageStmt;
@@ -252,68 +237,6 @@ public class RMTxStore implements RMStor
}
}
- public DestinationSequence getDestinationSequence(Identifier sid) {
- if (LOG.isLoggable(Level.FINE)) {
- LOG.info("Getting destination sequence for id: " + sid);
- }
- try {
- if (null == selectDestSequenceStmt) {
- selectDestSequenceStmt =
- connection.prepareStatement(SELECT_DEST_SEQUENCE_STMT_STR);
- }
- selectDestSequenceStmt.setString(1, sid.getValue());
-
- ResultSet res = selectDestSequenceStmt.executeQuery();
- if (res.next()) {
- EndpointReferenceType acksTo = RMUtils.createReference(res.getString(1));
- long lm = res.getLong(2);
- InputStream is = res.getBinaryStream(3);
- SequenceAcknowledgement ack = null;
- if (null != is) {
- ack = PersistenceUtils.getInstance()
- .deserialiseAcknowledgment(is);
- }
- return new DestinationSequence(sid, acksTo, lm, ack);
- }
- } catch (SQLException ex) {
- LOG.log(Level.WARNING, new Message("SELECT_DEST_SEQ_FAILED_MSG", LOG).toString(), ex);
- }
- return null;
- }
-
- public SourceSequence getSourceSequence(Identifier sid) {
- if (LOG.isLoggable(Level.FINE)) {
- LOG.info("Getting source sequences for id: " + sid);
- }
- try {
- if (null == selectSrcSequenceStmt) {
- selectSrcSequenceStmt =
- connection.prepareStatement(SELECT_SRC_SEQUENCE_STMT_STR);
- }
- selectSrcSequenceStmt.setString(1, sid.getValue());
- ResultSet res = selectSrcSequenceStmt.executeQuery();
-
- if (res.next()) {
- long cmn = res.getLong(1);
- boolean lm = res.getBoolean(2);
- long lval = res.getLong(3);
- Date expiry = 0 == lval ? null : new Date(lval);
- String oidValue = res.getString(4);
- Identifier oi = null;
- if (null != oidValue) {
- oi = new Identifier();
- oi.setValue(oidValue);
- }
- return new SourceSequence(sid, expiry, oi, cmn, lm);
-
- }
- } catch (SQLException ex) {
- // ignore
- LOG.log(Level.WARNING, new Message("SELECT_SRC_SEQ_FAILED_MSG", LOG).toString(), ex);
- }
- return null;
- }
-
public void removeDestinationSequence(Identifier sid) {
try {
beginTransaction();
@@ -351,7 +274,8 @@ public class RMTxStore implements RMStor
}
}
- public Collection<DestinationSequence> getDestinationSequences(String endpointIdentifier) {
+ public Collection<DestinationSequence> getDestinationSequences(String endpointIdentifier,
+ ProtocolVariation protocol) {
if (LOG.isLoggable(Level.FINE)) {
LOG.info("Getting destination sequences for endpoint: " + endpointIdentifier);
}
@@ -375,7 +299,7 @@ public class RMTxStore implements RMStor
ack = PersistenceUtils.getInstance()
.deserialiseAcknowledgment(is);
}
- DestinationSequence seq = new DestinationSequence(sid, acksTo, lm, ack);
+ DestinationSequence seq = new DestinationSequence(sid, acksTo, lm, ack, protocol);
seqs.add(seq);
}
} catch (SQLException ex) {
@@ -384,7 +308,8 @@ public class RMTxStore implements RMStor
return seqs;
}
- public Collection<SourceSequence> getSourceSequences(String endpointIdentifier) {
+ public Collection<SourceSequence> getSourceSequences(String endpointIdentifier,
+ ProtocolVariation protocol) {
if (LOG.isLoggable(Level.FINE)) {
LOG.info("Getting source sequences for endpoint: " + endpointIdentifier);
}
@@ -410,7 +335,7 @@ public class RMTxStore implements RMStor
oi = new Identifier();
oi.setValue(oidValue);
}
- SourceSequence seq = new SourceSequence(sid, expiry, oi, cmn, lm);
+ SourceSequence seq = new SourceSequence(sid, expiry, oi, cmn, lm, protocol);
seqs.add(seq);
}
} catch (SQLException ex) {
@@ -604,17 +529,6 @@ public class RMTxStore implements RMStor
Statement stmt = null;
stmt = connection.createStatement();
try {
- stmt.executeUpdate(CREATE_ENDPOINTS_TABLE_STMT);
- } catch (SQLException ex) {
- if (!isTableExistsError(ex)) {
- throw ex;
- } else {
- LOG.fine("Table CXF_ENDPOINTS already exists.");
- }
- }
- stmt.close();
- stmt = connection.createStatement();
- try {
stmt.executeUpdate(CREATE_SRC_SEQUENCES_TABLE_STMT);
} catch (SQLException ex) {
if (!isTableExistsError(ex)) {
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/Messages.properties
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/Messages.properties?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/Messages.properties (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/Messages.properties Mon Jul 25 05:56:11 2011
@@ -20,6 +20,7 @@ SOAP_HEADER_ENCODE_FAILURE_MSG = Failed
SOAP_HEADER_DECODE_FAILURE_MSG = Failed to decode RM properties from SOAP headers.
NAMESPACE_ERROR_MSG = Unsupported addressing namespace {0} for RM.
+CANNOT_PROCESS = WS-RM Message cannot be processed.
RESEND_MSG = WS-RM retransmission of message {0}.
RESEND_CANDIDATES_CONCURRENT_MODIFICATION_MSG = Candidates were acknowledged while iterating for resend.
RESEND_FAILED_MSG = WS-RM retransmission failed.
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInterceptor.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInterceptor.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInterceptor.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInterceptor.java Mon Jul 25 05:56:11 2011
@@ -65,11 +65,13 @@ import org.apache.cxf.ws.addressing.Attr
import org.apache.cxf.ws.addressing.soap.MAPCodec;
import org.apache.cxf.ws.rm.AbstractRMInterceptor;
import org.apache.cxf.ws.rm.EncoderDecoder;
+import org.apache.cxf.ws.rm.ProtocolVariation;
import org.apache.cxf.ws.rm.RM10Constants;
import org.apache.cxf.ws.rm.RM11Constants;
import org.apache.cxf.ws.rm.RMConstants;
import org.apache.cxf.ws.rm.RMContextUtils;
import org.apache.cxf.ws.rm.RMEndpoint;
+import org.apache.cxf.ws.rm.RMException;
import org.apache.cxf.ws.rm.RMManager;
import org.apache.cxf.ws.rm.RMMessageConstants;
import org.apache.cxf.ws.rm.RMProperties;
@@ -173,9 +175,9 @@ public class RMSoapInterceptor extends A
discardRMHeaders(headers);
AddressingProperties maps = RMContextUtils.retrieveMAPs(message, false, true);
- EncoderDecoder codec = VersionTransformer
- .getEncoderDecoder(rmps.getNamespaceURI(), maps.getNamespaceURI());
- Element header = codec.buildHeaders(rmps, Soap11.getInstance().getHeader());
+ ProtocolVariation protocol = ProtocolVariation.findVariant(rmps.getNamespaceURI(),
+ maps.getNamespaceURI());
+ Element header = protocol.getCodec().buildHeaders(rmps, Soap11.getInstance().getHeader());
Node node = header.getFirstChild();
if (node != null && MessageUtils.isPartialResponse(message)) {
// make sure the response is returned as HTTP 200 and not 202
@@ -206,9 +208,9 @@ public class RMSoapInterceptor extends A
Message inmsg = message.getExchange().getInMessage();
RMProperties rmps = RMContextUtils.retrieveRMProperties(inmsg, false);
AddressingProperties maps = RMContextUtils.retrieveMAPs(inmsg, false, false);
- EncoderDecoder codec = VersionTransformer
- .getEncoderDecoder(rmps.getNamespaceURI(), maps.getNamespaceURI());
- Element header = codec.buildHeaderFault(sf, Soap11.getInstance().getHeader());
+ ProtocolVariation protocol = ProtocolVariation.findVariant(rmps.getNamespaceURI(),
+ maps.getNamespaceURI());
+ Element header = protocol.getCodec().buildHeaderFault(sf, Soap11.getInstance().getHeader());
Node node = header.getFirstChild();
if (node instanceof Element) {
Attr attr = header.getOwnerDocument().createAttributeNS("http://www.w3.org/2000/xmlns/",
@@ -281,7 +283,8 @@ public class RMSoapInterceptor extends A
} else {
wsauri = maps.getNamespaceURI();
}
- codec = VersionTransformer.getEncoderDecoder(rmUri, wsauri);
+ ProtocolVariation protocol = ProtocolVariation.findVariant(rmUri, wsauri);
+ codec = protocol.getCodec();
if (codec == null) {
LOG.log(Level.WARNING, "NAMESPACE_ERROR_MSG", wsauri);
break;
@@ -339,7 +342,7 @@ public class RMSoapInterceptor extends A
*
* @param message the message
*/
- private void updateServiceModelInfo(SoapMessage message) {
+ private void updateServiceModelInfo(SoapMessage message) throws Fault {
AddressingProperties maps = RMContextUtils.retrieveMAPs(message, false, false);
AttributedURIType actionURI = null == maps ? null : maps.getAction();
@@ -360,7 +363,13 @@ public class RMSoapInterceptor extends A
RMManager manager = getManager(message);
assert manager != null;
- RMEndpoint rme = manager.getReliableEndpoint(message);
+ RMEndpoint rme = null;
+ try {
+ rme = manager.getReliableEndpoint(message);
+ } catch (RMException e) {
+ throw new SoapFault(new org.apache.cxf.common.i18n.Message("CANNOT_PROCESS", LOG), e,
+ message.getVersion().getSender());
+ }
Exchange exchange = message.getExchange();
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RetransmissionQueueImpl.java Mon Jul 25 05:56:11 2011
@@ -55,6 +55,7 @@ import org.apache.cxf.ws.addressing.Endp
import org.apache.cxf.ws.policy.AssertionInfo;
import org.apache.cxf.ws.policy.builder.jaxb.JaxbAssertion;
import org.apache.cxf.ws.rm.RMContextUtils;
+import org.apache.cxf.ws.rm.RMException;
import org.apache.cxf.ws.rm.RMManager;
import org.apache.cxf.ws.rm.RMMessageConstants;
import org.apache.cxf.ws.rm.RMProperties;
@@ -255,7 +256,7 @@ public class RetransmissionQueueImpl imp
resend(c, message);
}
- private void serverResend(Message message) {
+ private void serverResend(Message message) throws RMException {
// get the message's to address
Modified: cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/SoapFaultFactory.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/SoapFaultFactory.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/SoapFaultFactory.java (original)
+++ cxf/trunk/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/SoapFaultFactory.java Mon Jul 25 05:56:11 2011
@@ -35,11 +35,11 @@ import org.apache.cxf.message.Message;
import org.apache.cxf.ws.addressing.AddressingProperties;
import org.apache.cxf.ws.rm.BindingFaultFactory;
import org.apache.cxf.ws.rm.EncoderDecoder;
+import org.apache.cxf.ws.rm.ProtocolVariation;
import org.apache.cxf.ws.rm.RMConstants;
import org.apache.cxf.ws.rm.RMContextUtils;
import org.apache.cxf.ws.rm.RMProperties;
import org.apache.cxf.ws.rm.SequenceFault;
-import org.apache.cxf.ws.rm.VersionTransformer;
import org.apache.cxf.ws.rm.v200702.Identifier;
import org.apache.cxf.ws.rm.v200702.SequenceAcknowledgement;
@@ -85,8 +85,8 @@ public class SoapFaultFactory implements
try {
RMProperties rmps = RMContextUtils.retrieveRMProperties(msg, false);
AddressingProperties maps = RMContextUtils.retrieveMAPs(msg, false, false);
- EncoderDecoder codec = VersionTransformer
- .getEncoderDecoder(rmps.getNamespaceURI(), maps.getNamespaceURI());
+ EncoderDecoder codec = ProtocolVariation.findVariant(rmps.getNamespaceURI(),
+ maps.getNamespaceURI()).getCodec();
setDetail(fault, detail, codec);
} catch (Exception ex) {
LogUtils.log(LOG, Level.SEVERE, "MARSHAL_FAULT_DETAIL_EXC", ex);
Modified: cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java (original)
+++ cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java Mon Jul 25 05:56:11 2011
@@ -88,7 +88,8 @@ public class DestinationSequenceTest ext
Identifier otherId = factory.createIdentifier();
otherId.setValue("otherSeq");
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
assertEquals(id, seq.getIdentifier());
assertEquals(0, seq.getLastMessageNumber());
assertSame(ref, seq.getAcksTo());
@@ -96,7 +97,7 @@ public class DestinationSequenceTest ext
assertNotNull(seq.getMonitor());
SequenceAcknowledgement ack = new SequenceAcknowledgement();
- seq = new DestinationSequence(id, ref, 10, ack);
+ seq = new DestinationSequence(id, ref, 10, ack, ProtocolVariation.RM10WSA200408);
assertEquals(id, seq.getIdentifier());
assertEquals(10, seq.getLastMessageNumber());
assertSame(ref, seq.getAcksTo());
@@ -108,15 +109,16 @@ public class DestinationSequenceTest ext
@Test
public void testEqualsAndHashCode() {
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
DestinationSequence otherSeq = null;
assertTrue(!seq.equals(otherSeq));
- otherSeq = new DestinationSequence(id, ref, destination);
+ otherSeq = new DestinationSequence(id, ref, destination, ProtocolVariation.RM10WSA200408);
assertEquals(seq, otherSeq);
assertEquals(seq.hashCode(), otherSeq.hashCode());
Identifier otherId = factory.createIdentifier();
otherId.setValue("otherSeq");
- otherSeq = new DestinationSequence(otherId, ref, destination);
+ otherSeq = new DestinationSequence(otherId, ref, destination, ProtocolVariation.RM10WSA200408);
assertTrue(!seq.equals(otherSeq));
assertTrue(seq.hashCode() != otherSeq.hashCode());
assertTrue(!seq.equals(this));
@@ -125,7 +127,8 @@ public class DestinationSequenceTest ext
@Test
public void testGetSetDestination() {
control.replay();
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
seq.setDestination(destination);
assertSame(destination, seq.getDestination());
}
@@ -137,7 +140,8 @@ public class DestinationSequenceTest ext
EasyMock.expect(destination.getName()).andReturn(name);
control.replay();
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
assertEquals("Unexpected endpoint identifier", name, seq.getEndpointIdentifier());
control.verify();
}
@@ -150,7 +154,8 @@ public class DestinationSequenceTest ext
Message message2 = setUpMessage("2");
control.replay();
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
List<AcknowledgementRange> ranges = seq.getAcknowledgment().getAcknowledgementRange();
assertEquals(0, ranges.size());
@@ -207,7 +212,8 @@ public class DestinationSequenceTest ext
control.replay();
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
List<AcknowledgementRange> ranges = seq.getAcknowledgment().getAcknowledgementRange();
for (int i = 0; i < messages.length; i++) {
seq.acknowledge(messages[i]);
@@ -238,7 +244,8 @@ public class DestinationSequenceTest ext
};
control.replay();
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
List<AcknowledgementRange> ranges = seq.getAcknowledgment().getAcknowledgementRange();
for (int i = 0; i < messages.length; i++) {
seq.acknowledge(messages[i]);
@@ -272,7 +279,8 @@ public class DestinationSequenceTest ext
};
control.replay();
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
List<AcknowledgementRange> ranges = seq.getAcknowledgment().getAcknowledgementRange();
for (int i = 0; i < messages.length; i++) {
seq.acknowledge(messages[i]);
@@ -290,7 +298,8 @@ public class DestinationSequenceTest ext
@Test
public void testMerge() {
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
List<AcknowledgementRange> ranges = seq.getAcknowledgment().getAcknowledgementRange();
AcknowledgementRange r;
for (int i = 0; i < 5; i++) {
@@ -342,7 +351,8 @@ public class DestinationSequenceTest ext
}
control.replay();
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
SequenceMonitor monitor = seq.getMonitor();
assertNotNull(monitor);
monitor.setMonitorInterval(500);
@@ -382,7 +392,8 @@ public class DestinationSequenceTest ext
Message message = setUpMessage("1");
control.replay();
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
assertTrue(!seq.sendAcknowledgement());
seq.acknowledge(message);
@@ -400,7 +411,8 @@ public class DestinationSequenceTest ext
RMEndpoint rme = control.createMock(RMEndpoint.class);
setUpDestination(timer, rme);
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
Proxy proxy = control.createMock(Proxy.class);
EasyMock.expect(rme.getProxy()).andReturn(proxy).anyTimes();
proxy.acknowledge(seq);
@@ -442,7 +454,8 @@ public class DestinationSequenceTest ext
@Test
public void testCorrelationID() {
setUpDestination();
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
String correlationID = "abdc1234";
assertNull("unexpected correlation ID", seq.getCorrelationID());
seq.setCorrelationID(correlationID);
@@ -465,7 +478,7 @@ public class DestinationSequenceTest ext
EasyMock.expect(da.isSetAtMostOnce()).andReturn(true);
control.replay();
- DestinationSequence ds = new DestinationSequence(id, ref, 0, ack);
+ DestinationSequence ds = new DestinationSequence(id, ref, 0, ack, ProtocolVariation.RM10WSA200408);
ds.setDestination(destination);
ds.applyDeliveryAssurance(mn, null);
control.verify();
@@ -511,7 +524,8 @@ public class DestinationSequenceTest ext
factory.createSequenceAcknowledgementAcknowledgementRange();
r.setUpper(new Long(messages.length));
ranges.add(r);
- final DestinationSequence ds = new DestinationSequence(id, ref, 0, ack);
+ final DestinationSequence ds = new DestinationSequence(id, ref, 0, ack,
+ ProtocolVariation.RM10WSA200408);
ds.setDestination(destination);
class Acknowledger extends Thread {
@@ -569,7 +583,8 @@ public class DestinationSequenceTest ext
EasyMock.expect(rme.getProxy()).andReturn(control.createMock(Proxy.class)).anyTimes();
setUpDestination(timer, rme);
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
destination.removeSequence(seq);
EasyMock.expectLastCall();
@@ -597,7 +612,8 @@ public class DestinationSequenceTest ext
@Test
public void testSequenceTermination() {
destination = control.createMock(Destination.class);
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
RMEndpoint rme = control.createMock(RMEndpoint.class);
EasyMock.expect(destination.getReliableEndpoint()).andReturn(rme);
DestinationSequence.SequenceTermination st = seq.new SequenceTermination();
@@ -619,7 +635,8 @@ public class DestinationSequenceTest ext
EasyMock.expect(destination.getManager()).andReturn(manager);
Timer t = new Timer();
EasyMock.expect(manager.getTimer()).andReturn(t);
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
RMEndpoint rme = control.createMock(RMEndpoint.class);
EasyMock.expect(destination.getReliableEndpoint()).andReturn(rme);
DestinationSequence.SequenceTermination st = seq.new SequenceTermination();
@@ -635,7 +652,8 @@ public class DestinationSequenceTest ext
@Test
public void testCanPiggybackAckOnPartialResponse() {
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
AttributedURIType uri = control.createMock(AttributedURIType.class);
EasyMock.expect(ref.getAddress()).andReturn(uri);
String addr = "http://localhost:9999/reponses";
@@ -654,7 +672,8 @@ public class DestinationSequenceTest ext
@Test
public void testPurgeAcknowledged() {
destination = control.createMock(Destination.class);
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
manager = control.createMock(RMManager.class);
EasyMock.expect(destination.getManager()).andReturn(manager);
RMStore store = control.createMock(RMStore.class);
@@ -674,7 +693,8 @@ public class DestinationSequenceTest ext
EasyMock.expect(destination.getManager()).andReturn(manager);
Timer t = new Timer();
EasyMock.expect(manager.getTimer()).andReturn(t);
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
control.replay();
seq.scheduleDeferredAcknowledgement(30000L);
seq.cancelDeferredAcknowledgments();
@@ -691,7 +711,8 @@ public class DestinationSequenceTest ext
EasyMock.expect(destination.getManager()).andReturn(manager);
Timer t = new Timer();
EasyMock.expect(manager.getTimer()).andReturn(t);
- DestinationSequence seq = new DestinationSequence(id, ref, destination);
+ DestinationSequence seq = new DestinationSequence(id, ref, destination,
+ ProtocolVariation.RM10WSA200408);
control.replay();
seq.scheduleSequenceTermination(30000L);
seq.cancelTermination();
Modified: cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationTest.java
URL: http://svn.apache.org/viewvc/cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationTest.java?rev=1150554&r1=1150553&r2=1150554&view=diff
==============================================================================
--- cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationTest.java (original)
+++ cxf/trunk/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationTest.java Mon Jul 25 05:56:11 2011
@@ -114,7 +114,7 @@ public class DestinationTest extends Ass
EasyMock.expect(rmps.getSequence()).andReturn(st);
Identifier id = control.createMock(Identifier.class);
EasyMock.expect(st.getIdentifier()).andReturn(id).times(2);
- EasyMock.expect(rme.getEncoderDecoder()).andReturn(EncoderDecoder10Impl.INSTANCE);
+ EasyMock.expect(rme.getProtocol()).andReturn(ProtocolVariation.RM10WSA200408).anyTimes();
String sid = "sid";
EasyMock.expect(id.getValue()).andReturn(sid);
control.replay();