You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cxf.apache.org by dk...@apache.org on 2016/07/22 17:06:06 UTC
[1/9] cxf git commit: FIx compile issue
Repository: cxf
Updated Branches:
refs/heads/3.1.x-fixes 661c0bd93 -> 2e62f6e30
FIx compile issue
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/347a54ca
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/347a54ca
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/347a54ca
Branch: refs/heads/3.1.x-fixes
Commit: 347a54ca943ce3de9750e2a623fa92a41e7ef526
Parents: 573fffc
Author: Daniel Kulp <dk...@apache.org>
Authored: Fri Jul 22 12:13:03 2016 -0400
Committer: Daniel Kulp <dk...@apache.org>
Committed: Fri Jul 22 12:13:14 2016 -0400
----------------------------------------------------------------------
.../src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/347a54ca/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java
index 87ae210..bc7e8ea 100755
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java
@@ -108,7 +108,7 @@ public class RMCaptureInInterceptor extends AbstractRMInterceptor<Message> {
return false;
}
- private void setCloseable(Message message, CachedOutputStream cos, InputStream is) {
+ private void setCloseable(final Message message, final CachedOutputStream cos, final InputStream is) {
message.put("org.apache.cxf.ws.rm.content.closeable", new Closeable() {
@Override
public void close() throws IOException {
[3/9] cxf git commit: Adding public key
Posted by dk...@apache.org.
Adding public key
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/7a365ca6
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/7a365ca6
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/7a365ca6
Branch: refs/heads/3.1.x-fixes
Commit: 7a365ca6064803abfff7782a0f972dc8b4e29687
Parents: 0dd2950
Author: Christian Schneider <ch...@die-schneider.net>
Authored: Tue Jul 5 13:39:23 2016 +0200
Committer: Daniel Kulp <dk...@apache.org>
Committed: Fri Jul 22 12:13:14 2016 -0400
----------------------------------------------------------------------
KEYS | 371 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 371 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/7a365ca6/KEYS
----------------------------------------------------------------------
diff --git a/KEYS b/KEYS
index 031b7af..dbb9672 100644
--- a/KEYS
+++ b/KEYS
@@ -651,3 +651,374 @@ hSmC1VrA3Xo6NklFA2DKKHNlOcw6tjgkriv99prqvPjzihylO8L/fnoYfgjAHs6v
jhtqQ2HLF8bidnMgrMeWFA==
=ELYH
-----END PGP PUBLIC KEY BLOCK-----
+pub 2048R/6428BDBA 2012-11-05
+uid Christian Schneider <ch...@die-schneider.net>
+sig 3 6428BDBA 2012-11-05 Christian Schneider <ch...@die-schneider.net>
+sig 3 88817402 2012-11-08 Thomas Vandahl <th...@vandahl.org>
+sig EEED47DA 2012-11-07 Ioan Eugen Stan <eu...@ieugen.ro>
+sig F5CFD915 2012-11-07 Jakob Frank <ja...@apache.org>
+sig F5CBAE6B 2012-11-07 Akitoshi Yoshida (CODE SIGNING KEY) <ay...@apache.org>
+sig 3E3CB8C9 2012-11-09 Herbert D�rr (CODE SIGNING KEY) <hd...@apache.org>
+sig 02E9F65B 2012-11-07 Luc Maisonobe (CODE SIGNING KEY) <lu...@apache.org>
+sig 51B5FDE8 2012-11-09 Juergen Schmidt <js...@apache.org>
+sig 1 A097C9AE 2012-11-13 Igor Gali\u0107 <i....@brainsware.org>
+sig 65422A3D 2012-11-10 Juan Pablo Santos Rodr�guez <ju...@gmail.com>
+sig 30B94B5C 2012-11-12 \u694a\u58eb\u9752 (Yang Shih-Ching) <im...@mail.imacat.idv.tw>
+sig 1 16A0DE01 2012-11-12 Ben Reser <be...@reser.org>
+sig E70D2357 2012-12-15 David Nalley <da...@gnsa.us>
+sig 2 47E4610C 2012-11-09 Martin Goldhahn <mg...@gmail.com>
+sig 0BD936F1 2012-11-13 Ivan Habunek (Apache Codesigning) <ih...@apache.org>
+sig 6FE50F1C 2012-12-15 David Nalley (Code signing key) <ke...@apache.org>
+sig A15F30B9 2012-12-15 David Nalley <da...@gnsa.us>
+sig E5C532A6 2013-01-02 Fabian Christ (CODE SIGNING KEY) <fc...@apache.org>
+sig B6CD3280 2014-09-14 Udo Schnurpfeil <lo...@apache.org>
+sig 3 22D7F6EC 2015-10-01 Hendrik Saly <we...@saly.de>
+sig 3 90910A83 2015-10-01 Hendrik Saly <we...@saly.de>
+sig 3D0C92B9 2015-10-01 Owen O'Malley (Code signing) <om...@apache.org>
+uid [jpeg image of size 5725]
+sig 3 6428BDBA 2013-06-27 Christian Schneider <ch...@die-schneider.net>
+sig 3D0C92B9 2015-10-01 Owen O'Malley (Code signing) <om...@apache.org>
+sub 2048R/54A04D55 2012-11-05
+sig 6428BDBA 2012-11-05 Christian Schneider <ch...@die-schneider.net>
+
+-----BEGIN PGP PUBLIC KEY BLOCK-----
+Version: GnuPG v1
+
+mQENBFCYM3MBCADWne/6CKEYXvNFT8ZO63Ij+8T85o+WIRURClNx3vth5JKzHQok
+8xMNkSIYLhYYjAOMPLsCShi+arVvswHzOIqoOBdjD4cJZzTN1JnM/VEVdVui14fe
+12xfundD9ElnQhOldTXjyjSyYgElJQJY8uT/nmme8EFaDLk3yP0wi32urYwgrqvN
+5mTmHXiI8JBXKGXPRCAR1PTfVk5hMq6EgMu7TPErSpbhRfU9jVBf91AZS8bmQg4X
+9q9jW84NJ4/b55kPTUFtIHLXOmAPtm/XpEAO8NHbq5vPHq7HG5BszcrGB+QT2kve
+LLSfizxOxgU2fs0ehiq3nagbQHJABGLMSdsbABEBAAG0LUNocmlzdGlhbiBTY2hu
+ZWlkZXIgPGNocmlzQGRpZS1zY2huZWlkZXIubmV0PokBOAQTAQIAIgUCUJgzcwIb
+AwYLCQgHAwIGFQgCCQoLBBYCAwECHgECF4AACgkQRxXcAmQovborkgf/e5/uu0Fa
+U+gDJgfRnLdtxHh6FecIS4s61nzCms1OkchQZdBTI4mEbb3m6gkZnCHqzVO+irY8
+1BzV0Gk0ElEdnsAeMwfmj1Vt29jT+IyzDjGWjmjoXdV9M8v5/ydCSGTyfJAmLw1T
+apH5veIdnsF/1RNU05erSkXaMWdkbfToTZwH/mgi2d1ml48ItQeQWJG3xaMmUBvS
+qcaz7MCXlnBWQt2fAYPIcNCjSP5nZ7GlOXitG1HRZPcHSEupeTi0iVjVEKn0HIqR
+8Qq5BnbEGmfTUqEQkHNr7aJCE70XBp8C2Evyk+bTc2r5uki6V5w1k8xW4HYHw9wM
+HXpa1y4YYkprVIhGBBMRAgAGBQJQm/KHAAoJEC65RoKIgXQCY+YAnRftd5zG0KWK
+6nVUN9J7JnMIYYklAKD3n4hvCVvf3NDWwT+RFGiFG1lrs4kBHAQQAQIABgUCUJom
+HgAKCRDtCW2T7u1H2tO8CADR7gGcWp1Zh+8zRQSxNSzYv3Sni5TH0wtjDlDdxNPr
+u8402xuJ9X1b659DAHoLYaaTqLc6WdTBENP1ftKJP/f42QAqwnAMb47dbyouNJk3
+E759zmnPX+iabBUrHsf6AfsYmceYGhJfKq5lNLBvis1ApNPpgU1kQ+6Vw/9WXKCy
+cjIqkI2fCPPu5Xm4x7TFPXpUZ67aVQ9dTJdpLeANlIkoXA2PGxfXX/8KnoRtlUNA
+SrUay1CCq6Roozx1GK5ETwfJpxXvA4yxqYC69h2oh49ZlXwqOePawpIjnXT2C3jd
+TCeo6MkPRI3/esxwbk5Ad6VZ+H+XBJ9OJiSTVx29Y9gaiQIcBBABAgAGBQJQmj8f
+AAoJEDYlkKf1z9kVTfMP/2YA79+2sqlTUoGfiWxbWDumXY+VPhV8vrNAE6tXwbpq
+grCQUOTVpq7CaJbKL6FxZAHIMxFYQdPNCSv/5O/e81oUZQg33Y4KvZG6nWdGVDhE
+feKRfrkBAXtvED28qumOjAL+9zBGHB5gOEaHTYRMsejcCmvCLhpRL85k02Y8kLyP
+o6zqmZOxNPaJWRrKF7orODGIVNPeiFOT56bklew762R68WYnO3Kpi/MCefvZjpIZ
+JbkW54859lth9CuVV3czzuG5xVhw+5V7chYGf9qSR9SWPqC/khZzvjSRpl7yV+Kj
+LeQflzfUL2HdaDbjTLLZeX9jl2e95OKDzzrzO5eUIszb/YjkYo1+9z9jOxZZvFiD
+ziCpAysI24fBt0Fn4keOrOLjwSbpufbNQ7r/HLjlOJyOfxQOf7uxVBgqFZ2UXLgC
+a5ForgQ23qA0AAGHbUkftsv4hiiduPp81bQvCb6ldoZIkhMLmpHnpfAv1YTDchC8
+dFy4qfznZpOiibSU+J1/Im1bGrEHjVnX6MxulWXlv6mB/82GkHoZkv4AgEPa0nNa
+scSqEo128kGBj+G8Kbt4JID+xJaaZwa/LXLAIIMXsn1+db6RLOq+kjRe6gyP+XnA
+x186FzLzEAEJ3T0IHuHdZ1dj99VM0uH82pUh8UuneYJlE47ZF0MddP15hj/vFX5l
+iQIcBBABAgAGBQJQmqQ9AAoJEPtjPej1y65rhB8P+wQtmn9kTYL+6lia4MFW0ORj
+blIA1L409p8bujkch9uhtGpxm0V8eR8NHl1nhzmQm/+R2IqnZHl1kcWMOsYMiK6b
+d8naxRh1CwPEIZlU8napQAuu1TH8C1LFzUrNeInOJ8ZFVAfNLFf+vsjO9ctL2tzX
+4jX4heY6WUSpx7TJazSX8C4IyCbkxl0UALQHAl2I4Ci+a6lpbWljNPc4JLAwKl1k
+VbEP32U4+oJcV7PdsxqAjXOGNKcLbwts+Q39IGuhAtne8Ce1zKE+KqtlZTfG0gzA
+2jTAzStyYj5rXZ2ExveUkecKXIZJXyQQj5K+TNCBusMGGkVeVrhQPINxcAV0cVb6
+yQx33uwOd5jhZDA/LULw4KLu8YmEdnTK3u0ARW5ngh98Xgtj2hVyJoy2rsrFBQr+
+bZVlmyin4N5pppXqVDLwemzJ4UyWof/pd1YMPgr1R1NoW9BLSyogS8ULYdwy0oZy
+zmTDpklIx+S2sMc6Lfla50/FGo5aSxiMY0Gy4jxHY19ZvWECCOD2955EadvJLCrQ
+Z/z4sZC5uxZ29pLNOr+QEfqxgAVduAcYOyOTokL3Sc+gK6i01JX/gGaqtPAi0dQp
+c8tgYtZtHiAofJSnmMThU5MAOoNCWmJqrrnApIGMon7iTAqM1O8orxjGOTkK2OpO
+a69Ha5CPz9nwIrnYnybmiQIcBBABAgAGBQJQnPRiAAoJEDfnuKc+PLjJ73MQAKGd
++MjbcGYcnMjILDaWnUmz/81eIoXhCh7GhVQoecvsREGPUSJhKMEyLotnHOX7Hnrb
+s/J7eloStih7J6mliadjEGBpehsiOBzwG9o5rEz5lGVe/mIsiWM9ul7jBpPMRNnU
++MQvMuz1JQyS18jUgdjjqQaEUeQC6MESQt2NOv15K8VITP1NUu8xEr5LjcA/HIZv
+qiqrZIt6HqlQ4dsnbuLfYdOCpiyCXvJcUAJ0STIbNWNx57uRZoMVBYnW3vTL3T/+
+Q/8E8+04WGKhe0M5xvPsWMx4fw4cqKqYRE8NrWXkjWgnxFtbMy52nFYUGltBNxBO
+WKiqE654+rMsIZWVulVGx/uqW/LyKQ7axt329179UxQyRvSRYPpl3gHhXdgGwQKf
+OO9pyvIgRpTw0P8WqmuvcIHl0P80ygHeI4hjDcKXFo2wqqn0vhtaT2yyGCV1hWAQ
+sUK0jaFwhJ4X+ifFtbYR8LQe7LRdX0KHUkNWqtkpl3ySuBGZwuvL4HTJcgnDL0cQ
+vblTdRR52ORPGj6MlH6nHUyXPt389UAEa7sSiFXOZIPu2CY7lljSNnwzfk7GgTcH
+9dje4BFirGsMiVL9hC7efuUMQNzKGSe+TE5Z3JrxHOy6sZumVLG1+kHlt9UZtuyO
+F0dZ4q5wtfRh6T9KkYcNAWSJF53vLYhTbu7rBMZTiQIcBBABCgAGBQJQmp+BAAoJ
+EJrilv0C6fZbMpoP/ihZ1/biPhmIlHe2mLN0Zp4o5ATP3iKwZgriVSTy9hZF9Iat
+pSz0TCMG2dBI5yQ1cedPrNSwOcR0B3uczUtOf6Igt1TW36eMrnpAdZ0CtBkRxF9D
+wljGWR5QzfD6Biy1epB+K/SG/YhInVfEby5rZzRiJRDw1E3J4UBp95QTbILi57g0
+b1zNC5Ff8NAErPmjLx1TVFWZ5+LQNvEeeW7NcIZYBhvfr5XgP1UhWsKj/W6RSvqg
+psDmQlt46k2YwOu4km91mp4A2hoKHeIHyLvnLGir1L/JHCr+lF90G6ln3rBKRW9z
+7e6hz7I7ixFkRDgf0XtktXevl/vFuEoGjqKiAkEjKNonw0c+u33XZBtHcS5IvEjd
+P8Ud3Oh6Xy27dr90oiGLiN7FhriEMT3cTyysG7NiZSdez0tuKI30AlJXug0UOCv/
+t1rI7qtuknrF2FZRyj1OWSVrLPPSJSel1m4Y5T8LWnnRIuoyVql9vIOhw86Y1Nmk
+4K+6VPmOLVgNwDPfPzbAunvGSB8/bnziorCju6LEliT63Ta2mrwpD9t4Lw0AKPWx
+j/nHw4TD3onCDu0QreLXrf356AvoC3+9t1YRsJC4RDk3A71Tox/bXuA0DVklJDvC
+81tcKV4upACGZDaeUhrHEYO4iXQCog1w6JZlemCRghMnHYEcSErstRrQ8mWPiQIc
+BBABCgAGBQJQnN0gAAoJEM/u8xZRtf3oD2AQAIoLHtIfOJLiSBpLOdtpL2hvClAT
+F0AFO1d4krLCzbDikqF3VyutESG2r8HohXMSPYMlT8j6bboxPnptXSYyGUwjMaar
+RbeZwWMsnKF+1H6jw1T4OtLWBvQnWw4G+9q8T66CmLp+ahQ7mygAQIBf7u64BWgr
+Ls1kVtdwcPW4Rr5VjjNXu5kOvs5FWj0kc5MUYbrDjzpCGg8j84SdGiNDnS7g/nN+
+A2ThI9fwZP1kJtz7MxHe2XYJr/RnAa/lGOXcPWuH1QrTGueEwosSfvMsgpM1UJ3f
+46cZecmyGZRQ3YpzaDVrDEmAOThix3R9lGt5I0vgO5PrvXnfiwNUpvFTpuUD8b67
+dTq+p+wVqKSjETE5XktJJAG6eiisHSWGQMuV55TmFBQOr9Cr2U3aGaOY9/6EbYCY
+odcavcajaMUm1ZvXMAHc4A8WbDyvXhJPJqdunT4H/dgDDSSVRR4c62wzRo0FvquG
+l7CgKln8XG4PIwEjnDmnteob5eyvX+zxa1oob4fOD5ZJMKYfTFHceubO2inD2eHl
+VLhnH7pS99kFtph8G7Oa8F6PJbENAFXNBrg7twURq0yAmiMXQJfluj3uXHhzZYbL
+GVO8Z6eenVRHmbQAecO1wvjjINDJxxo18ES2l6CQizu4Itqw3fPFYoZmyrQIB9+4
+5kCTc9Lg4mZJ4xkpiQIcBBEBAgAGBQJQokWmAAoJEEsdngigl8mutScQAJtZJDLI
+Caakbyb2VlT42XPVSKmMejM66wW6U55CqwbGr3CgSzMSqSjPPXzNDT6kuWhsGyMJ
+pJmFuXXlVhIp1HV6LYlzmlx0TzcxZWDlsrBFAbdVE+fwHOWa21Ma8SBi19TEMRUH
+zBG7DmMxv2k14+Q9Vx2w6sKw112QRrs9E9OS91YYr4K3X1sLWmy3+nXEWKyt1eu+
+dbq8ieJJ7MNdK8wfk6HNF74p+0v5PXw9vAwbyVjML6wV/gsvPpzeTLjFjwUpKlFI
+RXQQ6FwcdWABj22dYDff5ePlwpxntqDHXPvT4aizQOqy6wTxlmGkLv469IWo3wET
+5Xyu/Nb7VtxpyZA5U4sSLFt9YGJ2GhqdgRNVZE2nod3g8eZoLfCi8BMKc7lXd4av
+ZL21PFwIIgdAGfeK7X5b7jD0WZ3t51KnrgSytNpe+YAE5GWHpmUILdC/nnY2D0eX
+mjRYK0MM3aFR7ymZyyKrRJzI3tZsuq+NumvZMFaFyGtiKHFNYBaXEo6JYG7MBgpB
+aeE1RHFBbiOuGZRRkxP5XBEYLL2/l9D2lxXm1Nj4HJW/4wz3jSmJM5ZzImO9kp2t
+BikkQ3iVE92YUOXk+vZ+yhjQYfH0qD3x5f7srChKpgv/swFgh1SZtuv2puf3DsCf
+sy5EC98Qo+1Wps6SXbl0aL5lwDBZs00d6NN2iEYEEBECAAYFAlCeyTgACgkQ5hZn
+KWVCKj0wUACgnBfBjl7ioomh48nSxEdTXv7H+H0An3tiLD2APrkeN310xiodY+W0
+YPrXiEYEEBECAAYFAlChQuAACgkQi9gubzC5S1w5tACfbvSCBgQHQMK6iJH6K+CC
+LzlsR48AoJnMaxtOeL2RlDX+GhcDR0/jhuOfiQIgBBABCgAKBQJQoUBCAwUBPAAK
+CRBi1I+tFqDeAfTGD/0UdZDZUDOKzGNiuopgmr8XJDfAychutWDC+uN+Phc43/Ol
+e+Tii8oTxBFJSDXQr04oa/iEAjEQQcKB2lrc/ajTZFoqsrtfhwT3ncEoT5WXMCat
+AaS3QmNy5e3gG0a1tjH0GB0o9CL9RVUMOBnbcUq4T3GLlUG0uXZ+AGn6BnWjDgjZ
+AYjsr4sPTte4LC62HgJOY/tCshWzNEvBWLYyuHvUygQwRacjelGo9jFHbspstDh/
+ULV1NoHEy9yfzAeIqsE5pvrtatu0LX90e3XgN5HakaSCrUOJGrrK2VbOjZc74Mk5
+S+ukauDAAJeKKDlvC42BPbkhU6EQr5WAH6CTb3R3A5HqVRhOo5Wb3tSKkebL+mzx
+0weOx42Fl6tqHm+Y/xO6ubvetrB/GXvK609AXr83e4csR8rDTGbCslIZsuqze4+V
+BNpQzAgNwH6BYAjioBp15TWr0ZyBnRfQjJ9hy32HbbNTqy0l4/ul3Uibh0RN7Sto
+fcZK5lGvlm1Ln69p+nqo/IQ9OokOvkYgaRPmhgKvEomWVpM7MbZbZeSgqwxFdhQd
+uIkav9PjqqiuTnjRlHp/VrlODzHKKfT74nU+8oA/RHZPkyR3k2/FF1PpE1KA7bRB
+s+P8h1kRB2xDUaMZ1Q1EziLS4wB0STiVgnQaDgJC0TRvCS254hjjHizBrbqMsYhG
+BBARAgAGBQJQzN4wAAoJEJGTmI/nDSNX4z4AoIU5YeE/J57p4dklBSs76lae1Ge6
+AJ0YQsUAiMo3OGoEpMKW4CdwRrBMWokBIgQSAQIADAUCUJ2RzAWDB4YfgAAKCRBN
+jMRwR+RhDP/KB/9C8hPM/QtV2uC/vWEXuIXOvCAKzJLOKfbXmKuiQY9OJNXLz00p
+wtG+qDrdyra9hZ+D1w0pfITzb4CeOCTL4LFNtbla1x3dLDvpBLXtzGnp1TdjYIUo
+CyvFqoq9xIc0qtMt8j1XvdR+5VAfCa+ru0hA03MEBWk+YNcVmJZsOP/pMpIlr8Zp
+yiRA9xkLc5tjm4FHEfFSCNoG76QXMT7mvtPv9BA5g910Oqssng0d8klTiQDkXzTP
+0Anvpe85NI1VDdTCGNBdUXXFhLh6sYqWSEL/wtx004ZYtcz4+z9c3Qkxr/6Vr3+c
++9pzGbGG14KN5Spk0YfCT9H9o3soiU/f4oDtiQIcBBABAgAGBQJQomV2AAoJEEZj
+QHcL2TbxwBgQAKGkSfnCJgTOCbd+BQH56dihrBqRgf0uq10K+uE12aFp9QyISFLm
+6RJOO+JrVJWn7D8TWv3+XfAaqHA4zh3NiaZBHx32uZcTIxbDtnAdKWcj2SNuVb47
+CSaPrFaIVPfLG57vxaJSnZvANR04nGYuFznm0e87+vEk/o+A+VzUazZcQFAnGhQV
+EKTIuTgxOrdFWqNrUtupg5kyftV/cRRJDfxLRyfDkFGtjt9mKRD6yT3CE9SjTkBw
+Y1OKRkn32QLFlK1Q+d40VFGD+KvV27BgNTB20NPmt+NwbMKoje8KvJS4ivMe2OMG
+kIjOaa0pTqv17A9uxMw94lWB8UwSy3JLOaSH70pPMrSkBCp/AZgarEnL8S1Kvfot
+mAfjoIwbqy2zX1uDc4VsvoL0EEy6/mvARggeLyW5Bz676QGxM+1JfzJWtIChQOYB
+NSxjm2aCg1Go8dB5VOMliYgI+3DATvySrj6SuyWCRSe7UmMku7ztWVUdVFvdmLPE
+gmOFTw9HpFK1+d8dgkEcFVWswU/P7RB97O54aZjchc8DkYXuthFp/9lfuyrsUe9k
+AkpzdEY6FHdsAaiocg3g/0/hf+qT+TaCVPMBRnNRqFaYvhlKJfC0CdXshdIotoe3
+fagG0TZTPIXYbyBju82/juhsi6MbuYeEqp9tQpJUXm+TYp+ylXUt/gtJiQIcBBAB
+AgAGBQJQzN9zAAoJEExw8Ghv5Q8cnM4P/iGDK7cgTg74XMCqEB0T9Sb87hREVYVl
+EISfSBbrTyynAow/RL/RmvySCXqWHa9iobdZtDY2EI0US6LGBAjS/h2KlCqnyf7k
+mcc6o+2JgSdjdYYqj1x8d+De9negDIANpXxEe0HKXc+H0Cd9aVe8cUTULAAKcrIq
+i/3dbSA70CXe+SfA5FA/Ryt6j2uXiXdzPtFgArsZ267iTEUhbjnbwTjs4fcarbLo
+P3yttASbnR2iv4k1CWlMKJVHzdwOV7HPS4pYQ/6CXzW15pLdpWus11UcVxDQKZNH
+V6bkj1v0XjYMAWz+8ibq5XB+AyqP50JObAUZRcXU2fP/46RDtlX3Jc4BlJIz6Lca
+Bolft7peh6ZkGoZ6ZLJHTLfHbD5BVAobIIbezF2Vfmj/H+0PFdWJVkWC9kDRydub
+d8DtQKlyH6t0fJsO7kEZXaA7lH+16gbljlN6RIGhl077RWuVTWW1MlqDxJ6r7a6F
+3FJtQq9nFeoQRtCGhE+WqlFJc4KsmoYcvLyTqIg2cNkIOlC2TmoA3txLW2SsRHhY
+nYOBNw6Da8yZEe73iF7T/9trYktGqCwaYb+e9bjp9mG0GxXHDi7F4PHP25J7detF
+ihGLZU2YHMqcSzMHKfgT2ETm5K5HlECKq07j9fsaVeZFl9JcMwuLR9B+XQ0Dqn9u
+F47Xz01PRT7HiQIcBBABAgAGBQJQzOBFAAoJECsRil+hXzC5CKUP+gI2VeHSVGBD
+nBBYFfuQVqEtA0sV5SOi5tqmPeeQQtMeNEQ5tq7FWA5p75Se+fAfdzpeG8t0vURD
+kdW3T4MHvLFv6EXGKMMQ7J9mburgdRxH5mIS7GwrBR/p2ePH8aN9PeD5eQ5T/OLb
+dg/h4Tt34Xsonuu0I1v1rmaVkUIoPUEmIK6GteXfOjlPdqwMmWeb7Cns/oHCRuoz
+6lXJ6WbUtqthNB8Dp9C+uwu+5EPQeuV92CvmDST8QfxjKAnBefXqPucr40c2Mk/8
+SlW7aajj0ZsDSM1NQn8KVgCc/mBxz9Goc7JujZDkwPcKmFNr8ALakgDs5WdfHiKz
+FGxexLy+tYieV0Lk9L3dVQ4+bqzaaVbY03ePA89c0eT2ZzQl3xAgXww3ix4tJfcz
+ScnLKpqB9tXXyMkT5qrtmz9gjghXU5bqR5oquVoi8E7Fu9O51ZJ50WV+Mik5NtPl
+Xq9Y03fC7WlSM9RoZ7lPcImy52i0s6E8ecmBQ5q8KHOh74jKET9JtMivRRBv8UAl
+aHHeVHQAHvD6bAVmCgUsXxar/0Nh32HgdVJAazJEP/BenzgH3rT/utRIGNIrTWnE
+63zEQeJVDb21LIqBu4Urw+ztnTVJBChYH2nqMp02YTMuInK0s26gSWxlSdLwuorY
+M+rPz56Uo6Y1LmOt6+7Lb862kXCEADwuiQIcBBABAgAGBQJQ5CaMAAoJEB9A1arl
+xTKm2dkP/RowQc7K9D24oxC/3lDmAQA/6VVq2dysMdB7eq02ua1h5zWNwTNr+4pC
+Zw84NuSozfBXQ4a0ho9VP2/R48igD1fsBwIMW3Xjy1fvUiQDG7hMGGdTUEInJqBx
+VA1uWynn86SqnGunsTR41M5Suqyb1Q/KBQ5jcWkvKZGjkeOgjwu7sVCpVvARVLh2
+v88g70BhLHhLdN7eeJ+daP5VegSh/dovd0auTUxkecwtIDo8QzPNlEvwBZg+6kzX
+ZL2XE7FbaBaKyykwiKKgyIyvKtltIMGhjjwqN0HR+SsODCvfhX1dugoorCgQhT7S
+YoCHDCVDCVy19vUPr5MVnZZd7kk1bKLdns7IVGpYmbtoBFRq0RyC+a06uf00Uzuh
+qKDshQQNWorlH4C/sN6vagrb+SdtS/xbHbxRoZWwculjxSKFxykYNsPJHmaogBTG
+pHSYrTJEsOQAoLsOV4v+lM/nOXiHT4C08mNILdsdf051OSDrhbP6Hmd45uKvx2vx
+clgiTCQSiVKapT7c6DjEHNs2q6yttLQ8Ga/3n8HJFLOhpqEZCykyZ3eNmTsaQl0G
+YaAvHRFv582B/mWCZjb2VQwVEYv1MpU+wHgbdS72G8C/FiINiow4wpUZsLtr3pF1
+vgi5S4IKdgACyjjyVX1xBH3lIa10ud6xpJINlgX9cCdJKkiAylK5iQIcBBABCgAG
+BQJUFgnyAAoJEAEbRra2zTKABZwP/3WrxjSENwaHh+F/lvTVDzqyIwH/A9Gjj6v4
++XAjrx0s0KWHyvNJVPGloR4gMpOyzqjXVaB4IJ8IPmQICGBYWf7qEQcPM/dn89z3
+ofr809LuCQcEi5B50JLaPioRTuSXGd/AzCJXPlH97ez5qoME4gu0kka8xch6zlyZ
+P6MkrBedMdNQ58nEwp7xHKr+g1e84Ibc6/DsY9J4cqg8Aid6Ts/iuvzEDHpzD0uw
+Am/S1yyF5y1zqEq7OmYICa1C/ij2YBoIBuOA9/byMi+d/2aCCEthHO0dQAXyzsvc
+2Yedff3gGBrdgzsaSmS0KoMJJNG2SvM/3fzuVqHZke4w4zUB5fHcFU3eWH25OWMt
+/mb94nSZw0C+VNpbdQpVGLmpuNj5H8tyai22UX5sUvTR3ZBUgkGTxDvSa47mgDYr
+kQig1bvVlVnqJnB7EATCm+NealZ2575Ju5BTAsrTy1jY/018rDNBNIopDRnh5ffv
+f2YhYfCjJfWls56hWWBhLQWyiEZmeLxm/Xmxbs0gRXYAFerswGGvI+sKPrIxCKUa
+G1sxArrMEgeSYKWI2rQs6Vm2xjkCdUyCnP0BJ5tQQSjK6BJuwe30PA9mVySiWmlU
+Z33Z1/h5Rp7DndSWKBuOKfvwe5y8CNm0jyZL1WQvfi21TR6IZjm2AoosUReXDuBY
+Jde4VtMGiEYEExECAAYFAlYNhEcACgkQiPPPTCLX9uxulACfU8UDOiB9G392gbv6
+8B6qTyab6SgAn3rLqJOnQqlT9WVYhvYCV7HxA23XiQEcBBMBCAAGBQJWDYQgAAoJ
+EHkD+BGQkQqDQdUH/A2ez/BXO1El76Zviz6sMh3XEdbEJczpSHekl8lpmmcetIj3
+PYyMcmODqTePH7ipsFestTEPKW/dG589iPIG7oCN5n+fnEqHlLdn3WQwdAKCjwpy
+GrblPShAc30OWARQPUOjd2btyWzgeLYj1RO4M/A3Z11x6R1uIy6mksn7itrOEaL0
+mV7Gm4Edmfth3iSmu6DqcUGmvlarTGcYstmWt6jx/dkfxV7MbQGVpU+rKfQQRt3A
+tw0pm38oNAAMO1N14ZZ/n5xMroSRPi39xfZEdGZBl6+JsInVfis6yrP+HI9XoWD+
+COSTw8qz2xWlft+CM69VQY2m8xIBZ39LewZmGnaJAhwEEAEKAAYFAlYNh0gACgkQ
+Egnn8T0MkrkH5hAAgWnFNtxHUcXvRCdrEMBRdlYv3PwWQMvHVNBULjuxmvX+xHYd
+FyyjjYugp3sj3OjjOlyklWSoNvoY3PSGaxtJXbTe2r8TVXMAw7Hh4bU2t05O3aPM
+8lC1PhEUNMBHb5ZB8BNjH8h4v9EMJJIlwih3vKfl0UW1gw3+JvV2BMeRUfZvjiMN
+lSpYUOkAGzX0ZtVyTCW3PzEpCvwUf15LWhjHybehCH43MrP83gbhvWqguINtVICg
+EHr7S4kCsp6A6pDn24dqfyZ5DQi3bl8H7Q1gmcPimAIs45/mkvbYsJD6nlT++fIp
+YETSqvSOE2bbS/nGfkCpwJN/TeWgwT/VQQ0Zyfw4ELajnj0u1O2huHrDSde2M25n
+hRTqPciWFkE1kwi4pmFg7F/yvyhAAnvHQ8QxRTPsnVn6zQ6f+lDnHDTiVLi2Hstq
+iE9Z0PxO+ObXbG5R2i919v0nKW98PGKBPEm7Rt+5X7D6n6SpnEyhfhypOdzHPVWf
+65kUuHXkqoRITiMQ7WaO98msmyEte+5i68tICFtTRJM9HS0dSy4DAZiSnqIOflMi
+puLTxDx3BoiRJcXMrG/QghaR2FoBQAy13pcHBPqs4imoSca+CagJ9jngXm9+jUU2
+GtWT8hURcQnwIt4r0gNUIvfQLCAhzqu7Yjv60qdm09grcDW3TQwbiKt4XEDR1bDV
+rgEQAAEBAAAAAAAAAAAAAAAA/9j/4AAQSkZJRgABAQAAAQABAAD//gA+Q1JFQVRP
+UjogZ2QtanBlZyB2MS4wICh1c2luZyBJSkcgSlBFRyB2NjIpLCBkZWZhdWx0IHF1
+YWxpdHkK/9sAQwAIBgYHBgUIBwcHCQkICgwUDQwLCwwZEhMPFB0aHx4dGhwcICQu
+JyAiLCMcHCg3KSwwMTQ0NB8nOT04MjwuMzQy/9sAQwEJCQkMCwwYDQ0YMiEcITIy
+MjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIyMjIy
+/8AAEQgAyADIAwEiAAIRAQMRAf/EAB8AAAEFAQEBAQEBAAAAAAAAAAABAgMEBQYH
+CAkKC//EALUQAAIBAwMCBAMFBQQEAAABfQECAwAEEQUSITFBBhNRYQcicRQygZGh
+CCNCscEVUtHwJDNicoIJChYXGBkaJSYnKCkqNDU2Nzg5OkNERUZHSElKU1RVVldY
+WVpjZGVmZ2hpanN0dXZ3eHl6g4SFhoeIiYqSk5SVlpeYmZqio6Slpqeoqaqys7S1
+tre4ubrCw8TFxsfIycrS09TV1tfY2drh4uPk5ebn6Onq8fLz9PX29/j5+v/EAB8B
+AAMBAQEBAQEBAQEAAAAAAAABAgMEBQYHCAkKC//EALURAAIBAgQEAwQHBQQEAAEC
+dwABAgMRBAUhMQYSQVEHYXETIjKBCBRCkaGxwQkjM1LwFWJy0QoWJDThJfEXGBka
+JicoKSo1Njc4OTpDREVGR0hJSlNUVVZXWFlaY2RlZmdoaWpzdHV2d3h5eoKDhIWG
+h4iJipKTlJWWl5iZmqKjpKWmp6ipqrKztLW2t7i5usLDxMXGx8jJytLT1NXW19jZ
+2uLj5OXm5+jp6vLz9PX29/j5+v/aAAwDAQACEQMRAD8A5AU7FAFPFfONnoCAGlxT
+gOKcBmpuMQCngUAU8DiobAAKcBTlWnAVNxjQtPApeAMk1Ue8BmEaHv1rSFOU9guX
+VQngA012ZCF3KuRnmrqxiK2EqjJI55zXN6hqRR/u7e3HSt44Z3E3Y04btGbax+b2
+HWnS3QjIO07em6uXa73y5EmPfNWLO8lnkYK6YAwWfpWjwqWpKmdMjq/3SD9KkAzW
+M6zCLzYpAHHZOlSWWtKZBDdDax6NiuaeHkleJV7Grto21KAGGRS7a5myiHbSYqbb
+SFaVwIsUYqXZRtp3AixSFal20baLgQgciipNvIop3EYoHNOA5oAp4FbNiACnAcUA
+U8LUgAFSKvNNA5qUCpbGIBzTJ5kgTcx+g9aWaaO3TfI2AeAPU1harcxqRJK+WPRA
+1dGHoOo7vYmTsOudQuZJG8vO3oAFIqoHlgmEsuCevPGKzP7Rmd8RnHoFqtcTuTmQ
+nd+derGnbRGLkdHPrFwU+SbH0NYl1fO5Jd8k96y3vGGQM1UkmZ+9aRppESqXLjXb
+eZjdx61p2V+keAhyw/iIzXNhvm+ati0eBowNv6VcooiMtToJ/EEotth3bR6AVlnU
+hK2SfzqvMgC/KfwrPZnVyuPxFQoRNHNnfeH/ABAJmFrcOOeEY/yrqgua8YjmeF9y
+k+xFejeFPES6pH9lnOLhBwf7wry8ZhXFc8NjWFS+jOi2UmyrG3im7a8u5qQbKQrU
+5SkK+1O4EBWkx7VMUo2UXGQhckUVJs5FFO4jBC08ClA70orYQgHNSBaFWngcUmxi
+BfSkmlS3haWQ4ValArntaujcX62cbYVBlj71pQp+1nYUpWRWu783Ugby90hz5a54
+Uep96y5dLuJ5sysd55xnoKutEkPPJAGenao/NuLhGKLtQfrXswSirRMXruQ/ZFtU
+cRDP95z0/Cs64+UlSMsecZ6VdmSRXBn3EDkL6mqqoXlZyPpnqa0RDMqQfN6VCavz
+Jltqjn1quYTkgDOO9apmTWpXAyelWIH2tjmmmLjjrT4UOfcdqGxJMvmTK8/mKhA3
+n5ufQ0j5UZHQ03PcGoLI5fkbGetPsLm40++iuYThkYEHtUUj5yrD8aiViBiqsmrE
+vR3Pc9LvE1HToLlCp3qCcHoe9WiteW+CtamstTS2eUm2lOCDyAa9WGGGRXzeLoOj
+Ut0Z2wlzRuR7aaVqbFIV9q5blkBWkK1OVpNtO4EG3kZoqUrzRRcRzoFLtqTFGK3A
+RRmngUKtSKKTGA6GvO57iQarcNkj5yDXealerp9jJcMM7eAPU1wjI9yrysMO5LE1
+6OXxesjKp0NFytyURGHzAKackzQTeTEnmYOOOKw7SeSG5wzcJWtDIC5dQea73ElS
+GXMc8z/vNqZ/hBqA27AErzxyR6VeK+YckEn6VFcNsjIAOD2poTRkyRsR0AUdabHb
+STcIuAT1q9DbyXUi7unTArpNP0sKB8vBpSqcqLhR5jmk0VwoO3IPU006Q6ckHjoa
+786eqxhQvH0qjc2+ByvArNVma/V0cRcWbw/eHyn9KpvHhdw/KurvIkkhYEfMvqK5
+udcKQeOa2hK5z1KfKZ8nzDioNhB5qxjnFAXHJFa7GDVyezPkMshPAINe06bdfaNO
+tpEX5WRTn8K8Rzk85xXrPgWRptGUMxIXgZ7e1eZmcfcUjei9bHSgcCgrxU+0YppW
+vBOqxX20mKsFaYV4oFYgK8iipCvIop3E0c1TgM0AU9RXRcBQOKUCnLTwtTcDA8Qb
+msjG/Clh2zxXNRn5flyQK728tUmtn3DoCRXB6dLC10Y26MxAxXr4GalBpdDKe6KM
+sTCYyKMqa0oyiwDaeCO/au3h0ewg0lI5lA81c+Z9Otc7fW0Ic28EgManO4DOTXT7
+W7sWqWlzLjEgJYN8vv0pUgM78Kzse/at7TtCu9SCrFEfKHBbGBXe6P4HS3iV5hjP
+tyaHPsVGn3OM0PQHdt7JgDiurh0hY0A2jIrrE0iC3hAjjAA6DFVp4lQnA6/pWLTe
+50JpbHONZ44IzWXf2jIpOODXVyR+w+tZt4gcAFcr1JpWG5HB3SAspcYB+VvpXM6p
+ZNFGHwdr9DXoF9pnmMSuKpzabCbL7PIgK4x7irjPlMpw5keZNbSDORj+tIRtXmuu
+udGFtFlpWeMdq5XUMLIQPWuuMuY4akOUrKeTmvYvAVo0Xh5JGGC7lh9K8XBO/HrX
+vvhMg+GLDAI/dDrXn5q2qSXmOhqzVxQVqTbQRXgXOshK0wrU5HNMYUXArkc0VLt5
+opoDlgKfiminqK3bJHKtSqtIoqUCpbAjkj3xOo/iUivI2E1tqLR4Kur4+nNexqpP
+AGSa828WadcWniEtLGUEvzL6mvTy2WsomdRdT1nwvbWmr+HkguYllG35gw6GtyLw
+dp0saRtBGIl5AUc/nVTwRbrD4Ztjgb2XLGuoiJBGOldTep0xWgtrpcFoqpFGqog4
+GKsS4TpyT39KlBJXPSoZx8uQCaqwt2V5mwCMVkXP3uavlyX2nnNQXMOcNg5pFmTK
+3GaoTsMEd60p068DArNuFyuaTAz5ccmsy55BFXJiwJzVCZyc96kDK1BcwkV5/qcZ
+SZsjp616TIwyMqCD1zXFeJ4UjvsKMAqDiuii9bHNiI+7c560iM93FGFLF2A2jvzX
+0bZQx29nDDGmxEQAD0rwjwtGh8SWckmfJjkDO2OgFe+QSRzxLLEwdGHDDvXnZvJ3
+ijPDx0bJAKCBS0YrxrnSMK0xhUpprCgRAV5op7DBoqkwOTAyKeopAKkUVsIegqZR
+mo1FSrUsRfsB5ccsy48wFUQkdCT1rN+J+lCTSrG9KZvFbZvBwAuMnIrTtSRYz46q
+6N+tUviHfec0WlqfmXT3nwB1JZR/IGvWwPwKxtJful5m/wCD/m8O2oB4K8Vtz6tZ
+2TBCwdx1A5xWF4XTyPDtpFnH7r+da8aWj24YCIjGfMJwPzro3ZN7Kww+NNLQlGkO
+fpViPX7S8A8p/wADXIaydJgnLNPYFz2EmCarW17bgboYwFHdWyP0qtUCs9juHlVn
+UrjH9KjubpNuBx71m6ddC7iGxs8dAap6tcvZqxYH2pXKJLzUYLckucjFcnqPikJK
+yRQkoO9KZ2viZJAqRZxvc8H6etBOmQ5Mk9smO7inYluxlDWrmf5lgJGe9TLeCZfn
+jKNnt2rQTUdPbCQ3drIPRVqrdou4MmGQ9wOlDVgUrlZ8Y5rh/E7M+oAdPlArtp5Y
+reNndxtAzz2rkbi1l1fUklSFxBjhiPvAdTV0t7mdbVJI6Xw9pEVrpaShdrsAS34V
+1nhe5LfaLYkkI2RWPpkkVzYLHCGRUG3YwwRWp4ThIuLyUj+LFcWMfNSlc6OXlhY6
+cikqQ03FeEYjdtNIzUhphFAEZA4opT1oqkByQqVBUarU6DFbMSHgU9aRRT1WokBd
+0/55XhJ4lQr+PUVH4osf7Sm0ufAV4xgSbcnGMFfp7UyJmikV16qc12NlCtxCZFVX
+RvnQMPutXo4GppynRTa5bM898WX+p6DbwQqyi0SaNGdV2krg5B5PcCuoj02HUdLt
+SzSGARgqikgEH+dXdZ0SHXre7s7leX9Ox9vxqbQLOSw0W30+4cNNbp5YPQsB0P5V
+3vXVGSTUnc4bxN4Xtr6aN0m8pU2/LHEOce/Wo9P0cJdJ5QcIFAI6fjXc3Vu0jlSc
+evFUZI4oI2UdW4+Xqabbe5UYxWxkaI0k+tSWGnyNEqKZC+zPcg/qKz/FkF9axySt
+fSTqvLKwHSu30HS/7PEl1IgE8w/JR0Fct4lcNKykZDEgii5UYaHEJfnUrrycH9x8
+ir049R/OtaLSoJLGS3k3kyYJf+IfQ1Fa2EEEqNIxQp8qShcjb2D4/LNbEcZbhSje
+8bhqpvXQzUdLSMO18PWdgpRBI44+/wBau21ixlEaZCNnIzwB3rWW3fI3Oi+pkYCn
+TT29vA0cPzO42vKRgY9F/wAaTberDlSVkcvcaVFc2sPnGVmj5fL5Dn6VPaAGZUAA
+6ir02Nhx0qhbnbeo3YHNTdspRSKsDywap9niPy7snHauy8ML+5uZMcNJxXPravHC
+zEr5kp+Ygdq7DSLX7LpsSAYJG4/jXFj52p2LnL3TQBoIpvSnV45gNIphHNSmozQB
+EetFObmincDlVWpRUa+1SDrWwrkqipFWmJUgqGCHV0fh7UEjiNvI2CDlc1zuOKkj
+zWlGq6cuZFJ2OzSVTeybWByAcirjIjoQ4B+orl9Hl2yupJyelbBuWGQxxj0r2KVT
+nipI0WpVvhDFglzjsuaq2Ukcs+2KMfLzn0rP1G5ee58mEZdjj6VsaZaCzi3kFnI5
+rS7bNdEaobdnjoK868TNtuc4713iXsHlyEvtYdj1rh/EpjlAfcAc5ptAjBFywGFX
+J/nVm0mhnP7yFQ3fFU0uYMeUoZn7kDgfjVi0tgpLk/MxzTsGhvW8EAXcEA/CobyJ
+DkqKrx3ZtiBLkp/eHOKkuZ1ePcp4PQ1ImZr8Eg9KpIP9KHXHsKtvlsnr71peF7YS
+apNOeRGmB9TWVSp7ODkZN2G2VjLdyRryYwcs3YCusUBRgdBTiABwMfSm5ya8aviH
+WdyHK4h60opuaXNYCFPNMalLc0hNMCM9qKCeaKAOXQd6lXrUSVMtbisSLTw1MAp4
+FSwJAeKkSq4ODzU6VIzQ059l0vvWxcqWTCcMRXPxPsdWHbmujhkWWNX6hhXpYKd4
+uPYuLMy2t0t3Bc/vW5JNa67ynC9vzqpeWC3LKD+Y7Vz2qHxLojCSzk+3Wu7JjcfO
+q+gPevRiXa5u3VsXB3IefauQv9P86Vhtbg8Z5rrLaW+vbeOeK5iZXKjY3BGR3qhe
+Raoly0KxRbiC27PHpTBOPVnJNp8iPtC4pwV4T87AcVLqCXFulw91dohjcDC8k5x/
+jXMXljeapdtHHNJHbZwWJ5Ye1NK+4m101N+K9haUxiTe3fHNWhbmO3kc/dz8o9Kj
+0jTLewgWONRgdT3NXdQlUIsaDis5bj2M5sCI+ldF4YhEenPKfvSuT+A4rmZ2Kwgc
+ZriZdZ1GOeX7LdB0Vj+73EMKiph3Xhyp2OetV5D3ZjUZIB614MfFmtKcC7lX23Gp
+F8Y6yDk3T/nXOsnfWRh9YXY9waVFPLL+dMa6hU4Mg/OvGV8ZaiP9ZKzH1zTm8X3L
+9Vz9a0WUw6sX1hnrraraLn96DjrUD65YpyZfwxXkLeI5XJIJBPeq0msyyHJY5+ta
+rK6S7kvEM9ek8SWKnqx/CivHTqkxYfWiq/s2j2D6xI9SQ1MvWoF7VMprwmdZOKcD
+TF6U8c1DAXrUyccVGBxTwQASTgUhlbU71raERxczSkIgHXniu6vbY6fcW9vs2oLa
+PJ9TjB/lXBeGok1z4gWETDdDE5lwfRRkfrivYPENmssKS7fu/KT7GvosPhlSoeZz
+Kreqc9Cd3vipyiyxFGH0qgpaCXY56dD61b3457ULRncmZN7pkTFpI28qUc7l45rG
+kWTzFEt0zNGDhieea6K+BkU4/KuauopgTtjJ56mr0OiLi17yMy4ggBdlXc7dWbk/
+nUEMPBOOautbMPv8GmlSvygYFK5M3fREZcRqMdqqySAtuNFyfLO0mqZbzDtHfrUb
+swloI2XJc/d6CuN8YaQNN1uYwcRuBKgHofT8a7S7YW9nI/ZFLVn+MoGuPD+larGP
+mT92x9jyK3oaM48QtDzz7QXXEo3Hse4poIJ4NE6KH3pwrc49PamD2rrOMlxx6ilB
+x9Kar5PP5+tPxz70AIR3oNOAP4U7FIBqjkA+tFPA5FFAHri1MPWoVqUHivkmemSq
+TUi1CDUitjmosBMDWbql4ViaGInJ+8R2pbjUACUjOfUg1lyt5gbceD2FetgsC7qp
+U+45qtXojd8BXaWfjTTiwIWUNFuIxyRxXutzCLm3eM/xCvmbz3tpYriNiHjcOpHY
+jpX0Z4e1aLXNCtdQiPEqDcPRh1H517MdVZnNezucxe2/LxuCGU4+hqG3Yunlv94V
+1Os6W11EZoADMB93+8PSuLaXc7AZSVDhlYYIPoRXHUpuLPRpVVNX6miLcEENVC7t
+I1BJamrqwTKy8OKz73Ug5OX+XrmkdMWQzwx5JHNZsm0E5wCDii61GModkgP0NY8l
+6zny48sx9O1A5S0IbyRpZ2VOuadBCFPPJ706KEqCT949TViOPGaLHO22ZHiN/K0W
+4PcrgfjW5c6Y158P3tgvziEMv1AzWNrMP2y4srAc+dKNw/2Ryf5V6FbWwXT/ACwO
+NuK1pHPX2PnGVD6c1Biuj8UaY2k65cQY/du29D7GsFlwSR0rqRxkYXv+lSocdab9
+48U4elDAkxzmlxg5JpqN27U8c0AOUcj60U5B8wooA9UU1KrUUV8ieoSBgFPPAqjc
+3qkMFPyL1Ioor0sBRhJ8zOevJrQy7WVrl5XH3AdoNSsMcHn6UUV7COQZIwK4Ax2x
+/wDqrvfhL4hFtfz6HcSgLNmSAejdx+VFFUgZ7F2rF1rw7baqvmKfJugPlmTr9D6i
+iiqlFSVmKMnF3RwOoW91p0/2bUoBg8LIB8rVny6bZXIPD89g5xRRXBPSVkenCTlG
+7Kkmj2sa4SPI9yaptbiHKpGFHtRRSGMWIs2T0qfYETPSiimK+pR0KD+0vFkkn8Fr
+Hgf7zf8A1hXowt9sRGMDFFFdFNaHJWbueV/EvTt1tFeKvMT7WPsf/r15pgGiitkc
+4wrt6UmPSiimBLjA+vWnDA4oooAcvUZ7GiiikwP/2YkBOAQTAQIAIgUCUcwZMwIb
+AwYLCQgHAwIGFQgCCQoLBBYCAwECHgECF4AACgkQRxXcAmQovbrZdggAgfYYlqVy
+iGfZWDCfvC4S6d1RncQFfmTv0SAMGv+uUgD3V23MH927VVNCPPA86ImYGMVEp98j
+3ZOsOssYc6c6mL9ld7icG2pEFeZZjwQwCE/5S79UCIb/gsOXPPY2KDu2WrNte8MX
+Rnf1VNmoYVzpTkrOfxJ4UOAmWv79+vBTwjjfJDz/7QhTDnSmKRsDLF5ZLWa/Cc0X
+NfTU1Ocgc4wMQ/zsgoRND3pqav918B6UlIVBRgMWt3oBceYG2xi/RsSO8a69n4b6
+Rm2hXFev849TKwl4ipaxQVPJiQ0KYgvyg3RQosQ3EkRLRJMxBq1/Zy5ptvsRJ4is
+YirA/B52cmygOYkCHAQQAQoABgUCVg2HSAAKCRASCefxPQySuVFbEACdAIWuLbiX
+UfWlSMX9U8nImLYPisFKY+53q75aOQcrrDOZvB3jncJtcVE1Kg92OBbC1oru6W5q
+AL+2UrJzSn5a59trsWRkSPPv7SaGYrLDGBCiWt8/l5Lctyi7Tj2YsQQ4xN3qXXQm
+oWU8auFUq/MqSUfGjzOi/9xvvjZRGeZxNb5SngDbP9EVFkSafWNpyzLo3NHM1qvk
+ko/bmnfhnHxdXzOlS3Zypx7VuMHiiVhUl972L4P2KRXrIj6L0Qebj7lRsPDcoSg9
+d1cO72WV64bLJJJO1j9R5mPUZObe3zmYt3qa0LCPd/eLEObsfWZlyHdSMO3kTYv9
+mFyvqshATukpXJoJ0J7EbV0suUaAQKc2rb0WAeGPwT/QQ1oAibS/YSs9qWzizNMl
+L7K/Da+sVA02UoccZzc9sH+xE6qZGVRbNPrOisF+CnxPo9c888W93bY1VskIthwb
+z4FTYA+SQbXjr3SMniH3jg8YBydQbBDSerjOSSzvtW8d2/XmCvH/2sC2u13if0Xi
+wD3xWf2tHf5ls+dnhjX1f5RbjXMJsB5xRmqo9M36DbtuOSL0eZk6fH4/MsiNftU6
+nYZ0nSP5ebRmp8pxeBNHbv/dm+/igMXk6Ylc2mQTOpcScw34S282EZpOwyfHGVZK
+7+y90WuvgEdj7A65wltrz6CcRrVn9eo99rkBDQRQmDNzAQgAwQXlWLjaY/3nIUNX
+LXvUHjG7HZ02fv+rcfeH+3ZNyqmyE/+vpKHqedNySPvfYJ2OKIvsjThaVX31qjOk
++xXliUHi15bQsqOPTtc16dGZtzGT4wOwvc5fQPsSXPRauw2I7XnpNy2pbw9ZG+RH
+XVgnuVNYs93CR5uHW9fL5d1Tl/gA0EBKdoPGP2eFBgtYXnqIMv8VG2JVLLajfdUS
+8ktl+AaTWQB+MH6LjtTDMpwj9W6yEM+5EgM9vtLpSLNyo/gL0egDXdXNO685wbNl
+8Lto+WP9a/7dochUcEGhAgg9lRwpsovgwFuwG6x7KfCLeJjPvuaNQZ+fJt6lgM4i
+Vv0sHQARAQABiQEfBBgBAgAJBQJQmDNzAhsMAAoJEEcV3AJkKL26r5cH/1TkTV0d
+6xs5qyZ6+5Yl5U1uSwvLy01NemcF5YTfpokyrPUENZ2LodPP+FcJmQ6QdQ++L/5Q
+zMmCEOixse235XryzDkXj6cvhNNu23a6vqj6zb0VoEtJMBO9yy8HOzmPfbOUQ2G7
+z0ipITawdhZ1Ay1ovXJS2X55F0SF3jxn40mLbliPseVjD3L+TE0e4IQs77hQ14LS
+Ug+N8t93JV/afWA7xVk7zQv/GtYEHkYChMjJW3l65BQA2ZMwreDgvN9/qHySrYAB
+Oj4Z1wXzPsjUogSLO3/a2XEGvxzLAcxgaeKnJXLaSgKrTP1xhsnBE0s0mmORAHgh
+TQyhrAr1Czb150Y=
+=wbZq
+-----END PGP PUBLIC KEY BLOCK-----
[2/9] cxf git commit: Update -Psetup.eclipse to setup workspace based
on changes needed for latest checkstyle
Posted by dk...@apache.org.
Update -Psetup.eclipse to setup workspace based on changes needed for latest checkstyle
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/3198d8b2
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/3198d8b2
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/3198d8b2
Branch: refs/heads/3.1.x-fixes
Commit: 3198d8b2bba180fe40fb6729695b0d0226dd02a8
Parents: 7a365ca
Author: Daniel Kulp <dk...@apache.org>
Authored: Thu Jul 21 10:58:37 2016 -0400
Committer: Daniel Kulp <dk...@apache.org>
Committed: Fri Jul 22 12:13:14 2016 -0400
----------------------------------------------------------------------
etc/eclipse/template.checkstyle-config.xml | 9 +++++++--
pom.xml | 4 ++++
.../java/org/apache/cxf/binding/corba/CorbaConduit.java | 9 ++++-----
3 files changed, 15 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/3198d8b2/etc/eclipse/template.checkstyle-config.xml
----------------------------------------------------------------------
diff --git a/etc/eclipse/template.checkstyle-config.xml b/etc/eclipse/template.checkstyle-config.xml
index 8f6dbe4..ac7151a 100644
--- a/etc/eclipse/template.checkstyle-config.xml
+++ b/etc/eclipse/template.checkstyle-config.xml
@@ -18,6 +18,11 @@
under the License.
-->
<checkstyle-configurations file-format-version="5.0.0">
- <check-configuration name="CXF Checks" location="@CHECKSTYLE_CONFIG_FILE@" type="external" description=""/>
- <check-configuration name="CXF CORBA Checks" location="@CHECKSTYLE_CORBA_CONFIG_FILE@" type="external" description=""/>
+ <check-configuration name="CXF Checks" location="@CHECKSTYLE_CONFIG_FILE@" type="external" description="">
+ <property name="checkstyle.suppressions.file" value="@CHECKSTYLE_SUPPRESSION_FILE@" />
+ </check-configuration>
+ <check-configuration name="CXF CORBA Checks" location="@CHECKSTYLE_CORBA_CONFIG_FILE@" type="external" description="">
+ <property name="checkstyle.suppressions.file" value="@CHECKSTYLE_SUPPRESSION_FILE@" />
+ </check-configuration>
+
</checkstyle-configurations>
http://git-wip-us.apache.org/repos/asf/cxf/blob/3198d8b2/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index d2c233e..26d8201 100644
--- a/pom.xml
+++ b/pom.xml
@@ -276,14 +276,17 @@
<whichresource resource="/cxf-eclipse-checkstyle" property="eclipse.checkstyle.url" />
<whichresource resource="/cxf-checkstyle.xml" property="checkstyle.url" />
<whichresource resource="/cxf-checkstyle-corba.xml" property="corba.checkstyle.url" />
+ <whichresource resource="/cxf-checkstyle-suppressions.xml" property="checkstyle.suppression.url" />
<mkdir dir="${full.eclipse.workspace}/.metadata/.plugins/org.eclipse.core.runtime/.settings" />
<mkdir dir="${full.eclipse.workspace}/.metadata/.plugins/com.atlassw.tools.eclipse.checkstyle" />
<mkdir dir="${full.eclipse.workspace}/.metadata/.plugins/net.sourceforge.pmd.eclipse" />
<get src="${checkstyle.url}" dest="${full.eclipse.workspace}/cxf-checkstyle.xml" />
+ <get src="${checkstyle.suppression.url}" dest="${full.eclipse.workspace}/cxf-checkstyle-suppressions.xml" />
<get src="${corba.checkstyle.url}" dest="${full.eclipse.workspace}/cxf-checkstyle-corba.xml" />
<!-- Add checkstyle config -->
<copy file="${basedir}/etc/eclipse/template.checkstyle-config.xml" tofile="${full.eclipse.workspace}/.metadata/.plugins/com.atlassw.tools.eclipse.checkstyle/checkstyle-config.xml" overwrite="no">
<filterset>
+ <filter token="CHECKSTYLE_SUPPRESSION_FILE" value="${full.eclipse.workspace}/cxf-checkstyle-suppressions.xml" />
<filter token="CHECKSTYLE_CONFIG_FILE" value="${full.eclipse.workspace}/cxf-checkstyle.xml" />
<filter token="CHECKSTYLE_CORBA_CONFIG_FILE" value="${full.eclipse.workspace}/cxf-checkstyle-corba.xml" />
<filter token="APACHE_HEADER_FILE" value="${full.eclipse.workspace}/apache-header.txt" />
@@ -297,6 +300,7 @@
<!-- Add checkstyle config for 5.0 plugin -->
<copy file="${basedir}/etc/eclipse/template.checkstyle-config.xml" tofile="${full.eclipse.workspace}/.metadata/.plugins/net.sf.eclipsecs.core/checkstyle-config.xml" overwrite="no">
<filterset>
+ <filter token="CHECKSTYLE_SUPPRESSION_FILE" value="${full.eclipse.workspace}/cxf-checkstyle-suppressions.xml" />
<filter token="CHECKSTYLE_CONFIG_FILE" value="${full.eclipse.workspace}/cxf-checkstyle.xml" />
<filter token="CHECKSTYLE_CORBA_CONFIG_FILE" value="${full.eclipse.workspace}/cxf-checkstyle-corba.xml" />
<filter token="APACHE_HEADER_FILE" value="${full.eclipse.workspace}/apache-header.txt" />
http://git-wip-us.apache.org/repos/asf/cxf/blob/3198d8b2/rt/bindings/corba/src/main/java/org/apache/cxf/binding/corba/CorbaConduit.java
----------------------------------------------------------------------
diff --git a/rt/bindings/corba/src/main/java/org/apache/cxf/binding/corba/CorbaConduit.java b/rt/bindings/corba/src/main/java/org/apache/cxf/binding/corba/CorbaConduit.java
index 7518ece..98218ca 100644
--- a/rt/bindings/corba/src/main/java/org/apache/cxf/binding/corba/CorbaConduit.java
+++ b/rt/bindings/corba/src/main/java/org/apache/cxf/binding/corba/CorbaConduit.java
@@ -189,14 +189,13 @@ public class CorbaConduit implements Conduit {
if (request == null) {
throw new CorbaBindingException("Couldn't build the corba request");
}
+ Exception ex = null;
try {
request.invoke();
- } catch (SystemException ex) {
- message.setContent(Exception.class, new Fault(ex));
- message.setSystemException(ex);
- return;
+ ex = request.env().exception();
+ } catch (SystemException sysex) {
+ ex = sysex;
}
- Exception ex = request.env().exception();
if (ex != null) {
if (ex instanceof SystemException) {
message.setContent(Exception.class, new Fault(ex));
[9/9] cxf git commit: Recording .gitmergeinfo Changes
Posted by dk...@apache.org.
Recording .gitmergeinfo Changes
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/2e62f6e3
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/2e62f6e3
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/2e62f6e3
Branch: refs/heads/3.1.x-fixes
Commit: 2e62f6e3009c0b68d9dbe2c33f878a2fbc8f9126
Parents: 63456b2
Author: Daniel Kulp <dk...@apache.org>
Authored: Fri Jul 22 12:13:28 2016 -0400
Committer: Daniel Kulp <dk...@apache.org>
Committed: Fri Jul 22 12:13:28 2016 -0400
----------------------------------------------------------------------
.gitmergeinfo | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/2e62f6e3/.gitmergeinfo
----------------------------------------------------------------------
diff --git a/.gitmergeinfo b/.gitmergeinfo
index 90b4b68..aebb000 100644
--- a/.gitmergeinfo
+++ b/.gitmergeinfo
@@ -71,6 +71,7 @@ B d26570e26de80d0e0c6870ae5d959bfc8c0d19e4
B dfc84b0fac84e001069800b32f01529d50339cdb
B e14d803ff84be7c59dd7c2321f1090231967d45c
B e1c60863ac10b56d423613c7d3d2f45c7ce18e14
+B e2efd0386e210d162c7646f4a4f2aa15e9752924
B e61a83d4ac18fc760fc5d4bc8fb19d6072aaac52
B e6f0722615190861192972a51b3986b34efd1f0e
B e824784db1bda81c68aaa407f471588acea38865
[8/9] cxf git commit: [CXF-6966] If classloader is null,
don't use it for loading the bundles
Posted by dk...@apache.org.
[CXF-6966] If classloader is null, don't use it for loading the bundles
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/63456b2e
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/63456b2e
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/63456b2e
Branch: refs/heads/3.1.x-fixes
Commit: 63456b2e231e61fb6cb3f12b2a06af27f77a14b5
Parents: 347a54c
Author: Daniel Kulp <dk...@apache.org>
Authored: Fri Jul 22 11:31:05 2016 -0400
Committer: Daniel Kulp <dk...@apache.org>
Committed: Fri Jul 22 12:13:28 2016 -0400
----------------------------------------------------------------------
.../org/apache/cxf/common/i18n/BundleUtils.java | 24 ++++++++++++++++----
1 file changed, 20 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/63456b2e/core/src/main/java/org/apache/cxf/common/i18n/BundleUtils.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/cxf/common/i18n/BundleUtils.java b/core/src/main/java/org/apache/cxf/common/i18n/BundleUtils.java
index 9aedfcf..5fdd3b4 100644
--- a/core/src/main/java/org/apache/cxf/common/i18n/BundleUtils.java
+++ b/core/src/main/java/org/apache/cxf/common/i18n/BundleUtils.java
@@ -77,13 +77,21 @@ public final class BundleUtils {
public static ResourceBundle getBundle(Class<?> cls) {
try {
+ ClassLoader loader = cls.getClassLoader();
+ if (loader == null) {
+ return ResourceBundle.getBundle(getBundleName(cls), Locale.getDefault());
+ }
return ResourceBundle.getBundle(getBundleName(cls),
Locale.getDefault(),
- cls.getClassLoader());
+ loader);
} catch (MissingResourceException ex) {
+ ClassLoader loader = Thread.currentThread().getContextClassLoader();
+ if (loader == null) {
+ return ResourceBundle.getBundle(getBundleName(cls), Locale.getDefault());
+ }
return ResourceBundle.getBundle(getBundleName(cls),
Locale.getDefault(),
- Thread.currentThread().getContextClassLoader());
+ loader);
}
}
@@ -98,13 +106,21 @@ public final class BundleUtils {
*/
public static ResourceBundle getBundle(Class<?> cls, String name) {
try {
+ ClassLoader loader = cls.getClassLoader();
+ if (loader == null) {
+ return ResourceBundle.getBundle(getBundleName(cls, name), Locale.getDefault());
+ }
return ResourceBundle.getBundle(getBundleName(cls, name),
Locale.getDefault(),
- cls.getClassLoader());
+ loader);
} catch (MissingResourceException ex) {
+ ClassLoader loader = Thread.currentThread().getContextClassLoader();
+ if (loader == null) {
+ return ResourceBundle.getBundle(getBundleName(cls, name), Locale.getDefault());
+ }
return ResourceBundle.getBundle(getBundleName(cls, name),
Locale.getDefault(),
- Thread.currentThread().getContextClassLoader());
+ loader);
}
}
[7/9] cxf git commit: [CXF-4209] Server side message redelivery
support for WS-RM
Posted by dk...@apache.org.
[CXF-4209] Server side message redelivery support for WS-RM
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/0dd29509
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/0dd29509
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/0dd29509
Branch: refs/heads/3.1.x-fixes
Commit: 0dd29509e42fc412ec0cf214e66885d26da9850e
Parents: 661c0bd
Author: Kai Rommel <ka...@sap.com>
Authored: Tue Jun 7 17:47:58 2016 +0200
Committer: Daniel Kulp <dk...@apache.org>
Committed: Fri Jul 22 12:13:14 2016 -0400
----------------------------------------------------------------------
.../java/org/apache/cxf/ws/rm/Destination.java | 23 +
.../apache/cxf/ws/rm/DestinationSequence.java | 100 ++-
.../org/apache/cxf/ws/rm/ManagedRMEndpoint.java | 192 ++---
.../cxf/ws/rm/RMCaptureInInterceptor.java | 227 +++++-
.../cxf/ws/rm/RMCaptureOutInterceptor.java | 1 +
.../apache/cxf/ws/rm/RMDeliveryInterceptor.java | 12 +
.../java/org/apache/cxf/ws/rm/RMEndpoint.java | 3 +
.../org/apache/cxf/ws/rm/RMInInterceptor.java | 74 +-
.../java/org/apache/cxf/ws/rm/RMManager.java | 76 +-
.../java/org/apache/cxf/ws/rm/RMProperties.java | 18 +
.../org/apache/cxf/ws/rm/RedeliveryQueue.java | 106 +++
.../main/java/org/apache/cxf/ws/rm/Servant.java | 2 +-
.../org/apache/cxf/ws/rm/feature/RMFeature.java | 4 +-
.../apache/cxf/ws/rm/persistence/RMMessage.java | 9 +
.../cxf/ws/rm/persistence/jdbc/RMTxStore.java | 74 +-
.../cxf/ws/rm/soap/RMSoapInInterceptor.java | 5 +-
.../cxf/ws/rm/soap/RedeliveryQueueImpl.java | 699 +++++++++++++++++++
.../configuration/wsrm-manager-types.xsd | 15 +
.../schemas/configuration/wsrm-policy.xjb | 9 +
.../cxf/ws/rm/DestinationSequenceTest.java | 4 +-
.../org/apache/cxf/ws/rm/RMEndpointTest.java | 4 +
.../apache/cxf/ws/rm/RMInInterceptorTest.java | 25 +
.../org/apache/cxf/ws/rm/RMManagerTest.java | 43 +-
.../rm/persistence/jdbc/RMTxStoreTestBase.java | 4 +
.../cxf/systest/ws/rm/RedeliveryTest.java | 186 +++++
25 files changed, 1703 insertions(+), 212 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java
index 178a63c..3d3489b 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/Destination.java
@@ -29,6 +29,7 @@ import java.util.logging.Logger;
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.endpoint.Endpoint;
import org.apache.cxf.helpers.CastUtils;
+import org.apache.cxf.io.CachedOutputStream;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageImpl;
@@ -75,6 +76,14 @@ public class Destination extends AbstractEndpoint {
processingSequenceCount.incrementAndGet();
}
+ // this method ensures to keep the sequence until all the messages are delivered
+ public void terminateSequence(DestinationSequence seq) {
+ seq.terminate();
+ if (seq.allAcknowledgedMessagesDelivered()) {
+ removeSequence(seq);
+ }
+ }
+
public void removeSequence(DestinationSequence seq) {
DestinationSequence o;
o = map.remove(seq.getIdentifier().getValue());
@@ -208,6 +217,20 @@ public class Destination extends AbstractEndpoint {
long mn = sequenceType.getMessageNumber().longValue();
seq.processingComplete(mn);
seq.purgeAcknowledged(mn);
+ // remove acknowledged undelivered message
+ seq.removeDeliveringMessageNumber(mn);
+ if (seq.isTerminated() && seq.allAcknowledgedMessagesDelivered()) {
+ removeSequence(seq);
+ }
+ }
+ CachedOutputStream saved = (CachedOutputStream)message.remove(RMMessageConstants.SAVED_CONTENT);
+ if (saved != null) {
+ saved.releaseTempFileHold();
+ try {
+ saved.close();
+ } catch (IOException e) {
+ // ignore
+ }
}
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
index 58b7906..3442fc5 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/DestinationSequence.java
@@ -19,6 +19,8 @@
package org.apache.cxf.ws.rm;
+import java.io.IOException;
+import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -33,12 +35,15 @@ import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.continuations.Continuation;
import org.apache.cxf.continuations.ContinuationProvider;
import org.apache.cxf.continuations.SuspendedInvocationException;
+import org.apache.cxf.interceptor.Fault;
import org.apache.cxf.io.CachedOutputStream;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageUtils;
import org.apache.cxf.ws.addressing.EndpointReferenceType;
+import org.apache.cxf.ws.policy.PolicyVerificationInInterceptor;
import org.apache.cxf.ws.rm.RMConfiguration.DeliveryAssurance;
import org.apache.cxf.ws.rm.manager.AcksPolicyType;
+import org.apache.cxf.ws.rm.persistence.PersistenceUtils;
import org.apache.cxf.ws.rm.persistence.RMMessage;
import org.apache.cxf.ws.rm.persistence.RMStore;
import org.apache.cxf.ws.rm.v200702.Identifier;
@@ -55,6 +60,7 @@ public class DestinationSequence extends AbstractSequence {
private long lastMessageNumber;
private SequenceMonitor monitor;
private boolean acknowledgeOnNextOccasion;
+ private boolean terminated;
private List<DeferredAcknowledgment> deferredAcknowledgments;
private SequenceTermination scheduledTermination;
private String correlationID;
@@ -62,18 +68,25 @@ public class DestinationSequence extends AbstractSequence {
private volatile long highNumberCompleted;
private long nextInOrder;
private List<Continuation> continuations = new LinkedList<Continuation>();
+ // this map is used for robust and redelivery tracking. for redelivery it holds the beingDeliverd messages
private Set<Long> deliveringMessageNumbers = new HashSet<Long>();
public DestinationSequence(Identifier i, EndpointReferenceType a, Destination d, ProtocolVariation pv) {
- this(i, a, 0, null, pv);
+ this(i, a, 0, false, null, pv);
destination = d;
}
public DestinationSequence(Identifier i, EndpointReferenceType a,
long lmn, SequenceAcknowledgement ac, ProtocolVariation pv) {
+ this(i, a, lmn, false, ac, pv);
+ }
+
+ public DestinationSequence(Identifier i, EndpointReferenceType a,
+ long lmn, boolean t, SequenceAcknowledgement ac, ProtocolVariation pv) {
super(i, pv);
acksTo = a;
lastMessageNumber = lmn;
+ terminated = t;
acknowledgement = ac;
if (null == acknowledgement) {
acknowledgement = new SequenceAcknowledgement();
@@ -122,6 +135,7 @@ public class DestinationSequence extends AbstractSequence {
}
monitor.acknowledgeMessage();
+ boolean updated = false;
synchronized (this) {
boolean done = false;
@@ -136,11 +150,13 @@ public class DestinationSequence extends AbstractSequence {
long diff = r.getLower() - messageNumber;
if (diff == 1) {
r.setLower(messageNumber);
+ updated = true;
done = true;
} else if (diff > 0) {
break;
} else if (messageNumber - r.getUpper().longValue() == 1) {
r.setUpper(messageNumber);
+ updated = true;
done = true;
break;
}
@@ -152,6 +168,7 @@ public class DestinationSequence extends AbstractSequence {
AcknowledgementRange range = new AcknowledgementRange();
range.setLower(messageNumber);
range.setUpper(messageNumber);
+ updated = true;
acknowledgement.getAcknowledgementRange().add(i, range);
if (acknowledgement.getAcknowledgementRange().size() > 1) {
@@ -163,18 +180,45 @@ public class DestinationSequence extends AbstractSequence {
mergeRanges();
}
- RMStore store = destination.getManager().getStore();
- if (null != store) {
- RMMessage msg = null;
- if (!MessageUtils.isTrue(message.getContextualProperty(Message.ROBUST_ONEWAY))) {
- msg = new RMMessage();
- CachedOutputStream cos = (CachedOutputStream)message.get(RMMessageConstants.SAVED_CONTENT);
- msg.setContent(cos);
- msg.setContentType((String) message.get(Message.CONTENT_TYPE));
- msg.setMessageNumber(st.getMessageNumber());
+ if (updated) {
+ RMStore store = destination.getManager().getStore();
+ if (null != store) {
+ // only save message, when policy verification is successful
+ // otherwise msgs will be stored and redelivered which do not pass initial verification
+ // as interceptor is called in a later phase than the capturing
+ PolicyVerificationInInterceptor intercep = new PolicyVerificationInInterceptor();
+ boolean policiesVerified = false;
+ try {
+ intercep.handleMessage(message);
+ policiesVerified = true;
+ } catch (Fault e) {
+ // Ignore
+ }
+ RMMessage msg = null;
+ if (policiesVerified
+ && !MessageUtils.isTrue(message.getContextualProperty(Message.ROBUST_ONEWAY))) {
+ try {
+ msg = new RMMessage();
+ CachedOutputStream cos = (CachedOutputStream)message
+ .get(RMMessageConstants.SAVED_CONTENT);
+ msg.setMessageNumber(st.getMessageNumber());
+ msg.setCreatedTime(rmps.getCreatedTime());
+ // in case no attachments are available, cos can be saved directly
+ if (message.getAttachments() == null) {
+ msg.setContent(cos);
+ msg.setContentType((String)message.get(Message.CONTENT_TYPE));
+ } else {
+ InputStream is = cos.getInputStream();
+ PersistenceUtils.encodeRMContent(msg, message, is);
+ }
+ store.persistIncoming(this, msg);
+ } catch (IOException e) {
+ throw new Fault(e);
+ }
+ }
}
- store.persistIncoming(this, msg);
}
+ deliveringMessageNumbers.add(messageNumber);
RMEndpoint reliableEndpoint = destination.getReliableEndpoint();
RMConfiguration cfg = reliableEndpoint.getConfiguration();
@@ -277,7 +321,7 @@ public class DestinationSequence extends AbstractSequence {
return false;
}
if (robustDelivering) {
- deliveringMessageNumbers.add(mn);
+ addDeliveringMessageNumber(mn);
}
if (config.isInOrder()) {
return waitInQueue(mn, canSkip, message, cont);
@@ -286,9 +330,21 @@ public class DestinationSequence extends AbstractSequence {
}
void removeDeliveringMessageNumber(long mn) {
- deliveringMessageNumbers.remove(mn);
+ synchronized (deliveringMessageNumbers) {
+ deliveringMessageNumbers.remove(mn);
+ }
+ }
+ void addDeliveringMessageNumber(long mn) {
+ synchronized (deliveringMessageNumbers) {
+ deliveringMessageNumbers.add(mn);
+ }
}
+ // this method is only used for redelivery
+ boolean allAcknowledgedMessagesDelivered() {
+ return deliveringMessageNumbers.isEmpty();
+ }
+
private Continuation getContinuation(Message message) {
if (message == null) {
return null;
@@ -496,6 +552,22 @@ public class DestinationSequence extends AbstractSequence {
}
}
}
+
+ void terminate() {
+ if (!terminated) {
+ terminated = true;
+ RMStore store = destination.getManager().getStore();
+ if (null == store) {
+ return;
+ }
+ // only updating the sequence
+ store.persistIncoming(this, null);
+ }
+ }
+
+ public boolean isTerminated() {
+ return terminated;
+ }
final class SequenceTermination extends TimerTask {
@@ -521,7 +593,7 @@ public class DestinationSequence extends AbstractSequence {
LogUtils.log(LOG, Level.WARNING, "TERMINATING_INACTIVE_SEQ_MSG",
DestinationSequence.this.getIdentifier().getValue());
- DestinationSequence.this.destination.removeSequence(DestinationSequence.this);
+ DestinationSequence.this.destination.terminateSequence(DestinationSequence.this);
} else {
// reschedule
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ManagedRMEndpoint.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ManagedRMEndpoint.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ManagedRMEndpoint.java
index b96361d..c14bd84 100755
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ManagedRMEndpoint.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/ManagedRMEndpoint.java
@@ -136,8 +136,7 @@ public class ManagedRMEndpoint implements ManagedComponent {
if (outbound) {
return endpoint.getManager().getRetransmissionQueue().countUnacknowledged();
} else {
-// return endpoint.getManager().getRedeliveryQueue().countUndelivered();
- return 0;
+ return endpoint.getManager().getRedeliveryQueue().countUndelivered();
}
}
@@ -155,12 +154,11 @@ public class ManagedRMEndpoint implements ManagedComponent {
}
return manager.getRetransmissionQueue().countUnacknowledged(ss);
} else {
-// DestinationSequence ds = getDestinationSeq(sid);
-// if (null == ds) {
-// throw new IllegalArgumentException("no sequence");
-// }
-// return manager.getRedeliveryQueue().countUndelivered(ds);
- return 0;
+ DestinationSequence ds = getDestinationSeq(sid);
+ if (null == ds) {
+ throw new IllegalArgumentException("no sequence");
+ }
+ return manager.getRedeliveryQueue().countUndelivered(ds);
}
}
@@ -280,55 +278,55 @@ public class ManagedRMEndpoint implements ManagedComponent {
return rsps;
}
-// @ManagedOperation(description = "Redelivery Status")
-// @ManagedOperationParameters({
-// @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier"),
-// @ManagedOperationParameter(name = "messageNumber", description = "The message number")
-// })
-// public CompositeData getRedeliveryStatus(String sid, long num) throws JMException {
-// DestinationSequence ds = getDestinationSeq(sid);
-// if (null == ds) {
-// throw new IllegalArgumentException("no sequence");
-// }
-// RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
-// RetryStatus rs = rq.getRedeliveryStatus(ds, num);
-// return getRetryStatusProperties(num, rs);
-// }
-
-// @ManagedOperation(description = "Redelivery Statuses")
-// @ManagedOperationParameters({
-// @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier")
-// })
-// public CompositeData[] getRedeliveryStatuses(String sid) throws JMException {
-// DestinationSequence ds = getDestinationSeq(sid);
-// if (null == ds) {
-// throw new IllegalArgumentException("no sequence");
-// }
-// RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
-// Map<Long, RetryStatus> rsmap = rq.getRedeliveryStatuses(ds);
-//
-// CompositeData[] rsps = new CompositeData[rsmap.size()];
-// int i = 0;
-// for (Map.Entry<Long, RetryStatus> rs : rsmap.entrySet()) {
-// rsps[i++] = getRetryStatusProperties(rs.getKey(), rs.getValue());
-// }
-// return rsps;
-// }
-
-// @ManagedOperation(description = "List of UnDelivered Message Numbers")
-// @ManagedOperationParameters({
-// @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier")
-// })
-// public Long[] getUnDeliveredMessageIdentifiers(String sid) {
-// RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
-// DestinationSequence ds = getDestinationSeq(sid);
-// if (null == ds) {
-// throw new IllegalArgumentException("no sequence");
-// }
-//
-// List<Long> numbers = rq.getUndeliveredMessageNumbers(ds);
-// return numbers.toArray(new Long[numbers.size()]);
-// }
+ @ManagedOperation(description = "Redelivery Status")
+ @ManagedOperationParameters({
+ @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier"),
+ @ManagedOperationParameter(name = "messageNumber", description = "The message number")
+ })
+ public CompositeData getRedeliveryStatus(String sid, long num) throws JMException {
+ DestinationSequence ds = getDestinationSeq(sid);
+ if (null == ds) {
+ throw new IllegalArgumentException("no sequence");
+ }
+ RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
+ RetryStatus rs = rq.getRedeliveryStatus(ds, num);
+ return getRetryStatusProperties(num, rs);
+ }
+
+ @ManagedOperation(description = "Redelivery Statuses")
+ @ManagedOperationParameters({
+ @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier")
+ })
+ public CompositeData[] getRedeliveryStatuses(String sid) throws JMException {
+ DestinationSequence ds = getDestinationSeq(sid);
+ if (null == ds) {
+ throw new IllegalArgumentException("no sequence");
+ }
+ RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
+ Map<Long, RetryStatus> rsmap = rq.getRedeliveryStatuses(ds);
+
+ CompositeData[] rsps = new CompositeData[rsmap.size()];
+ int i = 0;
+ for (Map.Entry<Long, RetryStatus> rs : rsmap.entrySet()) {
+ rsps[i++] = getRetryStatusProperties(rs.getKey(), rs.getValue());
+ }
+ return rsps;
+ }
+
+ @ManagedOperation(description = "List of UnDelivered Message Numbers")
+ @ManagedOperationParameters({
+ @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier")
+ })
+ public Long[] getUnDeliveredMessageIdentifiers(String sid) {
+ RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
+ DestinationSequence ds = getDestinationSeq(sid);
+ if (null == ds) {
+ throw new IllegalArgumentException("no sequence");
+ }
+
+ List<Long> numbers = rq.getUndeliveredMessageNumbers(ds);
+ return numbers.toArray(new Long[numbers.size()]);
+ }
@ManagedOperation(description = "List of Source Sequence IDs")
@ManagedOperationParameters({
@@ -383,31 +381,31 @@ public class ManagedRMEndpoint implements ManagedComponent {
rq.resume(ss);
}
-// @ManagedOperation(description = "Suspend Redelivery Queue")
-// @ManagedOperationParameters({
-// @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier")
-// })
-// public void suspendDestinationQueue(String sid) throws JMException {
-// DestinationSequence ds = getDestinationSeq(sid);
-// if (null == ds) {
-// throw new IllegalArgumentException("no sequence");
-// }
-// RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
-// rq.suspend(ds);
-// }
-
-// @ManagedOperation(description = "Resume Redelivery Queue")
-// @ManagedOperationParameters({
-// @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier")
-// })
-// public void resumeDestinationQueue(String sid) throws JMException {
-// DestinationSequence ds = getDestinationSeq(sid);
-// if (null == ds) {
-// throw new JMException("no source sequence");
-// }
-// RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
-// rq.resume(ds);
-// }
+ @ManagedOperation(description = "Suspend Redelivery Queue")
+ @ManagedOperationParameters({
+ @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier")
+ })
+ public void suspendDestinationQueue(String sid) throws JMException {
+ DestinationSequence ds = getDestinationSeq(sid);
+ if (null == ds) {
+ throw new IllegalArgumentException("no sequence");
+ }
+ RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
+ rq.suspend(ds);
+ }
+
+ @ManagedOperation(description = "Resume Redelivery Queue")
+ @ManagedOperationParameters({
+ @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier")
+ })
+ public void resumeDestinationQueue(String sid) throws JMException {
+ DestinationSequence ds = getDestinationSeq(sid);
+ if (null == ds) {
+ throw new JMException("no source sequence");
+ }
+ RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
+ rq.resume(ds);
+ }
@ManagedOperation(description = "Current Source Sequence Properties")
public CompositeData getCurrentSourceSequence() throws JMException {
@@ -572,10 +570,13 @@ public class ManagedRMEndpoint implements ManagedComponent {
public void removeDestinationSequence(String sid) throws JMException {
DestinationSequence ds = getDestinationSeq(sid);
if (null == ds) {
- throw new JMException("no source sequence");
+ throw new JMException("no destination sequence");
}
-// RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
-// rq.suspend(ds);
+ RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
+ if (rq.countUndelivered(ds) > 0) {
+ throw new JMException("sequence not empty");
+ }
+ rq.stop(ds);
ds.getDestination().removeSequence(ds);
}
@@ -591,6 +592,19 @@ public class ManagedRMEndpoint implements ManagedComponent {
RetransmissionQueue rq = endpoint.getManager().getRetransmissionQueue();
rq.purgeAll(ss);
}
+
+ @ManagedOperation(description = "Purge UnDelivered Messages")
+ @ManagedOperationParameters({
+ @ManagedOperationParameter(name = "sequenceId", description = "The sequence identifier")
+ })
+ public void purgeUnDeliverededMessages(String sid) {
+ DestinationSequence ds = getDestinationSeq(sid);
+ if (null == ds) {
+ throw new IllegalArgumentException("no sequence");
+ }
+ RedeliveryQueue rq = endpoint.getManager().getRedeliveryQueue();
+ rq.purgeAll(ds);
+ }
private static String getAddressValue(EndpointReferenceType epr) {
if (null != epr && null != epr.getAddress()) {
@@ -675,10 +689,10 @@ public class ManagedRMEndpoint implements ManagedComponent {
// return endpoint.getManager().countCompleted();
// }
-// @ManagedAttribute(description = "Number of Inbound Queued Messages", currencyTimeLimit = 10)
-// public int getQueuedMessagesInboundCount() {
-// return endpoint.getManager().getRedeliveryQueue().countUndelivered();
-// }
+ @ManagedAttribute(description = "Number of Inbound Queued Messages", currencyTimeLimit = 10)
+ public int getQueuedMessagesInboundCount() {
+ return endpoint.getManager().getRedeliveryQueue().countUndelivered();
+ }
// @ManagedAttribute(description = "Number of Inbound Completed Messages", currencyTimeLimit = 10)
// public int getCompletedMessagesInboundCount() {
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java
index 9d48cbc..87ae210 100755
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureInInterceptor.java
@@ -19,51 +19,242 @@
package org.apache.cxf.ws.rm;
+import java.io.Closeable;
+import java.io.IOException;
import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.logging.Logger;
+import javax.xml.soap.SOAPException;
+import javax.xml.soap.SOAPMessage;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.XMLStreamReader;
+import javax.xml.stream.XMLStreamWriter;
+import javax.xml.transform.stream.StreamSource;
+
import org.apache.cxf.common.logging.LogUtils;
-import org.apache.cxf.helpers.IOUtils;
import org.apache.cxf.interceptor.Fault;
+import org.apache.cxf.interceptor.StaxInInterceptor;
import org.apache.cxf.io.CachedOutputStream;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageUtils;
+import org.apache.cxf.phase.AbstractPhaseInterceptor;
import org.apache.cxf.phase.Phase;
+import org.apache.cxf.staxutils.StaxUtils;
+import org.apache.cxf.staxutils.transform.OutTransformWriter;
+import org.apache.cxf.ws.addressing.AddressingProperties;
/**
*
*/
public class RMCaptureInInterceptor extends AbstractRMInterceptor<Message> {
+
private static final Logger LOG = LogUtils.getLogger(RMCaptureInInterceptor.class);
-
+
public RMCaptureInInterceptor() {
- super(Phase.PRE_STREAM);
+ super(Phase.POST_STREAM);
+ addAfter(StaxInInterceptor.class.getName());
}
+ @Override
protected void handle(Message message) throws SequenceFault, RMException {
- LOG.entering(getClass().getName(), "handleMessage");
- // This message capturing mechanism will need to be changed at some point.
- // Until then, we keep this interceptor here and utilize the robust
- // option to avoid the unnecessary message capturing/caching.
- if (!MessageUtils.isTrue(message.getContextualProperty(Message.ROBUST_ONEWAY))) {
- InputStream is = message.getContent(InputStream.class);
- if (is != null) {
+
+ // all messages are initially captured as they cannot be distinguished at this phase
+ // Non application messages temp files are released (cos.releaseTempFileHold()) in RMInInterceptor
+ if (!MessageUtils.isTrue(message.getContextualProperty(Message.ROBUST_ONEWAY))
+ && (getManager().getStore() != null || (getManager().getDestinationPolicy() != null && getManager()
+ .getDestinationPolicy().getRetryPolicy() != null))) {
+
+ message.getInterceptorChain().add(new RMCaptureInEnd());
+ XMLStreamReader reader = message.getContent(XMLStreamReader.class);
+
+ if (null != reader) {
CachedOutputStream saved = new CachedOutputStream();
+ // REVISIT check factory for READER
try {
- IOUtils.copy(is, saved);
-
+ StaxUtils.copy(reader, saved);
saved.flush();
- is.close();
- saved.lockOutputStream();
-
+ saved.holdTempFile();
+ reader.close();
+ LOG.fine("Create new XMLStreamReader");
+ InputStream is = saved.getInputStream();
+ // keep References to clean-up tmp files in RMDeliveryInterceptor
+ setCloseable(message, saved, is);
+ XMLStreamReader newReader = StaxUtils.createXMLStreamReader(is);
+ StaxUtils.configureReader(reader, message);
+ message.setContent(XMLStreamReader.class, newReader);
LOG.fine("Capturing the original RM message");
- //RewindableInputStream ris = RewindableInputStream.makeRewindable(saved.getInputStream());
- message.setContent(InputStream.class, saved.getInputStream());
message.put(RMMessageConstants.SAVED_CONTENT, saved);
- } catch (Exception e) {
+ } catch (XMLStreamException | IOException e) {
throw new Fault(e);
}
+ } else {
+ org.apache.cxf.common.i18n.Message msg = new org.apache.cxf.common.i18n.Message(
+ "No message found for redeliver", LOG, Collections.<String> emptyList());
+ RMException ex = new RMException(msg);
+ throw new Fault(ex);
+ }
+ }
+ }
+
+ private boolean isApplicationMessage(Message message) {
+ final AddressingProperties maps = RMContextUtils.retrieveMAPs(message, false, false);
+ if (null != maps && null != maps.getAction()) {
+ return !RMContextUtils.isRMProtocolMessage(maps.getAction().getValue());
+ }
+ return false;
+ }
+
+ private void setCloseable(Message message, CachedOutputStream cos, InputStream is) {
+ message.put("org.apache.cxf.ws.rm.content.closeable", new Closeable() {
+ @Override
+ public void close() throws IOException {
+ try {
+ is.close();
+ } catch (IOException e) {
+ // Ignore
+ }
+ try {
+ cos.close();
+ } catch (IOException e) {
+ // Ignore
+ }
+ }
+ });
+ }
+
+ /**
+ * RMCaptureInEnd interceptor is used to switch saved_content, in case WSS is activated.
+ */
+ private class RMCaptureInEnd extends AbstractPhaseInterceptor<Message> {
+ RMCaptureInEnd() {
+ super(Phase.PRE_LOGICAL);
+ addBefore(RMInInterceptor.class.getName());
+ }
+
+ @Override
+ public void handleFault(Message message) {
+ // in case of a SequenceFault SAVED_CONTENT must be released
+ Exception ex = message.getContent(Exception.class);
+ if (ex instanceof SequenceFault) {
+ Closeable closable = (Closeable)message.get("org.apache.cxf.ws.rm.content.closeable");
+ if (null != closable) {
+ try {
+ closable.close();
+ } catch (IOException e) {
+ // Ignore
+ }
+ }
+ CachedOutputStream saved = (CachedOutputStream)message.get(RMMessageConstants.SAVED_CONTENT);
+ if (saved != null) {
+ saved.releaseTempFileHold();
+ try {
+ saved.close();
+ } catch (IOException e) {
+ // ignore
+ }
+ }
+ }
+ }
+
+ public void handleMessage(Message message) {
+ LOG.entering(getClass().getName(), "handleMessage");
+ // Capturing the soap envelope. In case of WSS was activated, decrypted envelope is captured.
+ if (!MessageUtils.isTrue(message.getContextualProperty(Message.ROBUST_ONEWAY))
+ && isApplicationMessage(message)
+ && (getManager().getStore() != null || (getManager().getDestinationPolicy() != null && getManager()
+ .getDestinationPolicy().getRetryPolicy() != null))) {
+
+ CachedOutputStream saved = new CachedOutputStream();
+ SOAPMessage soapMessage = message.getContent(SOAPMessage.class);
+
+ if (soapMessage != null) {
+ try {
+ javax.xml.transform.Source envelope = soapMessage.getSOAPPart().getContent();
+ StaxUtils.copy(envelope, saved);
+ saved.flush();
+ // create a new source part from cos
+ InputStream is = saved.getInputStream();
+ // close old saved content
+ closeOldSavedContent(message);
+ // keep References to clean-up tmp files in RMDeliveryInterceptor
+ setCloseable(message, saved, is);
+ StreamSource source = new StreamSource(is);
+ soapMessage.getSOAPPart().setContent(source);
+ // when WSS was activated, saved content still contains soap headers to be removed
+ message.put(RMMessageConstants.SAVED_CONTENT, removeUnnecessarySoapHeaders(saved));
+ } catch (SOAPException | IOException | XMLStreamException e) {
+ throw new Fault(e);
+ }
+ }
+ }
+ }
+
+ private void closeOldSavedContent(Message message) {
+ CachedOutputStream saved = (CachedOutputStream)message.get(RMMessageConstants.SAVED_CONTENT);
+ if (saved != null) {
+ saved.releaseTempFileHold();
+ try {
+ saved.close();
+ } catch (IOException e) {
+ // ignore
+ }
+ }
+ Closeable closable = (Closeable)message.get("org.apache.cxf.ws.rm.content.closeable");
+ if (null != closable) {
+ try {
+ closable.close();
+ } catch (IOException e) {
+ // Ignore
+ }
+ }
+ }
+
+ private CachedOutputStream removeUnnecessarySoapHeaders(CachedOutputStream saved) {
+ CachedOutputStream newSaved = new CachedOutputStream();
+
+ InputStream is = null;
+ try {
+ is = saved.getInputStream();
+ XMLStreamWriter capture = StaxUtils.createXMLStreamWriter(newSaved,
+ StandardCharsets.UTF_8.name());
+ Map<String, String> map = new HashMap<String, String>();
+ map.put("{http://schemas.xmlsoap.org/ws/2005/02/rm}Sequence", "");
+ map.put("{http://schemas.xmlsoap.org/ws/2005/02/rm}SequenceAcknowledgement", "");
+ map.put("{http://docs.oasis-open.org/ws-rx/wsrm/200702}Sequence", "");
+ map.put("{http://docs.oasis-open.org/ws-rx/wsrm/200702}SequenceAcknowledgement", "");
+ map.put("{http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd}Security",
+ "");
+ // attributes to be removed
+ Map<String, String> amap = new HashMap<String, String>();
+ amap.put("{http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd}Id",
+ "");
+
+ capture = new OutTransformWriter(capture, map, Collections.<String, String> emptyMap(),
+ Collections.<String> emptyList(), amap, false, null);
+ StaxUtils.copy(new StreamSource(is), capture);
+ capture.flush();
+ capture.close();
+ newSaved.flush();
+ // hold temp file, otherwise it will be deleted in case msg was written to RMTxStore
+ // or resend was executed
+ newSaved.holdTempFile();
+ is.close();
+ } catch (IOException | XMLStreamException e) {
+ throw new Fault(e);
+ } finally {
+ if (null != is) {
+ try {
+ is.close();
+ } catch (IOException e) {
+ // Ignore
+ }
+ }
}
+ return newSaved;
}
}
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
index b3c412d..af682c3 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMCaptureOutInterceptor.java
@@ -291,6 +291,7 @@ public class RMCaptureOutInterceptor extends AbstractRMInterceptor<Message> {
}
// serializes the message content and the attachments into
// the RMMessage content
+ msg.setCreatedTime(rmps.getCreatedTime());
PersistenceUtils.encodeRMContent(msg, message, is);
store.persistOutgoing(ss, msg);
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java
index 262e066..ff34c95 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMDeliveryInterceptor.java
@@ -19,6 +19,8 @@
package org.apache.cxf.ws.rm;
+import java.io.Closeable;
+import java.io.IOException;
import java.util.logging.Logger;
import org.apache.cxf.common.logging.LogUtils;
@@ -60,5 +62,15 @@ public class RMDeliveryInterceptor extends AbstractRMInterceptor<Message> {
dest.acknowledge(message);
}
dest.processingComplete(message);
+
+ // close InputStream of RMCaptureInInterceptor, to delete tmp files in filesystem
+ Closeable closable = (Closeable)message.get("org.apache.cxf.ws.rm.content.closeable");
+ if (null != closable) {
+ try {
+ closable.close();
+ } catch (IOException e) {
+ // Ignore
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/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 e393124..03f6d13 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
@@ -830,6 +830,9 @@ public class RMEndpoint {
for (SourceSequence ss : getSource().getAllSequences()) {
manager.getRetransmissionQueue().stop(ss);
}
+ for (DestinationSequence ds : getDestination().getAllSequences()) {
+ manager.getRedeliveryQueue().stop(ds);
+ }
// unregistering of this managed bean from the server is done by the bus itself
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java
index 5b516e4..c82412b 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMInInterceptor.java
@@ -24,6 +24,7 @@ import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.cxf.common.logging.LogUtils;
+import org.apache.cxf.io.CachedOutputStream;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageUtils;
@@ -32,6 +33,7 @@ import org.apache.cxf.ws.addressing.ContextUtils;
import org.apache.cxf.ws.addressing.MAPAggregator;
import org.apache.cxf.ws.rm.v200702.Identifier;
import org.apache.cxf.ws.rm.v200702.SequenceAcknowledgement;
+import org.apache.cxf.ws.rm.v200702.SequenceType;
import org.apache.cxf.ws.security.trust.STSUtils;
/**
@@ -49,17 +51,21 @@ public class RMInInterceptor extends AbstractRMInterceptor<Message> {
@Override
public void handleFault(Message message) {
message.put(MAPAggregator.class.getName(), true);
- if (RMContextUtils.getProtocolVariation(message) != null
- && MessageUtils.isTrue(message.get(RMMessageConstants.DELIVERING_ROBUST_ONEWAY))) {
- // revert the delivering entry from the destination sequence
- try {
- Destination destination = getManager().getDestination(message);
- if (destination != null) {
- destination.releaseDeliveringStatus(message);
+ if (RMContextUtils.getProtocolVariation(message) != null) {
+ if (MessageUtils.isTrue(message.get(RMMessageConstants.DELIVERING_ROBUST_ONEWAY))) {
+ // revert the delivering entry from the destination sequence
+ try {
+ Destination destination = getManager().getDestination(message);
+ if (destination != null) {
+ destination.releaseDeliveringStatus(message);
+ }
+ } catch (RMException e) {
+ LOG.log(Level.WARNING, "Failed to revert the delivering status");
}
- } catch (RMException e) {
- LOG.log(Level.WARNING, "Failed to revert the delivering status");
- }
+ } else if (isRedeliveryEnabled(message) && RMContextUtils.isServerSide(message)
+ && isApplicationMessage(message) && hasValidSequence(message)) {
+ getManager().getRedeliveryQueue().addUndelivered(message);
+ }
}
// make sure the fault is returned for an ws-rm related fault or an invalid ws-rm message
// note that OneWayProcessingInterceptor handles the robust case, hence not handled here.
@@ -82,6 +88,42 @@ public class RMInInterceptor extends AbstractRMInterceptor<Message> {
|| message.getContent(Exception.class) instanceof SequenceFault);
}
+ private boolean hasValidSequence(Message message) {
+ final RMProperties rmps = RMContextUtils.retrieveRMProperties(message, false);
+ if (rmps != null) {
+ SequenceType st = rmps.getSequence();
+ if (st != null && st.getIdentifier() != null) {
+ try {
+ Destination destination = getManager().getDestination(message);
+ if (destination != null && destination.getSequence(st.getIdentifier()) != null) {
+ return true;
+ }
+ } catch (RMException e) {
+ // fall through
+ }
+ }
+ }
+ return false;
+ }
+
+ private static boolean isApplicationMessage(Message message) {
+ final AddressingProperties maps = RMContextUtils.retrieveMAPs(message, false, false);
+ if (null != maps && null != maps.getAction()) {
+ return !RMContextUtils.isRMProtocolMessage(maps.getAction().getValue());
+ }
+ return false;
+ }
+
+ private boolean isRedeliveryEnabled(Message message) {
+ // deprecated redelivery mode check
+ if (MessageUtils.isTrue(message.getContextualProperty("org.apache.cxf.ws.rm.destination.redeliver"))) {
+ LOG.warning("Use RetryPolicy to enable the redelivery mode");
+ return true;
+ }
+ return getManager().getDestinationPolicy() != null
+ && getManager().getDestinationPolicy().getRetryPolicy() != null;
+ }
+
protected void handle(Message message) throws SequenceFault, RMException {
LOG.entering(getClass().getName(), "handleMessage");
@@ -138,10 +180,6 @@ public class RMInInterceptor extends AbstractRMInterceptor<Message> {
}
RMContextUtils.setProtocolVariation(message, protocol);
- // Destination destination = getManager().getDestination(message);
- // RMEndpoint rme = getManager().getReliableEndpoint(message);
- // Servant servant = new Servant(rme);
-
boolean isApplicationMessage = !RMContextUtils.isRMProtocolMessage(action);
LOG.fine("isApplicationMessage: " + isApplicationMessage);
@@ -164,6 +202,12 @@ public class RMInInterceptor extends AbstractRMInterceptor<Message> {
rme.receivedApplicationMessage();
}
} else {
+ // in case message is not an application message, release SAVED_CONTENT
+ // otherwise tmp files will not be closed
+ CachedOutputStream cos = (CachedOutputStream)message.get(RMMessageConstants.SAVED_CONTENT);
+ if (null != cos) {
+ cos.releaseTempFileHold();
+ }
rme.receivedControlMessage();
if (RM10Constants.SEQUENCE_ACKNOWLEDGMENT_ACTION.equals(action)
|| RM11Constants.SEQUENCE_ACKNOWLEDGMENT_ACTION.equals(action)) {
@@ -215,7 +259,7 @@ public class RMInInterceptor extends AbstractRMInterceptor<Message> {
final boolean robust =
MessageUtils.isTrue(message.getContextualProperty(Message.ROBUST_ONEWAY));
if (robust) {
- // set this property to change the acknlowledging behavior
+ // set this property to change the acknowledging behavior
message.put(RMMessageConstants.DELIVERING_ROBUST_ONEWAY, Boolean.TRUE);
}
destination.acknowledge(message);
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
index 9639cfe..6a0839e 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMManager.java
@@ -53,6 +53,9 @@ import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageImpl;
import org.apache.cxf.phase.PhaseInterceptorChain;
import org.apache.cxf.service.Service;
+import org.apache.cxf.service.model.BindingInfo;
+import org.apache.cxf.service.model.InterfaceInfo;
+import org.apache.cxf.service.model.ServiceInfo;
import org.apache.cxf.transport.Conduit;
import org.apache.cxf.ws.addressing.AddressingProperties;
import org.apache.cxf.ws.addressing.ContextUtils;
@@ -70,6 +73,7 @@ import org.apache.cxf.ws.rm.persistence.PersistenceUtils;
import org.apache.cxf.ws.rm.persistence.RMMessage;
import org.apache.cxf.ws.rm.persistence.RMStore;
import org.apache.cxf.ws.rm.policy.RMPolicyUtilities;
+import org.apache.cxf.ws.rm.soap.RedeliveryQueueImpl;
import org.apache.cxf.ws.rm.soap.RetransmissionQueueImpl;
import org.apache.cxf.ws.rm.soap.SoapFaultFactory;
import org.apache.cxf.ws.rm.v200702.CloseSequenceType;
@@ -112,6 +116,7 @@ public class RMManager {
private RMStore store;
private SequenceIdentifierGenerator idGenerator;
private RetransmissionQueue retransmissionQueue;
+ private RedeliveryQueue redeliveryQueue;
private Map<Endpoint, RMEndpoint> reliableEndpoints = new ConcurrentHashMap<Endpoint, RMEndpoint>();
private AtomicReference<Timer> timer = new AtomicReference<Timer>();
private RMConfiguration configuration;
@@ -185,6 +190,14 @@ public class RMManager {
retransmissionQueue = rq;
}
+ public RedeliveryQueue getRedeliveryQueue() {
+ return redeliveryQueue;
+ }
+
+ public void setRedeliveryQueue(RedeliveryQueue redeliveryQueue) {
+ this.redeliveryQueue = redeliveryQueue;
+ }
+
public SequenceIdentifierGenerator getIdGenerator() {
return idGenerator;
}
@@ -556,10 +569,10 @@ public class RMManager {
}
for (DestinationSequence ds : dss) {
- reconverDestinationSequence(endpoint, conduit, rme.getDestination(), ds);
+ recoverDestinationSequence(endpoint, conduit, rme.getDestination(), ds);
}
retransmissionQueue.start();
-
+ redeliveryQueue.start();
}
private void recoverSourceSequence(Endpoint endpoint, Conduit conduit, Source s,
@@ -570,7 +583,7 @@ public class RMManager {
return;
}
LOG.log(Level.FINE, "Number of messages in sequence: {0}", ms.size());
-
+ // only recover the sequence if there are pending messages
s.addSequence(ss, false);
// choosing an arbitrary valid source sequence as the current source sequence
if (s.getAssociatedSequence(null) == null && !ss.isExpired() && !ss.isLastMessage()) {
@@ -596,6 +609,7 @@ public class RMManager {
st.setMessageNumber(m.getMessageNumber());
RMProperties rmps = new RMProperties();
rmps.setSequence(st);
+ rmps.setCreatedTime(m.getCreatedTime());
rmps.exposeAs(ss.getProtocol().getWSRMNamespace());
if (ss.isLastMessage() && ss.getCurrentMessageNr() == m.getMessageNumber()) {
CloseSequenceType close = new CloseSequenceType();
@@ -623,10 +637,59 @@ public class RMManager {
}
}
- private void reconverDestinationSequence(Endpoint endpoint, Conduit conduit, Destination d,
+ private void recoverDestinationSequence(Endpoint endpoint, Conduit conduit, Destination d,
DestinationSequence ds) {
+ // always recover the sequence
d.addSequence(ds, false);
- //TODO add the redelivery code
+
+ Collection<RMMessage> ms = store.getMessages(ds.getIdentifier(), false);
+ if (null == ms || 0 == ms.size()) {
+ return;
+ }
+ LOG.log(Level.FINE, "Number of messages in sequence: {0}", ms.size());
+
+ for (RMMessage m : ms) {
+ Message message = new MessageImpl();
+ Exchange exchange = new ExchangeImpl();
+ message.setExchange(exchange);
+ if (null != conduit) {
+ exchange.setConduit(conduit);
+ }
+ exchange.put(Endpoint.class, endpoint);
+ exchange.put(Service.class, endpoint.getService());
+ if (endpoint.getEndpointInfo().getService() != null) {
+ exchange.put(ServiceInfo.class, endpoint.getEndpointInfo().getService());
+ exchange.put(InterfaceInfo.class, endpoint.getEndpointInfo().getService().getInterface());
+ }
+ exchange.put(Binding.class, endpoint.getBinding());
+ exchange.put(BindingInfo.class, endpoint.getEndpointInfo().getBinding());
+ exchange.put(Bus.class, bus);
+
+ SequenceType st = new SequenceType();
+ st.setIdentifier(ds.getIdentifier());
+ st.setMessageNumber(m.getMessageNumber());
+ RMProperties rmps = new RMProperties();
+ rmps.setSequence(st);
+ rmps.setCreatedTime(m.getCreatedTime());
+ RMContextUtils.storeRMProperties(message, rmps, false);
+ try {
+ // RMMessage is stored in a serialized way, therefore
+ // RMMessage content must be splitted into soap root message
+ // and attachments
+ PersistenceUtils.decodeRMContent(m, message);
+ redeliveryQueue.addUndelivered(message);
+ // add acknowledged undelivered message
+ ds.addDeliveringMessageNumber(m.getMessageNumber());
+ } catch (IOException e) {
+ LOG.log(Level.SEVERE, "Error reading persisted message data", e);
+ }
+ }
+
+ // if no messages are recovered and the sequence has been already terminated, remove the sequence
+ if (ds.isTerminated() && ds.allAcknowledgedMessagesDelivered()) {
+ d.removeSequence(ds);
+ store.removeDestinationSequence(ds.getIdentifier());
+ }
}
RMEndpoint createReliableEndpoint(final Endpoint endpoint) {
@@ -664,6 +727,9 @@ public class RMManager {
if (null == retransmissionQueue) {
retransmissionQueue = new RetransmissionQueueImpl(this);
}
+ if (null == redeliveryQueue) {
+ redeliveryQueue = new RedeliveryQueueImpl(this);
+ }
if (null == idGenerator) {
idGenerator = new DefaultSequenceIdentifierGenerator();
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMProperties.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMProperties.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMProperties.java
index 448bfc6..f137803 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMProperties.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RMProperties.java
@@ -37,6 +37,7 @@ public class RMProperties {
private CloseSequenceType closeSequence;
private String namespaceURI;
private boolean lastMessage;
+ private long createdTime = System.currentTimeMillis();
public Collection<SequenceAcknowledgement> getAcks() {
return acks;
@@ -120,4 +121,21 @@ public class RMProperties {
public void exposeAs(String uri) {
namespaceURI = uri;
}
+
+ /**
+ * Get the initial creation time of this RM properties instance.
+ * @return Returns the createdTime.
+ */
+ public long getCreatedTime() {
+ return createdTime;
+ }
+
+ /**
+ * Set the initial creation time of this RM properties instance.
+ *
+ * @param createdTime The createdTime to set.
+ */
+ public void setCreatedTime(long createdTime) {
+ this.createdTime = createdTime;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RedeliveryQueue.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RedeliveryQueue.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RedeliveryQueue.java
new file mode 100644
index 0000000..4737859
--- /dev/null
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/RedeliveryQueue.java
@@ -0,0 +1,106 @@
+/**
+ * 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;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.cxf.message.Message;
+
+public interface RedeliveryQueue {
+
+
+ String DEFAULT_BASE_REDELIVERY_INTERVAL = "3000";
+ int DEFAULT_EXPONENTIAL_BACKOFF = 2;
+
+ /**
+ * @param seq the sequence under consideration
+ * @return the number of unacknowledged messages for that sequence
+ */
+ int countUndelivered(DestinationSequence seq);
+
+ /**
+ * @return the total number of undelivered messages in this queue
+ */
+ int countUndelivered();
+
+ /**
+ * @return true if there are no unacknowledged messages in the queue
+ */
+ boolean isEmpty();
+
+ /**
+ * Accepts a failed message for possible future redelivery.
+ * @param message the message context.
+ */
+ void addUndelivered(Message message);
+
+ /**
+ * Purge all candiates for the given sequence.
+ *
+ * @param seq the sequence object
+ */
+ void purgeAll(DestinationSequence seq);
+
+ /**
+ *
+ * @param seq
+ * @return
+ */
+ List<Long> getUndeliveredMessageNumbers(DestinationSequence seq);
+
+ /**
+ * Returns the retransmission status for the specified message.
+ * @param seq
+ * @param num
+ * @return
+ */
+ RetryStatus getRedeliveryStatus(DestinationSequence seq, long num);
+
+ /**
+ * Return the retransmission status of all the messages assigned to the sequence.
+ * @param seq
+ * @return
+ */
+ Map<Long, RetryStatus> getRedeliveryStatuses(DestinationSequence seq);
+
+ /**
+ * Initiate resends.
+ */
+ void start();
+
+ /**
+ * Stops redelivery queue.
+ * @param seq
+ */
+ void stop(DestinationSequence seq);
+
+ /**
+ * Suspends the redelivery attempts for the specified sequence
+ * @param seq
+ */
+ void suspend(DestinationSequence seq);
+
+ /**
+ * Resumes the redelivery attempts for the specified sequence
+ * @param seq
+ */
+ void resume(DestinationSequence seq);
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/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 7cfeade..80a64bc 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
@@ -245,7 +245,7 @@ public class Servant implements Invoker {
return null;
}
- destination.removeSequence(terminatedSeq);
+ destination.terminateSequence(terminatedSeq);
// the following may be necessary if the last message for this sequence was a oneway
// request and hence there was no response to which a last message could have been added
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/feature/RMFeature.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/feature/RMFeature.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/feature/RMFeature.java
index 08cbf2a..90db17b 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/feature/RMFeature.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/feature/RMFeature.java
@@ -123,9 +123,7 @@ public class RMFeature extends AbstractFeature {
provider.getInInterceptors().add(rmLogicalIn);
provider.getInInterceptors().add(rmInCodec);
provider.getInInterceptors().add(rmDelivery);
- if (null != store) {
- provider.getInInterceptors().add(rmCaptureIn);
- }
+ provider.getInInterceptors().add(rmCaptureIn);
provider.getOutInterceptors().add(rmLogicalOut);
provider.getOutInterceptors().add(rmOutCodec);
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMMessage.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMMessage.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMMessage.java
index 348117c..ad61b04 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMMessage.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/RMMessage.java
@@ -32,6 +32,7 @@ public class RMMessage {
private String contentType;
private long messageNumber;
private String to;
+ private long createdTime;
/**
* Returns the message number of the message within its sequence.
@@ -120,4 +121,12 @@ public class RMMessage {
this.contentType = contentType;
}
+ public long getCreatedTime() {
+ return createdTime;
+ }
+
+ public void setCreatedTime(long createdTime) {
+ this.createdTime = createdTime;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java
index 641df46..e3cfa0b 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStore.java
@@ -75,6 +75,7 @@ public class RMTxStore implements RMStore {
{"LAST_MSG_NO", "DECIMAL(19, 0)"},
{"ENDPOINT_ID", "VARCHAR(1024)"},
{"ACKNOWLEDGED", "BLOB"},
+ {"TERMINATED", "CHAR(1)"},
{"PROTOCOL_VERSION", "VARCHAR(256)"}};
private static final String[] DEST_SEQUENCES_TABLE_KEYS = {"SEQ_ID"};
private static final String[][] SRC_SEQUENCES_TABLE_COLS
@@ -90,6 +91,7 @@ public class RMTxStore implements RMStore {
= {{"SEQ_ID", "VARCHAR(256) NOT NULL"},
{"MSG_NO", "DECIMAL(19, 0) NOT NULL"},
{"SEND_TO", "VARCHAR(256)"},
+ {"CREATED_TIME", "DECIMAL(19, 0)"},
{"CONTENT", "BLOB"},
{"CONTENT_TYPE", "VARCHAR(1024)"}};
private static final String[] MESSAGES_TABLE_KEYS = {"SEQ_ID", "MSG_NO"};
@@ -119,27 +121,27 @@ public class RMTxStore implements RMStore {
private static final String DELETE_SRC_SEQUENCE_STMT_STR =
"DELETE FROM CXF_RM_SRC_SEQUENCES WHERE SEQ_ID = ?";
private static final String UPDATE_DEST_SEQUENCE_STMT_STR =
- "UPDATE CXF_RM_DEST_SEQUENCES SET LAST_MSG_NO = ?, ACKNOWLEDGED = ? WHERE SEQ_ID = ?";
+ "UPDATE CXF_RM_DEST_SEQUENCES SET LAST_MSG_NO = ?, TERMINATED = ?, ACKNOWLEDGED = ? WHERE SEQ_ID = ?";
private static final String UPDATE_SRC_SEQUENCE_STMT_STR =
"UPDATE CXF_RM_SRC_SEQUENCES SET CUR_MSG_NO = ?, LAST_MSG = ? WHERE SEQ_ID = ?";
private static final String CREATE_MESSAGE_STMT_STR
- = "INSERT INTO {0} (SEQ_ID, MSG_NO, SEND_TO, CONTENT, CONTENT_TYPE) VALUES(?, ?, ?, ?, ?)";
+ = "INSERT INTO {0} (SEQ_ID, MSG_NO, SEND_TO, CREATED_TIME, CONTENT, CONTENT_TYPE) 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, PROTOCOL_VERSION, ACKNOWLEDGED FROM CXF_RM_DEST_SEQUENCES "
+ "SELECT ACKS_TO, LAST_MSG_NO, PROTOCOL_VERSION, TERMINATED, 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, PROTOCOL_VERSION 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, PROTOCOL_VERSION, ACKNOWLEDGED FROM CXF_RM_DEST_SEQUENCES "
+ "SELECT SEQ_ID, ACKS_TO, LAST_MSG_NO, PROTOCOL_VERSION, TERMINATED, ACKNOWLEDGED FROM CXF_RM_DEST_SEQUENCES "
+ "WHERE ENDPOINT_ID = ?";
private static final String SELECT_SRC_SEQUENCES_STMT_STR =
"SELECT SEQ_ID, CUR_MSG_NO, LAST_MSG, EXPIRY, OFFERING_SEQ_ID, PROTOCOL_VERSION "
+ "FROM CXF_RM_SRC_SEQUENCES WHERE ENDPOINT_ID = ?";
private static final String SELECT_MESSAGES_STMT_STR =
- "SELECT MSG_NO, SEND_TO, CONTENT, CONTENT_TYPE FROM {0} WHERE SEQ_ID = ?";
+ "SELECT MSG_NO, SEND_TO, CREATED_TIME, CONTENT, CONTENT_TYPE FROM {0} WHERE SEQ_ID = ?";
private static final String ALTER_TABLE_STMT_STR =
"ALTER TABLE {0} ADD {1} {2}";
private static final String CREATE_INBOUND_MESSAGE_STMT_STR =
@@ -395,13 +397,14 @@ public class RMTxStore implements RMStore {
EndpointReferenceType acksTo = RMUtils.createReference(res.getString(1));
long lm = res.getLong(2);
ProtocolVariation pv = decodeProtocolVersion(res.getString(3));
- InputStream is = res.getBinaryStream(4);
+ boolean t = res.getBoolean(4);
+ InputStream is = res.getBinaryStream(5);
SequenceAcknowledgement ack = null;
if (null != is) {
ack = PersistenceUtils.getInstance()
.deserialiseAcknowledgment(is);
}
- return new DestinationSequence(sid, acksTo, lm, ack, pv);
+ return new DestinationSequence(sid, acksTo, lm, t, ack, pv);
}
} catch (SQLException ex) {
conex = ex;
@@ -521,13 +524,14 @@ public class RMTxStore implements RMStore {
EndpointReferenceType acksTo = RMUtils.createReference(res.getString(2));
long lm = res.getLong(3);
ProtocolVariation pv = decodeProtocolVersion(res.getString(4));
- InputStream is = res.getBinaryStream(5);
+ boolean t = res.getBoolean(5);
+ InputStream is = res.getBinaryStream(6);
SequenceAcknowledgement ack = null;
if (null != is) {
ack = PersistenceUtils.getInstance()
.deserialiseAcknowledgment(is);
}
- DestinationSequence seq = new DestinationSequence(sid, acksTo, lm, ack, pv);
+ DestinationSequence seq = new DestinationSequence(sid, acksTo, lm, t, ack, pv);
seqs.add(seq);
}
} catch (SQLException ex) {
@@ -596,11 +600,13 @@ public class RMTxStore implements RMStore {
while (res.next()) {
long mn = res.getLong(1);
String to = res.getString(2);
- Blob blob = res.getBlob(3);
- String contentType = res.getString(4);
+ long ct = res.getLong(3);
+ Blob blob = res.getBlob(4);
+ String contentType = res.getString(5);
RMMessage msg = new RMMessage();
msg.setMessageNumber(mn);
msg.setTo(to);
+ msg.setCreatedTime(ct);
CachedOutputStream cos = new CachedOutputStream();
IOUtils.copyAndCloseInput(blob.getBinaryStream(), cos);
cos.flush();
@@ -752,8 +758,9 @@ public class RMTxStore implements RMStore {
stmt.setString(1, id);
stmt.setLong(2, nr);
stmt.setString(3, to);
- stmt.setBinaryStream(4, msgin);
- stmt.setString(5, contentType);
+ stmt.setLong(4, msg.getCreatedTime());
+ stmt.setBinaryStream(5, msgin);
+ stmt.setString(6, contentType);
stmt.execute();
if (LOG.isLoggable(Level.FINE)) {
LOG.log(Level.FINE, "Successfully stored {0} message number {1} for sequence {2}",
@@ -805,10 +812,11 @@ public class RMTxStore implements RMStore {
stmt = getStatement(con, UPDATE_DEST_SEQUENCE_STMT_STR);
long lastMessageNr = seq.getLastMessageNumber();
- stmt.setLong(1, lastMessageNr);
+ stmt.setLong(1, lastMessageNr);
+ stmt.setString(2, seq.isTerminated() ? "1" : "0");
InputStream is = PersistenceUtils.getInstance().serialiseAcknowledgment(seq.getAcknowledgment());
- stmt.setBinaryStream(2, is, is.available());
- stmt.setString(3, seq.getIdentifier().getValue());
+ stmt.setBinaryStream(3, is, is.available());
+ stmt.setString(4, seq.getIdentifier().getValue());
stmt.execute();
} finally {
releaseResources(stmt, null);
@@ -884,12 +892,11 @@ public class RMTxStore implements RMStore {
}
protected void verifyTable(Connection con, String tableName, String[][] tableCols) {
- List<String[]> newCols = new ArrayList<String[]>();
- ResultSet rs = null;
try {
DatabaseMetaData metadata = con.getMetaData();
- rs = metadata.getColumns(null, null, tableName, "%");
+ ResultSet rs = metadata.getColumns(null, null, tableName, "%");
Set<String> dbCols = new HashSet<String>();
+ List<String[]> newCols = new ArrayList<String[]>();
while (rs.next()) {
dbCols.add(rs.getString(4));
}
@@ -898,26 +905,12 @@ public class RMTxStore implements RMStore {
newCols.add(col);
}
}
- } catch (SQLException ex) {
- if (LOG.isLoggable(Level.FINE)) {
- LOG.fine("Table " + tableName + " cannot be verified.");
- }
- } finally {
- if (rs != null) {
- try {
- rs.close();
- } catch (SQLException e) {
- // ignore
+ if (newCols.size() > 0) {
+ // need to add the new columns
+ if (LOG.isLoggable(Level.FINE)) {
+ LOG.log(Level.FINE, "Table " + tableName + " needs additional columns");
}
- }
- }
- if (newCols.size() > 0) {
- // need to add the new columns
- if (LOG.isLoggable(Level.FINE)) {
- LOG.log(Level.FINE, "Table " + tableName + " needs additional columns");
- }
-
- try {
+
for (String[] newCol : newCols) {
Statement st = con.createStatement();
try {
@@ -931,9 +924,9 @@ public class RMTxStore implements RMStore {
st.close();
}
}
- } catch (SQLException ex) {
- LOG.log(Level.WARNING, "Table " + tableName + " cannot be altered.", ex);
}
+ } catch (SQLException ex) {
+ LOG.log(Level.WARNING, "Table " + tableName + " cannot be altered.", ex);
}
}
@@ -965,7 +958,6 @@ public class RMTxStore implements RMStore {
for (int i = 0; i < SET_SCHEMA_STMT_STRS.length; i++) {
try {
stmt.executeUpdate(MessageFormat.format(SET_SCHEMA_STMT_STRS[i], schemaName));
- ex0 = null;
break;
} catch (SQLException ex) {
ex.setNextException(ex0);
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInInterceptor.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInInterceptor.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInInterceptor.java
index 8e1b25e..9215031 100644
--- a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInInterceptor.java
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RMSoapInInterceptor.java
@@ -139,7 +139,10 @@ public class RMSoapInInterceptor extends AbstractSoapInterceptor {
* @return the RM properties
*/
public RMProperties unmarshalRMProperties(SoapMessage message) {
- RMProperties rmps = new RMProperties();
+ RMProperties rmps = (RMProperties)message.get(RMContextUtils.getRMPropertiesKey(false));
+ if (rmps == null) {
+ rmps = new RMProperties();
+ }
List<Header> headers = message.getHeaders();
if (headers != null) {
decodeHeaders(message, headers, rmps);
[4/9] cxf git commit: Recording .gitmergeinfo Changes
Posted by dk...@apache.org.
Recording .gitmergeinfo Changes
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/573fffce
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/573fffce
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/573fffce
Branch: refs/heads/3.1.x-fixes
Commit: 573fffce8b927738020cce6ec66b4ee11c9c80a8
Parents: 0f66cb0
Author: Daniel Kulp <dk...@apache.org>
Authored: Fri Jul 22 11:17:21 2016 -0400
Committer: Daniel Kulp <dk...@apache.org>
Committed: Fri Jul 22 12:13:14 2016 -0400
----------------------------------------------------------------------
.gitmergeinfo | 12 ++++++++++++
1 file changed, 12 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/573fffce/.gitmergeinfo
----------------------------------------------------------------------
diff --git a/.gitmergeinfo b/.gitmergeinfo
index f028a7b..90b4b68 100644
--- a/.gitmergeinfo
+++ b/.gitmergeinfo
@@ -35,6 +35,7 @@ B 671b1cfce0c423a1e9e8392fdd1750697fa968df
B 68f110e5bcfe26d050bd05988ecd9c863e6e8742
B 699f4c03de1acb1cf9eb95ad2db792dba5a56ead
B 6c324421a358919b4ecb3cfdd8a2e0e16ff81724
+B 6df5deab24016d07e4099712b07535688d164d94
B 6e7b79f6a8b6523b2248937275d92918e5682aa2
B 77c3f51dcb1f02c8799f4e09676c0697e76a0756
B 7b7629682d15345518e66d46e575bf1ac334cf00
@@ -88,25 +89,31 @@ B f98785bd8490c4717353f1a9688cae3e7a823ec2
B fb30f8bffc85fcc3208fcc0e1eda4b54a89b5d37
M 01d06cee06203b0ba616ca1924f1878aecc1a791
M 0222768baf6b60742c4a8332308edf2be0f4a2e4
+M 055f1d738016ae5caf3eba42e656782de8490890
M 0578e16146527b0782530c40ba2db17531756d96
M 0865c3a85b9687f857b51d1ae56c8f8e97bdbf5b
M 0e5fd5a54cb271ba494e7e30b45d4228b33364a9
M 1184e08f0d1c40c754e39fd5d6340d1057fbdd17
+M 11e18bf5622cafdb10723cee895885d676b35a7b
M 1afeea8bdc0ef96df1d52ea343957dc396723f8c
M 20d0a2811c15cb65a402eb063e1359236d7014ea
+M 21031e3ab2ae227540e2a078d0336d894361acba
M 2538ae42fb0c774023deed5264291b2fe6658cb8
M 26edcd457ea507075a4c82e2787f11f11a432876
M 2b1607831ccf97909122eae9247116d9a075c7cf
M 2e8219cf3d047abc3a7e2611bf284aadbc20b7d6
M 2f8a1f7645c30cb40d4f080ce4b4099964028751
+M 366a556210a1a7b047c4b4dd8da4fc74af6bd301
M 386805560479b35276d88605c5acf805e3004aa5
M 39c772a0764b323f98ab58e00345f4fca924c425
M 3a1fa0b23f1778594b6339dfcfbc90dcc43a87d4
+M 47691ae5c2c6528d95795a51e98e02b54a033c88
M 4799fef822a96f5d1b2827d5cc42ab886fd4e673
M 4837a6cb5ea319f6bf87aaf3bf49c1e4e99e45a7
M 4b0868a8ee6f53d8543860a458821c35ec2a79a2
M 4f4ce80ba2de80219e13943cbd6edbf411117234
M 574b2a9997c94011794fa2e083181e23c9936f15
+M 585df048b7730ee36644a4bb7c1a20fcf977d79e
M 6242c682dfce4a2ba4869978c28b14f1472484c1
M 63a1088a9253da0452497440e900d35a5415c3c9
M 7304dfbd8f3044d8d46981abf2058c6cb732d219
@@ -114,15 +121,18 @@ M 746914a5084f5db8d0c0f052802c7ef2217b9b6b
M 7690a1fd087aff24ad8e8e1ab2e0722e24bf3d81
M 8118c5a301f22dea16c9d235ad378584a0ae2466
M 8583a24ac541dc373503d7a6c59cd90890acdae3
+M 86667619dc61c8a84672fc5a583b4517f48176c5
M 896f1abec915897967905762d6e850cfb6bac3bc
M 8aebfa30047f32e3a6b4feb7ffdd89208ec4f435
M 8e131133c8566a124605cb06e0b6db98fddb5972
M 8e650cfe3efd63a06c25b7e912d9d4db61598eb1
+M 8f8e587cde7ec97410da6b0586810ccf8f0f3233
M 8fbf0941ca52aac406b3efcf2e0d73436c72d0f1
M 91bb2a18f4653f997e455c5089e7a009310591c6
M 9a9e0a8a37608195c4ef6fbf386728d13d025d2d
M 9b6666816a21d8b85cc051be59a8ab6e968c5d0d
M 9ecea375930d1a217f35da65e0d6073f7a80bf7d
+M a034e5853484a7ef0a9f0ad1ae6582178f622f31
M a1710bdd783afcd667d9e72ccb031480d3806850
M a89ca98653b95bcd2ab5cadbe9a419a4cec4893a
M a9fe8470ce9522f08c298e58f76636c48d57b7a4
@@ -131,9 +141,11 @@ M b2b6f0137cd9f1bafff5d487e8383453692a2ff0
M b2e5fb6583d64deeb8a42d2eeb6935c5f33dc0c6
M c01c68ddda10ec6db492b86d818498bec0bf4228
M cf8ac102e728b8745da265ec969f7b5160829fdb
+M df3854cefae5d2da71f8d222a86218e81df7bd05
M e578ea946aa13177adef98727e3e4606652ffc76
M e6d42f6d6684feb46fc0bfbc0520ed13f0f002aa
M e92e57c1f14de911cd2315a907443c79e91a94de
+M ed9298066428c0dfc4590a556876d696a4ba13c0
M f9a42a528f4edfa7bcc62d5885eebaeb25224cec
M fa973bd7ab43099151f83beea351b80c7140eaab
M fd6689948bc7ad153ccfd7f64554eafbf64b20d6
[6/9] cxf git commit: [CXF-4209] Server side message redelivery
support for WS-RM
Posted by dk...@apache.org.
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RedeliveryQueueImpl.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RedeliveryQueueImpl.java b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RedeliveryQueueImpl.java
new file mode 100644
index 0000000..e2ff271
--- /dev/null
+++ b/rt/ws/rm/src/main/java/org/apache/cxf/ws/rm/soap/RedeliveryQueueImpl.java
@@ -0,0 +1,699 @@
+/**
+ * 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.soap;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TimerTask;
+import java.util.TreeSet;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.xml.stream.XMLStreamReader;
+
+import org.w3c.dom.Node;
+
+import org.apache.cxf.Bus;
+import org.apache.cxf.binding.soap.SoapMessage;
+import org.apache.cxf.common.logging.LogUtils;
+import org.apache.cxf.endpoint.Endpoint;
+import org.apache.cxf.interceptor.Interceptor;
+import org.apache.cxf.interceptor.InterceptorChain;
+import org.apache.cxf.interceptor.InterceptorProvider;
+import org.apache.cxf.io.CachedOutputStream;
+import org.apache.cxf.message.Exchange;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.Phase;
+import org.apache.cxf.phase.PhaseInterceptorChain;
+import org.apache.cxf.phase.PhaseManager;
+import org.apache.cxf.staxutils.StaxSource;
+import org.apache.cxf.ws.rm.DestinationSequence;
+import org.apache.cxf.ws.rm.RMCaptureInInterceptor;
+import org.apache.cxf.ws.rm.RMContextUtils;
+import org.apache.cxf.ws.rm.RMManager;
+import org.apache.cxf.ws.rm.RMMessageConstants;
+import org.apache.cxf.ws.rm.RMProperties;
+import org.apache.cxf.ws.rm.RedeliveryQueue;
+import org.apache.cxf.ws.rm.RetryStatus;
+import org.apache.cxf.ws.rm.manager.RetryPolicyType;
+import org.apache.cxf.ws.rm.persistence.RMStore;
+import org.apache.cxf.ws.rm.v200702.Identifier;
+import org.apache.cxf.ws.rm.v200702.SequenceType;
+
+/**
+ *
+ */
+public class RedeliveryQueueImpl implements RedeliveryQueue {
+ private static final Logger LOG = LogUtils.getL7dLogger(RedeliveryQueueImpl.class);
+
+ private Map<String, List<RedeliverCandidate>> candidates =
+ new HashMap<String, List<RedeliverCandidate>>();
+ private Map<String, List<RedeliverCandidate>> suspendedCandidates =
+ new HashMap<String, List<RedeliverCandidate>>();
+
+ private RMManager manager;
+
+ private int undeliveredCount;
+
+ public RedeliveryQueueImpl(RMManager m) {
+ manager = m;
+ }
+
+ public RMManager getManager() {
+ return manager;
+ }
+
+ public void setManager(RMManager m) {
+ manager = m;
+ }
+
+ public void addUndelivered(Message message) {
+ cacheUndelivered(message);
+ }
+
+ /**
+ * @param seq the sequence under consideration
+ * @return the number of undelivered messages for that sequence
+ */
+ public synchronized int countUndelivered(DestinationSequence seq) {
+ List<RedeliverCandidate> sequenceCandidates = getSequenceCandidates(seq);
+ return sequenceCandidates == null ? 0 : sequenceCandidates.size();
+ }
+
+ public int countUndelivered() {
+ return undeliveredCount;
+ }
+
+ public boolean isEmpty() {
+ return 0 == getUndelivered().size();
+ }
+ public void purgeAll(DestinationSequence seq) {
+ Collection<Long> purged = new ArrayList<Long>();
+ synchronized (this) {
+ LOG.fine("Start purging redeliver candidates.");
+ List<RedeliverCandidate> sequenceCandidates = getSequenceCandidates(seq);
+ if (null != sequenceCandidates) {
+ for (int i = sequenceCandidates.size() - 1; i >= 0; i--) {
+ RedeliverCandidate candidate = sequenceCandidates.get(i);
+ long m = candidate.getNumber();
+ sequenceCandidates.remove(i);
+ candidate.resolved();
+ undeliveredCount--;
+ purged.add(m);
+ }
+ if (sequenceCandidates.isEmpty()) {
+ candidates.remove(seq.getIdentifier().getValue());
+ }
+ }
+ LOG.fine("Completed purging redeliver candidates.");
+ }
+ if (purged.size() > 0) {
+ RMStore store = manager.getStore();
+ if (null != store) {
+ store.removeMessages(seq.getIdentifier(), purged, false);
+ }
+ }
+ }
+
+ public List<Long> getUndeliveredMessageNumbers(DestinationSequence seq) {
+ List<Long> undelivered = new ArrayList<Long>();
+ List<RedeliverCandidate> sequenceCandidates = getSequenceCandidates(seq);
+ if (null != sequenceCandidates) {
+ for (int i = 0; i < sequenceCandidates.size(); i++) {
+ RedeliverCandidate candidate = sequenceCandidates.get(i);
+ RMProperties properties = RMContextUtils.retrieveRMProperties(candidate.getMessage(),
+ false);
+ SequenceType st = properties.getSequence();
+ undelivered.add(st.getMessageNumber());
+ }
+ }
+ return undelivered;
+ }
+
+ /**
+ * @param seq the sequence under consideration
+ * @return the list of resend candidates for that sequence
+ * @pre called with mutex held
+ */
+ protected List<RedeliverCandidate> getSequenceCandidates(DestinationSequence seq) {
+ return getSequenceCandidates(seq.getIdentifier().getValue());
+ }
+
+ /**
+ * @param key the sequence identifier under consideration
+ * @return the list of resend candidates for that sequence
+ * @pre called with mutex held
+ */
+ protected List<RedeliverCandidate> getSequenceCandidates(String key) {
+ List<RedeliverCandidate> sc = candidates.get(key);
+ if (null == sc) {
+ sc = suspendedCandidates.get(key);
+ }
+ return sc;
+ }
+
+ /**
+ * @param key the sequence identifier under consideration
+ * @return true if the sequence is currently suspended; false otherwise
+ * @pre called with mutex held
+ */
+ protected boolean isSequenceSuspended(String key) {
+ return suspendedCandidates.containsKey(key);
+ }
+
+ public RetryStatus getRedeliveryStatus(DestinationSequence seq, long num) {
+ List<RedeliverCandidate> sequenceCandidates = getSequenceCandidates(seq);
+ if (null != sequenceCandidates) {
+ for (int i = 0; i < sequenceCandidates.size(); i++) {
+ RedeliverCandidate candidate = sequenceCandidates.get(i);
+ RMProperties properties = RMContextUtils.retrieveRMProperties(candidate.getMessage(),
+ false);
+ SequenceType st = properties.getSequence();
+ if (num == st.getMessageNumber()) {
+ return candidate;
+ }
+ }
+ }
+ return null;
+ }
+
+
+ public Map<Long, RetryStatus> getRedeliveryStatuses(DestinationSequence seq) {
+ Map<Long, RetryStatus> cp = new HashMap<Long, RetryStatus>();
+ List<RedeliverCandidate> sequenceCandidates = getSequenceCandidates(seq);
+ if (null != sequenceCandidates) {
+ for (int i = 0; i < sequenceCandidates.size(); i++) {
+ RedeliverCandidate candidate = sequenceCandidates.get(i);
+ RMProperties properties = RMContextUtils.retrieveRMProperties(candidate.getMessage(),
+ false);
+ SequenceType st = properties.getSequence();
+ cp.put(st.getMessageNumber(), candidate);
+ }
+ }
+ return cp;
+ }
+
+
+ public void start() {
+ // TODO Auto-generated method stub
+
+ }
+
+
+ public void stop(DestinationSequence seq) {
+ synchronized (this) {
+ List<RedeliverCandidate> sequenceCandidates = getSequenceCandidates(seq);
+ if (null != sequenceCandidates) {
+ for (int i = sequenceCandidates.size() - 1; i >= 0; i--) {
+ RedeliverCandidate candidate = sequenceCandidates.get(i);
+ candidate.cancel();
+ }
+ LOG.log(Level.FINE, "Cancelled redeliveriss for sequence {0}.",
+ seq.getIdentifier().getValue());
+ }
+ }
+ }
+
+
+ public void suspend(DestinationSequence seq) {
+ synchronized (this) {
+ String key = seq.getIdentifier().getValue();
+ List<RedeliverCandidate> sequenceCandidates = candidates.remove(key);
+ if (null != sequenceCandidates) {
+ for (int i = sequenceCandidates.size() - 1; i >= 0; i--) {
+ RedeliverCandidate candidate = sequenceCandidates.get(i);
+ candidate.suspend();
+ }
+ suspendedCandidates.put(key, sequenceCandidates);
+ LOG.log(Level.FINE, "Suspended redeliveris for sequence {0}.", key);
+ }
+ }
+ }
+
+
+ public void resume(DestinationSequence seq) {
+ synchronized (this) {
+ String key = seq.getIdentifier().getValue();
+ List<RedeliverCandidate> sequenceCandidates = suspendedCandidates.remove(key);
+ if (null != sequenceCandidates) {
+ for (int i = 0; i < sequenceCandidates.size(); i++) {
+ RedeliverCandidate candidate = sequenceCandidates.get(i);
+ candidate.resume();
+ }
+ candidates.put(key, sequenceCandidates);
+ LOG.log(Level.FINE, "Resumed redeliveries for sequence {0}.", key);
+ }
+ }
+ }
+
+ /**
+ * Accepts a new resend candidate.
+ *
+ * @param ctx the message context.
+ * @return ResendCandidate
+ */
+ protected RedeliverCandidate cacheUndelivered(Message message) {
+ RMProperties rmps = RMContextUtils.retrieveRMProperties(message, false);
+ SequenceType st = rmps.getSequence();
+ Identifier sid = st.getIdentifier();
+ String key = sid.getValue();
+
+ RedeliverCandidate candidate = null;
+
+ synchronized (this) {
+ List<RedeliverCandidate> sequenceCandidates = getSequenceCandidates(key);
+ if (null == sequenceCandidates) {
+ sequenceCandidates = new ArrayList<RedeliverCandidate>();
+ candidates.put(key, sequenceCandidates);
+ }
+ candidate = getRedeliverCandidate(st, sequenceCandidates);
+ if (candidate == null) {
+ candidate = new RedeliverCandidate(message);
+ if (isSequenceSuspended(key)) {
+ candidate.suspend();
+ }
+ sequenceCandidates.add(candidate);
+ undeliveredCount++;
+ }
+ }
+ LOG.fine("Cached undelivered message.");
+ return candidate;
+ }
+
+ private RedeliverCandidate getRedeliverCandidate(SequenceType st, List<RedeliverCandidate> rcs) {
+ // assume the size of candidates to be relatively small; otherwise we should use message numbers as keys
+ for (RedeliverCandidate rc : rcs) {
+ if (st.getMessageNumber() == rc.getNumber()) {
+ return rc;
+ }
+ }
+ return null;
+ }
+
+ protected void purgeDelivered(RedeliverCandidate candidate) {
+ RMProperties rmps = RMContextUtils.retrieveRMProperties(candidate.getMessage(), false);
+ SequenceType st = rmps.getSequence();
+ Identifier sid = st.getIdentifier();
+ String key = sid.getValue();
+
+ synchronized (this) {
+ List<RedeliverCandidate> sequenceCandidates = getSequenceCandidates(key);
+ if (null != sequenceCandidates) {
+ // TODO use a constant op instead of this inefficient linear op
+ sequenceCandidates.remove(candidate);
+ undeliveredCount--;
+ }
+ if (sequenceCandidates.isEmpty()) {
+ candidates.remove(sid.getValue());
+ }
+
+ }
+ LOG.fine("Purged delivered message.");
+
+ }
+
+ /**
+ * @return a map relating sequence ID to a lists of un-acknowledged messages
+ * for that sequence
+ */
+ protected Map<String, List<RedeliverCandidate>> getUndelivered() {
+ return candidates;
+ }
+
+ private static InterceptorChain getRedeliveryInterceptorChain(Message m, String phase) {
+ Exchange exchange = m.getExchange();
+ Endpoint ep = exchange.getEndpoint();
+ Bus bus = exchange.getBus();
+
+ PhaseManager pm = bus.getExtension(PhaseManager.class);
+ SortedSet<Phase> phases = new TreeSet<Phase>(pm.getInPhases());
+ for (Iterator<Phase> it = phases.iterator(); it.hasNext();) {
+ Phase p = it.next();
+ if (phase.equals(p.getName())) {
+ break;
+ }
+ it.remove();
+ }
+ PhaseInterceptorChain chain = new PhaseInterceptorChain(phases);
+ List<Interceptor<? extends Message>> il = ep.getInInterceptors();
+ addInterceptors(chain, il);
+ il = ep.getService().getInInterceptors();
+ addInterceptors(chain, il);
+ il = ep.getBinding().getInInterceptors();
+ addInterceptors(chain, il);
+ il = bus.getInInterceptors();
+ addInterceptors(chain, il);
+ if (ep.getService().getDataBinding() instanceof InterceptorProvider) {
+ il = ((InterceptorProvider)ep.getService().getDataBinding()).getInInterceptors();
+ addInterceptors(chain, il);
+ }
+
+ return chain;
+ }
+
+ private static void addInterceptors(PhaseInterceptorChain chain,
+ List<Interceptor<? extends Message>> il) {
+ for (Interceptor<? extends Message> i : il) {
+ final String iname = i.getClass().getSimpleName();
+ if ("OneWayProcessorInterceptor".equals(iname)
+ || "MAPAggregatorImpl".equals(iname)
+ || "RMInInterceptor".equals(iname)) {
+ continue;
+ }
+ chain.add(i);
+ }
+ }
+
+ //TODO refactor this class to unify its functionality with that of ResendCandidate
+ protected class RedeliverCandidate implements Runnable, RetryStatus {
+ private Message message;
+ private long number;
+ private Date next;
+ private TimerTask nextTask;
+ private int retries;
+ private int maxRetries;
+ private long nextInterval;
+ private long backoff;
+ private boolean pending;
+ private boolean suspended;
+
+ protected RedeliverCandidate(Message m) {
+ message = m;
+ if (message instanceof SoapMessage) {
+ // remove old message headers like WSS headers
+ ((SoapMessage)message).getHeaders().clear();
+ }
+ RetryPolicyType rmrp = null != manager.getDestinationPolicy()
+ ? manager.getDestinationPolicy().getRetryPolicy() : null;
+ long baseRedeliveryInterval = Long.parseLong(DEFAULT_BASE_REDELIVERY_INTERVAL);
+ if (null != rmrp && rmrp.getInterval() > 0L) {
+ baseRedeliveryInterval = rmrp.getInterval();
+ }
+ backoff = RedeliveryQueue.DEFAULT_EXPONENTIAL_BACKOFF;
+ next = new Date(System.currentTimeMillis() + baseRedeliveryInterval);
+ nextInterval = baseRedeliveryInterval * backoff;
+ maxRetries = null != rmrp ? rmrp.getMaxRetries() : 0;
+
+ RMProperties rmprops = RMContextUtils.retrieveRMProperties(message, false);
+ if (null != rmprops) {
+ number = rmprops.getSequence().getMessageNumber();
+ }
+
+ if (null != manager.getTimer() && maxRetries != 0) {
+ schedule();
+ }
+
+ }
+
+ /**
+ * Initiate redelivery asynchronsly.
+ *
+ */
+ protected void initiate() {
+ pending = true;
+ Endpoint ep = message.getExchange().get(Endpoint.class);
+ Executor executor = ep.getExecutor();
+ if (null == executor) {
+ executor = ep.getService().getExecutor();
+ LOG.log(Level.FINE, "Using service executor {0}", executor.getClass().getName());
+ } else {
+ LOG.log(Level.FINE, "Using endpoint executor {0}", executor.getClass().getName());
+ }
+
+ try {
+ executor.execute(this);
+ } catch (RejectedExecutionException ex) {
+ LOG.log(Level.SEVERE, "RESEND_INITIATION_FAILED_MSG", ex);
+ }
+ }
+
+ public void run() {
+ try {
+ if (isPending()) {
+ // redeliver
+ redeliver();
+ purgeDelivered(this);
+ resolved();
+ }
+ } catch (Exception ex) {
+ LOG.log(Level.WARNING, "redelivery failed", ex);
+ } finally {
+ attempted();
+ }
+ }
+
+
+ private void redeliver() throws Exception {
+ LOG.log(Level.INFO, "Redelivering ... for " + (1 + retries));
+ String restartingPhase;
+ if (message.getContent(Exception.class) != null) {
+ message.removeContent(Exception.class);
+ message.getExchange().put(Exception.class, null);
+
+ // clean-up message for redelivery
+ closeStreamResources();
+ message.removeContent(Node.class);
+ }
+
+
+ InputStream is = null;
+ CachedOutputStream cos = (CachedOutputStream)message.get(RMMessageConstants.SAVED_CONTENT);
+ is = cos.getInputStream();
+ message.setContent(InputStream.class, is);
+ message = message.getExchange().getEndpoint().getBinding().createMessage(message);
+ restartingPhase = Phase.POST_STREAM;
+ // skip some interceptor chain phases for redelivery
+ InterceptorChain chain = getRedeliveryInterceptorChain(message, restartingPhase);
+ ListIterator<Interceptor<? extends Message>> iterator = chain.getIterator();
+ while (iterator.hasNext()) {
+ Interceptor<? extends Message> incept = iterator.next();
+ if (incept.getClass().getName().equals(RMCaptureInInterceptor.class.getName())) {
+ chain.remove(incept);
+ }
+ }
+ message.getExchange().setInMessage(message);
+ message.setInterceptorChain(chain);
+ chain.doIntercept(message);
+ Exception ex = message.getContent(Exception.class);
+ if (null != ex) {
+ throw ex;
+ }
+ }
+
+ public long getNumber() {
+ return number;
+ }
+
+ public Date getNext() {
+ return next;
+ }
+
+ public Date getPrevious() {
+ if (retries > 0) {
+ return new Date(next.getTime() - nextInterval / backoff);
+ }
+ return null;
+ }
+
+ public int getRetries() {
+ return retries;
+ }
+
+ public int getMaxRetries() {
+ return maxRetries;
+ }
+
+ public long getNextInterval() {
+ return nextInterval;
+ }
+
+ public long getBackoff() {
+ return backoff;
+ }
+
+ public boolean isPending() {
+ return pending;
+ }
+
+ public boolean isSuspended() {
+ return suspended;
+ }
+
+ /**
+ * the message has been delivered to the application
+ */
+ protected synchronized void resolved() {
+ pending = false;
+ next = null;
+ if (null != nextTask) {
+ nextTask.cancel();
+ }
+ }
+
+ /**
+ * Cancel further redelivery (although not successfully delivered).
+ */
+ protected void cancel() {
+ if (null != nextTask) {
+ nextTask.cancel();
+ closeStreamResources();
+ releaseSavedMessage();
+ }
+ }
+
+ protected void suspend() {
+ suspended = true;
+ pending = false;
+ //TODO release the message and later reload it upon resume
+ //cancel();
+ if (null != nextTask) {
+ nextTask.cancel();
+ }
+
+ }
+
+ protected void resume() {
+ suspended = false;
+ next = new Date(System.currentTimeMillis());
+ attempted();
+ }
+
+ private void releaseSavedMessage() {
+ CachedOutputStream saved = (CachedOutputStream)message.remove(RMMessageConstants.SAVED_CONTENT);
+ if (saved != null) {
+ saved.releaseTempFileHold();
+ try {
+ saved.close();
+ } catch (IOException e) {
+ // ignore
+ }
+ }
+ // Any unclosed resources must be closed to release the temp files.
+ Closeable closeable = (Closeable)message.get(RMMessageConstants.ATTACHMENTS_CLOSEABLE);
+ if (closeable != null) {
+ try {
+ closeable.close();
+ } catch (IOException e) {
+ // ignore
+ }
+ }
+ }
+
+ /*
+ * Close all stream-like resources stored in the message
+ */
+ private void closeStreamResources() {
+ InputStream oin = message.getContent(InputStream.class);
+ if (oin != null) {
+ try {
+ oin.close();
+ } catch (Exception e) {
+ // ignore
+ }
+ message.removeContent(InputStream.class);
+ }
+ XMLStreamReader oreader = message.getContent(XMLStreamReader.class);
+ if (oreader != null) {
+ try {
+ oreader.close();
+ } catch (Exception e) {
+ // ignore
+ }
+ message.removeContent(XMLStreamReader.class);
+ }
+ List olist = message.getContent(List.class);
+ if (olist != null && olist.size() == 1) {
+ Object o = olist.get(0);
+ if (o instanceof XMLStreamReader) {
+ oreader = (XMLStreamReader)o;
+ } else if (o instanceof StaxSource) {
+ oreader = ((StaxSource)o).getXMLStreamReader();
+ }
+
+ if (oreader != null) {
+ try {
+ oreader.close();
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ message.removeContent(List.class);
+ }
+ }
+
+ /**
+ * @return associated message context
+ */
+ protected Message getMessage() {
+ return message;
+ }
+
+ /**
+ * A resend has been attempted. Schedule the next attempt.
+ */
+ protected synchronized void attempted() {
+ pending = false;
+ retries++;
+ if (null != next && maxRetries != retries) {
+ next = new Date(next.getTime() + nextInterval);
+ nextInterval *= backoff;
+ schedule();
+ }
+ }
+
+ protected final synchronized void schedule() {
+ if (null == manager.getTimer()) {
+ return;
+ }
+ class RedeliverTask extends TimerTask {
+ RedeliverCandidate candidate;
+
+ RedeliverTask(RedeliverCandidate c) {
+ candidate = c;
+ }
+
+ @Override
+ public void run() {
+ if (!candidate.isPending()) {
+ candidate.initiate();
+ }
+ }
+ }
+ nextTask = new RedeliverTask(this);
+ try {
+ manager.getTimer().schedule(nextTask, next);
+ } catch (IllegalStateException ex) {
+ LOG.log(Level.WARNING, "SCHEDULE_RESEND_FAILED_MSG", ex);
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-manager-types.xsd
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-manager-types.xsd b/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-manager-types.xsd
index 99b60c4..e5425cd 100644
--- a/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-manager-types.xsd
+++ b/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-manager-types.xsd
@@ -214,6 +214,21 @@
</xs:documentation>
</xs:annotation>
</xs:attribute>
+ <xs:attribute name="interval" type="xs:unsignedLong" use="optional" default="3000">
+ <xs:annotation>
+ <xs:documentation>
+ The time interval between each retry in milliseconds.
+ </xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
+ <xs:attribute name="algorithm" type="xs:string" use="optional" default="ExponentialBackoff">
+ <xs:annotation>
+ <xs:documentation>
+ The name of the interval adjustment algorithm.
+ The default algorithm is ExponentialBackoff.
+ </xs:documentation>
+ </xs:annotation>
+ </xs:attribute>
</xs:complexType>
<xs:complexType name="DeliveryAssuranceType">
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-policy.xjb
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-policy.xjb b/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-policy.xjb
index 5bfd418..a049c3f 100644
--- a/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-policy.xjb
+++ b/rt/ws/rm/src/main/resources/schemas/configuration/wsrm-policy.xjb
@@ -53,4 +53,13 @@
</jaxb:property>
</jaxb:bindings>
+ <jaxb:bindings schemaLocation="wsrm-manager-types.xsd"
+ node="//xs:complexType[@name='RetryPolicyType']//xs:attribute[@name='interval']">
+ <jaxb:property>
+ <jaxb:baseType>
+ <jaxb:javaType name="java.lang.Long"/>
+ </jaxb:baseType>
+ </jaxb:property>
+ </jaxb:bindings>
+
</jaxb:bindings>
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java
index a189932..546f5f3 100644
--- a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java
+++ b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/DestinationSequenceTest.java
@@ -567,7 +567,7 @@ public class DestinationSequenceTest extends Assert {
DestinationSequence seq = new DestinationSequence(id, ref, destination,
ProtocolVariation.RM10WSA200408);
- destination.removeSequence(seq);
+ destination.terminateSequence(seq);
EasyMock.expectLastCall();
Message message = setUpMessage("1");
@@ -601,7 +601,7 @@ public class DestinationSequenceTest extends Assert {
long lastAppMessage = System.currentTimeMillis() - 30000L;
EasyMock.expect(rme.getLastControlMessage()).andReturn(0L);
EasyMock.expect(rme.getLastApplicationMessage()).andReturn(lastAppMessage);
- destination.removeSequence(seq);
+ destination.terminateSequence(seq);
EasyMock.expectLastCall();
control.replay();
st.run();
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMEndpointTest.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMEndpointTest.java b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMEndpointTest.java
index 7600cd5..acd179c 100644
--- a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMEndpointTest.java
+++ b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMEndpointTest.java
@@ -292,6 +292,10 @@ public class RMEndpointTest extends Assert {
EasyMock.expect(manager.getRetransmissionQueue()).andReturn(queue).anyTimes();
queue.stop(ss);
EasyMock.expectLastCall().anyTimes();
+ RedeliveryQueue dqueue = control.createMock(RedeliveryQueue.class);
+ EasyMock.expect(manager.getRedeliveryQueue()).andReturn(dqueue).anyTimes();
+ dqueue.stop(ds);
+ EasyMock.expectLastCall().anyTimes();
control.replay();
rme.getDestination().addSequence(ds, false);
rme.getSource().addSequence(ss, false);
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMInInterceptorTest.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMInInterceptorTest.java b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMInInterceptorTest.java
index a68d821..c9411d8 100644
--- a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMInInterceptorTest.java
+++ b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMInInterceptorTest.java
@@ -40,6 +40,8 @@ import org.apache.cxf.ws.addressing.JAXWSAConstants;
import org.apache.cxf.ws.addressing.MAPAggregator;
import org.apache.cxf.ws.addressing.VersionTransformer.Names200408;
import org.apache.cxf.ws.policy.AssertionInfoMap;
+import org.apache.cxf.ws.rm.manager.DestinationPolicyType;
+import org.apache.cxf.ws.rm.manager.RetryPolicyType;
import org.apache.cxf.ws.rm.v200702.CreateSequenceResponseType;
import org.apache.cxf.ws.rm.v200702.Identifier;
import org.apache.cxf.ws.rm.v200702.SequenceAcknowledgement;
@@ -367,6 +369,29 @@ public class RMInInterceptorTest extends Assert {
} catch (Exception e) {
fail("unexpected exception thrown from handleFault: " + e);
}
+
+ control.reset();
+ org.apache.cxf.transport.Destination td = control.createMock(org.apache.cxf.transport.Destination.class);
+ EasyMock.expect(exchange.getDestination()).andReturn(td).anyTimes();
+ EasyMock.expect(message.getExchange()).andReturn(exchange).anyTimes();
+ EasyMock.expect(message.get(RMMessageConstants.RM_PROTOCOL_VARIATION))
+ .andReturn(ProtocolVariation.RM10WSA200408).anyTimes();
+ EasyMock.expect(message.getContent(Exception.class)).andReturn(new SequenceFault("no sequence")).anyTimes();
+ DestinationPolicyType dp = new DestinationPolicyType();
+ RetryPolicyType rp = new RetryPolicyType();
+ dp.setRetryPolicy(rp);
+ EasyMock.expect(manager.getDestinationPolicy()).andReturn(dp).anyTimes();
+ RedeliveryQueue rq = control.createMock(RedeliveryQueue.class);
+ EasyMock.expect(manager.getRedeliveryQueue()).andReturn(rq).anyTimes();
+ rq.addUndelivered(message);
+ EasyMock.expectLastCall().andThrow(new RuntimeException("shouldn't be queued")).anyTimes();
+ control.replay();
+
+ try {
+ interceptor.handleFault(message);
+ } catch (Exception e) {
+ fail("unexpected exception thrown from handleFault: " + e);
+ }
}
@Test
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java
index 37fb6ac..5231bd4 100644
--- a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java
+++ b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/RMManagerTest.java
@@ -519,7 +519,7 @@ public class RMManagerTest extends Assert {
InterfaceInfo ii = control.createMock(InterfaceInfo.class);
setUpEndpointForRecovery(endpoint, ei, si, bi, ii);
Conduit conduit = control.createMock(Conduit.class);
- setUpRecoverReliableEndpoint(endpoint, conduit, null, null, null, null);
+ setUpRecoverReliableEndpoint(endpoint, conduit, null, null, null);
control.replay();
manager.recoverReliableEndpoint(endpoint, conduit);
control.verify();
@@ -528,7 +528,7 @@ public class RMManagerTest extends Assert {
setUpEndpointForRecovery(endpoint, ei, si, bi, ii);
SourceSequence ss = control.createMock(SourceSequence.class);
DestinationSequence ds = control.createMock(DestinationSequence.class);
- setUpRecoverReliableEndpoint(endpoint, conduit, ss, ds, null, null);
+ setUpRecoverReliableEndpoint(endpoint, conduit, ss, ds, null);
control.replay();
manager.recoverReliableEndpoint(endpoint, conduit);
control.verify();
@@ -536,16 +536,10 @@ public class RMManagerTest extends Assert {
control.reset();
setUpEndpointForRecovery(endpoint, ei, si, bi, ii);
RMMessage m = control.createMock(RMMessage.class);
- Capture<Message> mc = Capture.newInstance();
- setUpRecoverReliableEndpoint(endpoint, conduit, ss, ds, m, mc);
+ setUpRecoverReliableEndpoint(endpoint, conduit, ss, ds, m);
control.replay();
manager.recoverReliableEndpoint(endpoint, conduit);
control.verify();
-
- Message msg = mc.getValue();
- assertNotNull(msg);
- assertNotNull(msg.getExchange());
- assertSame(msg, msg.getExchange().getOutMessage());
}
@Test
@@ -595,9 +589,11 @@ public class RMManagerTest extends Assert {
DestinationSequence ds, RMMessage m,
Capture<Message> mc) throws IOException {
RMStore store = control.createMock(RMStore.class);
- RetransmissionQueue queue = control.createMock(RetransmissionQueue.class);
+ RetransmissionQueue oqueue = control.createMock(RetransmissionQueue.class);
+ RedeliveryQueue iqueue = control.createMock(RedeliveryQueue.class);
manager.setStore(store);
- manager.setRetransmissionQueue(queue);
+ manager.setRetransmissionQueue(oqueue);
+ manager.setRedeliveryQueue(iqueue);
Collection<SourceSequence> sss = new ArrayList<SourceSequence>();
if (null != ss) {
@@ -652,10 +648,11 @@ public class RMManagerTest extends Assert {
return;
}
- queue.addUnacknowledged(EasyMock.capture(mc));
-
+ oqueue.addUnacknowledged(EasyMock.capture(mc));
EasyMock.expectLastCall();
- queue.start();
+ oqueue.start();
+ EasyMock.expectLastCall();
+ iqueue.start();
EasyMock.expectLastCall();
}
@@ -678,12 +675,14 @@ public class RMManagerTest extends Assert {
void setUpRecoverReliableEndpoint(Endpoint endpoint,
Conduit conduit,
SourceSequence ss,
- DestinationSequence ds, RMMessage m, Capture<Message> mc)
+ DestinationSequence ds, RMMessage m)
throws IOException {
RMStore store = control.createMock(RMStore.class);
- RetransmissionQueue queue = control.createMock(RetransmissionQueue.class);
+ RetransmissionQueue oqueue = control.createMock(RetransmissionQueue.class);
+ RedeliveryQueue iqueue = control.createMock(RedeliveryQueue.class);
manager.setStore(store);
- manager.setRetransmissionQueue(queue);
+ manager.setRetransmissionQueue(oqueue);
+ manager.setRedeliveryQueue(iqueue);
Collection<SourceSequence> sss = new ArrayList<SourceSequence>();
if (null != ss) {
@@ -747,13 +746,11 @@ public class RMManagerTest extends Assert {
is.close();
EasyMock.expect(m.getContent()).andReturn(cos).anyTimes();
- if (mc != null) {
- queue.addUnacknowledged(EasyMock.capture(mc));
- } else {
- queue.addUnacknowledged(EasyMock.isA(Message.class));
- }
+ oqueue.addUnacknowledged(EasyMock.isA(Message.class));
+ EasyMock.expectLastCall();
+ oqueue.start();
EasyMock.expectLastCall();
- queue.start();
+ iqueue.start();
EasyMock.expectLastCall();
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStoreTestBase.java
----------------------------------------------------------------------
diff --git a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStoreTestBase.java b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStoreTestBase.java
index a5f9723..87d8e6e 100644
--- a/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStoreTestBase.java
+++ b/rt/ws/rm/src/test/java/org/apache/cxf/ws/rm/persistence/jdbc/RMTxStoreTestBase.java
@@ -68,6 +68,8 @@ public abstract class RMTxStoreTestBase extends Assert {
private static SequenceAcknowledgement ack1;
private static SequenceAcknowledgement ack2;
+ private static final long TIME = System.currentTimeMillis();
+
protected IMocksControl control;
public static void setUpOnce() {
@@ -867,6 +869,7 @@ public abstract class RMTxStoreTestBase extends Assert {
EasyMock.expect(msg.getTo()).andReturn(to).anyTimes();
EasyMock.expect(msg.getContentType()).andReturn("text/xml").anyTimes();
+ EasyMock.expect(msg.getCreatedTime()).andReturn(TIME);
byte[] value = ("Message " + mn.longValue()).getBytes();
ByteArrayInputStream bais = new ByteArrayInputStream(value);
CachedOutputStream cos = new CachedOutputStream();
@@ -936,6 +939,7 @@ public abstract class RMTxStoreTestBase extends Assert {
} else {
assertNull(msg.getTo());
}
+ assertEquals(TIME, msg.getCreatedTime());
try {
InputStream actual = msg.getContent().getInputStream();
assertEquals(new String("Message " + mn), IOUtils.readStringFromStream(actual));
http://git-wip-us.apache.org/repos/asf/cxf/blob/0dd29509/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/RedeliveryTest.java
----------------------------------------------------------------------
diff --git a/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/RedeliveryTest.java b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/RedeliveryTest.java
new file mode 100644
index 0000000..91ada49
--- /dev/null
+++ b/systests/ws-rm/src/test/java/org/apache/cxf/systest/ws/rm/RedeliveryTest.java
@@ -0,0 +1,186 @@
+/**
+ * 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.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.Greeter;
+import org.apache.cxf.greeter_control.GreeterService;
+import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
+import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
+import org.apache.cxf.ws.rm.RMConfiguration;
+import org.apache.cxf.ws.rm.RMManager;
+import org.apache.cxf.ws.rm.manager.RetryPolicyType;
+import org.apache.cxf.ws.rm.persistence.jdbc.RMTxStore;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Tests the redelivery of the message upon a delivery error.
+ */
+public class RedeliveryTest extends AbstractBusClientServerTestBase {
+ public static final String PORT = allocatePort(Server.class);
+ private static final Logger LOG = LogUtils.getLogger(RedeliveryTest.class);
+
+ private static GreeterRecorderImpl serverGreeter;
+ private static Bus serverBus;
+ private Greeter greeter;
+
+
+ public static class Server extends AbstractBusTestServerBase {
+ String port;
+ String pfx;
+ Endpoint ep;
+
+ public Server(String args[]) {
+ port = args[0];
+ pfx = args[1];
+ }
+
+ protected void run() {
+ SpringBusFactory bf = new SpringBusFactory();
+ // use a at-most-once server with sync ack processing
+ System.setProperty("db.name", pfx + "-server");
+ serverBus = bf.createBus("/org/apache/cxf/systest/ws/rm/sync-ack-persistent-server.xml");
+ System.clearProperty("db.name");
+ BusFactory.setDefaultBus(serverBus);
+ RMManager manager = serverBus.getExtension(RMManager.class);
+ RMConfiguration cfg = manager.getConfiguration();
+ cfg.setAcknowledgementInterval(0L);
+
+ RetryPolicyType rp = new RetryPolicyType();
+ rp.setMaxRetries(-1);
+ serverBus.getExtension(RMManager.class).getDestinationPolicy().setRetryPolicy(rp);
+ serverGreeter = new GreeterRecorderImpl();
+ String address = "http://localhost:" + port + "/SoapContext/GreeterPort";
+
+ // publish this robust oneway endpoint
+ ep = Endpoint.create(serverGreeter);
+ ep.publish(address);
+ LOG.info("Published greeter endpoint.");
+ BusFactory.setDefaultBus(null);
+ BusFactory.setThreadDefaultBus(null);
+ }
+ public void tearDown() {
+ ep.stop();
+ ep = null;
+ }
+
+ public static void main(String[] args) {
+ try {
+ Server s = new Server(args);
+ s.start();
+ } catch (Exception ex) {
+ ex.printStackTrace();
+ System.exit(-1);
+ } finally {
+ System.out.println("done!");
+ }
+ }
+ }
+
+ @BeforeClass
+ public static void startServers() throws Exception {
+ RMTxStore.deleteDatabaseFiles("redlv-server", true);
+ assertTrue("server did not launch correctly",
+ launchServer(Server.class, null, new String[]{PORT, "redlv"}, true));
+ }
+
+ @AfterClass
+ public static void cleanUpDerby() throws Exception {
+ RMTxStore.deleteDatabaseFiles("redlv-server", true);
+ }
+
+ @Test
+ public void testAutomaticRedeliveryAfterError() throws Exception {
+ LOG.fine("Creating greeter client");
+ SpringBusFactory bf = new SpringBusFactory();
+ bus = bf.createBus("/org/apache/cxf/systest/ws/rm/rminterceptors.xml");
+ // set the client retry interval much shorter than the slow processing delay
+ RMManager manager = bus.getExtension(RMManager.class);
+ RMConfiguration cfg = manager.getConfiguration();
+ cfg.setBaseRetransmissionInterval(3000L);
+
+ BusFactory.setDefaultBus(bus);
+ GreeterService gs = new GreeterService();
+ greeter = gs.getGreeterPort();
+ updateAddressPort(greeter, PORT);
+
+ assertNull("last greeted by none", serverGreeter.getValue());
+
+ LOG.fine("Invoking greeter for one");
+ greeter.greetMeOneWay("one");
+ LOG.fine("Wait for 4 secs ...");
+ Thread.sleep(4000);
+
+ assertEquals("last greeted by one", "one", serverGreeter.getValue());
+ assertTrue("retransmission running", manager.getRetransmissionQueue().isEmpty());
+
+ LOG.fine("Activating the error trigger and invoking greeter for two");
+ serverGreeter.setThrowAlways(true);
+ greeter.greetMeOneWay("two");
+ LOG.fine("Wait for 4 secs ...");
+ Thread.sleep(4000);
+
+ RMManager serverManager = serverBus.getExtension(RMManager.class);
+
+ assertEquals("last greeted by one", "one", serverGreeter.getValue());
+ assertTrue("retransmission running", manager.getRetransmissionQueue().isEmpty());
+ assertFalse("redelivery not running", serverManager.getRedeliveryQueue().isEmpty());
+
+ LOG.fine("Deactivating the error trigger and wait for 9 secs ...");
+ serverGreeter.setThrowAlways(false);
+ Thread.sleep(9000);
+
+ assertEquals("last greeted by two", "two", serverGreeter.getValue());
+ assertTrue("redelivery running", serverManager.getRedeliveryQueue().isEmpty());
+ }
+
+ private static class GreeterRecorderImpl extends GreeterImpl {
+ private String value;
+ private boolean ex;
+
+ public void greetMeOneWay(String arg0) {
+ if (ex) {
+ throw new RuntimeException("intentional exception");
+ }
+ super.greetMeOneWay(arg0);
+ value = arg0;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ @Override
+ public void setThrowAlways(boolean b) {
+ super.setThrowAlways(b);
+ ex = b;
+ }
+ }
+}
[5/9] cxf git commit: [CXF-6930] Make sure the webclient waits until
the reponse chain has run before doing it's processing
Posted by dk...@apache.org.
[CXF-6930] Make sure the webclient waits until the reponse chain has run before doing it's processing
Project: http://git-wip-us.apache.org/repos/asf/cxf/repo
Commit: http://git-wip-us.apache.org/repos/asf/cxf/commit/0f66cb02
Tree: http://git-wip-us.apache.org/repos/asf/cxf/tree/0f66cb02
Diff: http://git-wip-us.apache.org/repos/asf/cxf/diff/0f66cb02
Branch: refs/heads/3.1.x-fixes
Commit: 0f66cb025baf40cf84207c87f2428c9942fc8328
Parents: 3198d8b
Author: Daniel Kulp <dk...@apache.org>
Authored: Thu Jul 21 16:53:04 2016 -0400
Committer: Daniel Kulp <dk...@apache.org>
Committed: Fri Jul 22 12:13:14 2016 -0400
----------------------------------------------------------------------
.../apache/cxf/jaxrs/client/AbstractClient.java | 19 +++++++++++++++++++
.../cxf/systest/jaxrs/jms/JAXRSJmsTest.java | 2 ++
2 files changed, 21 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cxf/blob/0f66cb02/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java
----------------------------------------------------------------------
diff --git a/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java b/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java
index f9d9aec..93d8111 100644
--- a/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java
+++ b/rt/rs/client/src/main/java/org/apache/cxf/jaxrs/client/AbstractClient.java
@@ -558,6 +558,17 @@ public abstract class AbstractClient implements Client {
Exchange exchange = message.getExchange();
Exception ex = message.getContent(Exception.class);
+ if (ex == null) {
+ ex = message.getExchange().get(Exception.class);
+ }
+ if (ex == null && !exchange.isOneWay()) {
+ synchronized (exchange) {
+ while (exchange.get("IN_CHAIN_COMPLETE") == null) {
+ exchange.wait();
+ }
+ }
+ }
+ ex = message.getContent(Exception.class);
if (ex != null
|| PropertyUtils.isTrue(exchange.get(SERVICE_NOT_AVAIL_PROPERTY))
&& PropertyUtils.isTrue(exchange.get(COMPLETE_IF_SERVICE_NOT_AVAIL_PROPERTY))) {
@@ -1187,6 +1198,10 @@ public abstract class AbstractClient implements Client {
@Override
public void handleMessage(Message message) throws Fault {
+ synchronized (message.getExchange()) {
+ message.getExchange().put("IN_CHAIN_COMPLETE", Boolean.TRUE);
+ message.getExchange().notifyAll();
+ }
if (message.getExchange().isSynchronous()) {
return;
}
@@ -1195,6 +1210,10 @@ public abstract class AbstractClient implements Client {
@Override
public void handleFault(Message message) {
+ synchronized (message.getExchange()) {
+ message.getExchange().put("IN_CHAIN_COMPLETE", Boolean.TRUE);
+ message.getExchange().notifyAll();
+ }
if (message.getExchange().isSynchronous()) {
return;
}
http://git-wip-us.apache.org/repos/asf/cxf/blob/0f66cb02/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
----------------------------------------------------------------------
diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
index 4cf23e4..9567618 100644
--- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
+++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/jms/JAXRSJmsTest.java
@@ -42,6 +42,7 @@ import javax.xml.bind.JAXBContext;
import javax.xml.bind.Marshaller;
import javax.xml.bind.Unmarshaller;
+import org.apache.cxf.interceptor.LoggingInInterceptor;
import org.apache.cxf.jaxrs.client.JAXRSClientFactory;
import org.apache.cxf.jaxrs.client.WebClient;
import org.apache.cxf.jaxrs.model.AbstractResourceInfo;
@@ -87,6 +88,7 @@ public class JAXRSJmsTest extends AbstractBusClientServerTestBase {
+ "&jndiURL=tcp://localhost:" + JMS_PORT;
WebClient client = WebClient.create(endpointAddressUrlEncoded);
+ WebClient.getConfig(client).getInInterceptors().add(new LoggingInInterceptor());
WebClient.getConfig(client).getRequestContext()
.put(org.apache.cxf.message.Message.REQUEST_URI, "/bookstore/books/123");