Repository: cxf Updated Branches: refs/heads/2.7.x-fixes ae313a2ea -> a30956d3a
[CXF-6519] add CloseSequence handling in 2.7.x Project: http://git-wip-us.apache.org/repos/asf/cxf/repo Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/a30956d3 Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/a30956d3 Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/a30956d3 Branch: refs/heads/2.7.x-fixes Commit: a30956d3a5602021fab9567b33ab7dffd32dbac0 Parents: ae313a2 Author: Akitoshi Yoshida <[email protected]> Authored: Thu Aug 6 23:53:36 2015 +0200 Committer: Akitoshi Yoshida <[email protected]> Committed: Fri Aug 7 11:55:11 2015 +0200 ---------------------------------------------------------------------- rt/ws/rm/.gitignore | 1 + .../org/apache/cxf/ws/rm/RM11Constants.java | 12 +++++++ .../java/org/apache/cxf/ws/rm/RMEndpoint.java | 26 +++++++++++++- .../main/java/org/apache/cxf/ws/rm/Servant.java | 38 ++++++++++++++++++++ 4 files changed, 76 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cxf/blob/a30956d3/rt/ws/rm/.gitignore ---------------------------------------------------------------------- diff --git a/rt/ws/rm/.gitignore b/rt/ws/rm/.gitignore new file mode 100644 index 0000000..b83d222 --- /dev/null +++ b/rt/ws/rm/.gitignore @@ -0,0 +1 @@ +/target/ http://git-wip-us.apache.org/repos/asf/cxf/blob/a30956d3/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM11Constants.java ---------------------------------------------------------------------- diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM11Constants.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM11Constants.java index b317a57..c24daad 100644 --- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM11Constants.java +++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RM11Constants.java @@ -74,6 +74,8 @@ public final class RM11Constants extends RMConstants { new QName(NAMESPACE_URI, "SequenceAcknowledgement"); public static final QName CLOSE_SEQUENCE_QNAME = new QName(NAMESPACE_URI, "CloseSequence"); + + public static final QName CLOSE_SEQUENCE_RESPONSE_QNAME = new QName(NAMESPACE_URI, "CloseSequenceResponse"); public static final QName ACK_REQ_QNAME = new QName(NAMESPACE_URI, "AckRequested"); @@ -191,6 +193,11 @@ public final class RM11Constants extends RMConstants { public String getCloseSequenceAction() { return CLOSE_SEQUENCE_ACTION; } + + // only defined for WS-RM 1.1/1.2 + public String getCloseSequenceResponseAction() { + return CLOSE_SEQUENCE_RESPONSE_ACTION; + } public String getAckRequestedAction() { return ACK_REQUESTED_ACTION; @@ -242,6 +249,11 @@ public final class RM11Constants extends RMConstants { return CLOSE_SEQUENCE_QNAME; } + // not part of the interface, only in WS-RM 1.1/1.2 + public QName getCloseSequenceResponseOperationName() { + return CLOSE_SEQUENCE_RESPONSE_QNAME; + } + public QName getSequenceAckOperationName() { return SEQUENCE_ACKNOWLEDGEMENT_QNAME; } http://git-wip-us.apache.org/repos/asf/cxf/blob/a30956d3/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java ---------------------------------------------------------------------- diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java index 9d9a266..1f5a578 100644 --- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java +++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMEndpoint.java @@ -69,6 +69,8 @@ import org.apache.cxf.ws.policy.PolicyEngine; import org.apache.cxf.ws.policy.PolicyInterceptorProviderRegistry; import org.apache.cxf.ws.rm.manager.SequenceTerminationPolicyType; import org.apache.cxf.ws.rm.manager.SourcePolicyType; +import org.apache.cxf.ws.rm.v200702.CloseSequenceResponseType; +import org.apache.cxf.ws.rm.v200702.CloseSequenceType; import org.apache.neethi.Assertion; import org.apache.neethi.Policy; @@ -84,6 +86,8 @@ public class RMEndpoint { private static final String CREATE_RESPONSE_PART_NAME = "createResponse"; private static final String TERMINATE_PART_NAME = "terminate"; private static final String TERMINATE_RESPONSE_PART_NAME = "terminateResponse"; + private static final String CLOSE_PART_NAME = "close"; + private static final String CLOSE_RESPONSE_PART_NAME = "closeResponse"; private static Schema rmSchema; @@ -543,6 +547,21 @@ public class RMEndpoint { messageInfo = operationInfo.createMessage(consts.getCloseSequenceOperationName(), MessageInfo.Type.INPUT); operationInfo.setInput(messageInfo.getName().getLocalPart(), messageInfo); + if (RM11Constants.NAMESPACE_URI.equals(protocol.getWSRMNamespace())) { + MessagePartInfo partInfo = messageInfo.addMessagePart(CLOSE_PART_NAME); + partInfo.setElementQName(consts.getCloseSequenceOperationName()); + partInfo.setElement(true); + partInfo.setTypeClass(CloseSequenceType.class); + messageInfo = operationInfo.createMessage( + RM11Constants.INSTANCE.getCloseSequenceResponseOperationName(), + MessageInfo.Type.OUTPUT); + operationInfo.setOutput(messageInfo.getName().getLocalPart(), messageInfo); + partInfo = messageInfo.addMessagePart(CLOSE_RESPONSE_PART_NAME); + partInfo.setElementQName(RM11Constants.INSTANCE.getCloseSequenceResponseOperationName()); + partInfo.setElement(true); + partInfo.setTypeClass(CloseSequenceResponseType.class); + partInfo.setIndex(0); + } } void buildAckRequestedOperationInfo(InterfaceInfo ii, ProtocolVariation protocol) { @@ -599,7 +618,12 @@ public class RMEndpoint { bi.addOperation(boi); boi = bi.buildOperation(consts.getCloseSequenceOperationName(), null, null); - addAction(boi, consts.getCloseSequenceAction()); + if (RM11Constants.NAMESPACE_URI.equals(protocol.getWSRMNamespace())) { + addAction(boi, consts.getCloseSequenceAction(), + RM11Constants.INSTANCE.getCloseSequenceResponseAction()); + } else { + addAction(boi, consts.getCloseSequenceAction()); + } bi.addOperation(boi); boi = bi.buildOperation(consts.getAckRequestedOperationName(), null, null); http://git-wip-us.apache.org/repos/asf/cxf/blob/a30956d3/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java ---------------------------------------------------------------------- diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java index 43bea3d..319e320 100644 --- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java +++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Servant.java @@ -39,6 +39,8 @@ import org.apache.cxf.ws.addressing.AddressingProperties; import org.apache.cxf.ws.addressing.ContextUtils; import org.apache.cxf.ws.rm.manager.DestinationPolicyType; import org.apache.cxf.ws.rm.v200702.AcceptType; +import org.apache.cxf.ws.rm.v200702.CloseSequenceResponseType; +import org.apache.cxf.ws.rm.v200702.CloseSequenceType; import org.apache.cxf.ws.rm.v200702.CreateSequenceResponseType; import org.apache.cxf.ws.rm.v200702.CreateSequenceType; import org.apache.cxf.ws.rm.v200702.Expires; @@ -106,6 +108,8 @@ public class Servant implements Invoker { if (tsr != null) { return Collections.singletonList(tsr); } + } else if (RM11Constants.INSTANCE.getCloseSequenceOperationName().equals(oi.getName())) { + return Collections.singletonList(closeSequence(exchange.getInMessage())); } return null; @@ -290,6 +294,40 @@ public class Servant implements Invoker { return terminateResponse; } + public Object closeSequence(Message message) { + LOG.fine("Closing sequence"); + + CloseSequenceType close = (CloseSequenceType)getParameter(message); + + // check if the terminated sequence was created in response to a a createSequence + // request + + Destination destination = reliableEndpoint.getDestination(); + Identifier sid = close.getIdentifier(); + DestinationSequence closedSeq = destination.getSequence(sid); + if (null == closedSeq) { + // TODO + LOG.severe("No such sequence."); + return null; + } + closedSeq.scheduleImmediateAcknowledgement(); + closedSeq.setLastMessageNumber(close.getLastMsgNumber()); + CloseSequenceResponseType closeResponse = new CloseSequenceResponseType(); + closeResponse.setIdentifier(close.getIdentifier()); + AddressingProperties maps = RMContextUtils.retrieveMAPs(message, false, false); + Message outMessage = message.getExchange().getOutMessage(); + + if (null == outMessage) { + // outMessage may be null e.g. if ReplyTo is not set for TS + outMessage = ContextUtils.createMessage(message.getExchange()); + message.getExchange().setOutMessage(outMessage); + } + if (null != outMessage) { + RMContextUtils.storeMAPs(maps, outMessage, false, false); + } + return closeResponse; + } + Object getParameter(Message message) { List<?> resList = null; // assert message == message.getExchange().getInMessage();
