QPID-4984: WIP - Merge from trunk r.1534385.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/linearstore@1534394 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 24 insertions, 11 deletions

diff --git a/qpid/python/qpid/messaging/driver.py b/qpid/python/qpid/messaging/driver.py
index ac59bcf0a6..06bbe610b8 100644
--- a/qpid/python/qpid/messaging/driver.py
+++ b/qpid/python/qpid/messaging/driver.py
@@ -259,6 +259,15 @@ class LinkIn:
     reliability = link_opts.get("reliability")
     cmds = [MessageCancel(_rcv.destination)]
     cmds.extend(_rcv.on_unlink)
+    msgs = [] #release back messages for the closing receiver
+    msg = rcv.session._pop(rcv)
+    while msg is not None:
+      msgs.append(msg)
+      msg = rcv.session._pop(rcv)
+    if len(msgs) > 0:
+      ids = RangedSet(*[m._transfer_id for m in msgs])
+      log.debug("releasing back messages: %s, as receiver is closing", ids)
+      cmds.append(MessageRelease(ids, True))
     sst.write_cmds(cmds, action)
 
   def del_link(self, sst, rcv, _rcv):
@@ -1283,6 +1292,11 @@ class Engine:
     msg = self._decode(xfr)
     rcv = sst.destinations[xfr.destination].target
     msg._receiver = rcv
+    if rcv.closing or rcv.closed: # release message to a closing receiver
+      ids = RangedSet(*[msg._transfer_id])
+      log.debug("releasing back %s message: %s, as receiver is closing", ids, msg)
+      sst.write_cmd(MessageRelease(ids, True))
+      return
     if rcv.impending is not UNLIMITED:
       assert rcv.received < rcv.impending, "%s, %s" % (rcv.received, rcv.impending)
     rcv.received += 1
diff --git a/qpid/python/qpid/messaging/endpoints.py b/qpid/python/qpid/messaging/endpoints.py
index 3453971a77..6f0e9432dc 100644
--- a/qpid/python/qpid/messaging/endpoints.py
+++ b/qpid/python/qpid/messaging/endpoints.py
@@ -70,8 +70,8 @@ class Connection(Endpoint):
 
   def __init__(self, url=None, **options):
     """
-    Creates a connection. A newly created connection must be connected
-    with the Connection.connect() method before it can be used.
+    Creates a connection. A newly created connection must be opened
+    with the Connection.open() method before it can be used.
 
     @type url: str
     @param url: [ <username> [ / <password> ] @ ] <host> [ : <port> ]
@@ -88,11 +88,6 @@ class Connection(Endpoint):
     @param username: the username for authentication (overriden by url)
     @type password: str
     @param password: the password for authentication (overriden by url)
-
-     * - sasl_min_ssf: the minimum acceptable security strength factor
-     * - sasl_max_ssf: the minimum acceptable security strength factor
-     * - sasl_service: the service name if needed by the SASL mechanism in use
-
     @type sasl_mechanisms: str
     @param sasl_mechanisms: space separated list of permitted sasl mechanisms
     @type sasl_service: str
@@ -100,7 +95,7 @@ class Connection(Endpoint):
     @type sasl_min_ssf: int
     @param sasl_min_ssf: the minimum acceptable security strength factor
     @type sasl_max_ssf: int
-    @param sasl_max_ssf: the minimum acceptable security strength factor
+    @param sasl_max_ssf: the maximum acceptable security strength factor
 
     @type reconnect: bool
     @param reconnect: enable/disable automatic reconnect
diff --git a/qpid/python/qpid/tests/messaging/implementation.py b/qpid/python/qpid/tests/messaging/implementation.py
index c81fb6c99c..fce60c6f38 100644
--- a/qpid/python/qpid/tests/messaging/implementation.py
+++ b/qpid/python/qpid/tests/messaging/implementation.py
@@ -16,9 +16,13 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-try:
+import os
+if 'QPID_USE_SWIG_CLIENT' in os.environ and os.environ['QPID_USE_SWIG_CLIENT']:
+  try:
     from qpid_messaging import *
     from qpid.datatypes import uuid4
-except ImportError, e:
-    print "Falling back to pure bindings, %s" % e
+  except ImportError, e:
+    print "Swigged client not found. Falling back to pure bindings, %s\n" % e
     from qpid.messaging import *
+else:
+  from qpid.messaging import *