Fix a few leaks in the AMQP 1.0 driver.
Break any references that can form a cyclic reference. Several of proton's python classes define __del__ methods, so a cyclic reference must be broken manually. Change-Id: Icabc7abd01a02cb80c7eb4a0de682e90308459ef Closes-Bug: #1496540
This commit is contained in:
parent
8cbf3c170d
commit
e3d99b2127
@ -72,6 +72,10 @@ class Replies(pyngus.ReceiverEventHandler):
|
|||||||
self._credit = 0
|
self._credit = 0
|
||||||
self._receiver.open()
|
self._receiver.open()
|
||||||
|
|
||||||
|
def destroy(self):
|
||||||
|
self._correlation = None
|
||||||
|
self._receiver = None
|
||||||
|
|
||||||
def ready(self):
|
def ready(self):
|
||||||
return self._ready
|
return self._ready
|
||||||
|
|
||||||
@ -157,6 +161,7 @@ class Server(pyngus.ReceiverEventHandler):
|
|||||||
self._incoming = incoming
|
self._incoming = incoming
|
||||||
self._addresses = addresses
|
self._addresses = addresses
|
||||||
self._capacity = 500 # credit per link
|
self._capacity = 500 # credit per link
|
||||||
|
self._receivers = None
|
||||||
|
|
||||||
def attach(self, connection):
|
def attach(self, connection):
|
||||||
"""Create receiver links over the given connection for all the
|
"""Create receiver links over the given connection for all the
|
||||||
@ -180,6 +185,9 @@ class Server(pyngus.ReceiverEventHandler):
|
|||||||
r.open()
|
r.open()
|
||||||
self._receivers.append(r)
|
self._receivers.append(r)
|
||||||
|
|
||||||
|
def destroy(self):
|
||||||
|
self._receivers = None
|
||||||
|
|
||||||
# Pyngus ReceiverLink event callbacks:
|
# Pyngus ReceiverLink event callbacks:
|
||||||
|
|
||||||
def receiver_remote_closed(self, receiver, pn_condition):
|
def receiver_remote_closed(self, receiver, pn_condition):
|
||||||
@ -310,6 +318,15 @@ class Controller(pyngus.ConnectionEventHandler):
|
|||||||
LOG.debug("Waiting for eventloop to exit")
|
LOG.debug("Waiting for eventloop to exit")
|
||||||
self.processor.shutdown(wait, timeout)
|
self.processor.shutdown(wait, timeout)
|
||||||
self.processor = None
|
self.processor = None
|
||||||
|
self._tasks = None
|
||||||
|
self._senders = None
|
||||||
|
for server in self._servers.values():
|
||||||
|
server.destroy()
|
||||||
|
self._servers.clear()
|
||||||
|
self._socket_connection = None
|
||||||
|
if self._replies:
|
||||||
|
self._replies.destroy()
|
||||||
|
self._replies = None
|
||||||
LOG.debug("Eventloop exited, driver shut down")
|
LOG.debug("Eventloop exited, driver shut down")
|
||||||
|
|
||||||
# The remaining methods are reserved to run from the eventloop thread only!
|
# The remaining methods are reserved to run from the eventloop thread only!
|
||||||
|
@ -396,6 +396,7 @@ mech_list: ${mechs}
|
|||||||
super(TestCyrusAuthentication, self).tearDown()
|
super(TestCyrusAuthentication, self).tearDown()
|
||||||
if self._broker:
|
if self._broker:
|
||||||
self._broker.stop()
|
self._broker.stop()
|
||||||
|
self._broker = None
|
||||||
if self._conf_dir:
|
if self._conf_dir:
|
||||||
shutil.rmtree(self._conf_dir, ignore_errors=True)
|
shutil.rmtree(self._conf_dir, ignore_errors=True)
|
||||||
|
|
||||||
@ -546,12 +547,20 @@ class FakeBroker(threading.Thread):
|
|||||||
self.connection.pn_sasl.server()
|
self.connection.pn_sasl.server()
|
||||||
self.connection.open()
|
self.connection.open()
|
||||||
self.sender_links = set()
|
self.sender_links = set()
|
||||||
|
self.receiver_links = set()
|
||||||
self.closed = False
|
self.closed = False
|
||||||
|
|
||||||
def destroy(self):
|
def destroy(self):
|
||||||
"""Destroy the test connection."""
|
"""Destroy the test connection."""
|
||||||
while self.sender_links:
|
# destroy modifies the set, so make a copy
|
||||||
link = self.sender_links.pop()
|
tmp = self.sender_links.copy()
|
||||||
|
while tmp:
|
||||||
|
link = tmp.pop()
|
||||||
|
link.destroy()
|
||||||
|
# destroy modifies the set, so make a copy
|
||||||
|
tmp = self.receiver_links.copy()
|
||||||
|
while tmp:
|
||||||
|
link = tmp.pop()
|
||||||
link.destroy()
|
link.destroy()
|
||||||
self.connection.destroy()
|
self.connection.destroy()
|
||||||
self.connection = None
|
self.connection = None
|
||||||
@ -622,16 +631,21 @@ class FakeBroker(threading.Thread):
|
|||||||
"""An AMQP sending link."""
|
"""An AMQP sending link."""
|
||||||
def __init__(self, server, conn, handle, src_addr=None):
|
def __init__(self, server, conn, handle, src_addr=None):
|
||||||
self.server = server
|
self.server = server
|
||||||
|
self.conn = conn
|
||||||
cnn = conn.connection
|
cnn = conn.connection
|
||||||
self.link = cnn.accept_sender(handle,
|
self.link = cnn.accept_sender(handle,
|
||||||
source_override=src_addr,
|
source_override=src_addr,
|
||||||
event_handler=self)
|
event_handler=self)
|
||||||
|
conn.sender_links.add(self)
|
||||||
self.link.open()
|
self.link.open()
|
||||||
self.routed = False
|
self.routed = False
|
||||||
|
|
||||||
def destroy(self):
|
def destroy(self):
|
||||||
"""Destroy the link."""
|
"""Destroy the link."""
|
||||||
self._cleanup()
|
self._cleanup()
|
||||||
|
conn = self.conn
|
||||||
|
self.conn = None
|
||||||
|
conn.sender_links.remove(self)
|
||||||
if self.link:
|
if self.link:
|
||||||
self.link.destroy()
|
self.link.destroy()
|
||||||
self.link = None
|
self.link = None
|
||||||
@ -663,21 +677,31 @@ class FakeBroker(threading.Thread):
|
|||||||
"""An AMQP Receiving link."""
|
"""An AMQP Receiving link."""
|
||||||
def __init__(self, server, conn, handle, addr=None):
|
def __init__(self, server, conn, handle, addr=None):
|
||||||
self.server = server
|
self.server = server
|
||||||
|
self.conn = conn
|
||||||
cnn = conn.connection
|
cnn = conn.connection
|
||||||
self.link = cnn.accept_receiver(handle,
|
self.link = cnn.accept_receiver(handle,
|
||||||
target_override=addr,
|
target_override=addr,
|
||||||
event_handler=self)
|
event_handler=self)
|
||||||
|
conn.receiver_links.add(self)
|
||||||
self.link.open()
|
self.link.open()
|
||||||
self.link.add_capacity(10)
|
self.link.add_capacity(10)
|
||||||
|
|
||||||
|
def destroy(self):
|
||||||
|
"""Destroy the link."""
|
||||||
|
conn = self.conn
|
||||||
|
self.conn = None
|
||||||
|
conn.receiver_links.remove(self)
|
||||||
|
if self.link:
|
||||||
|
self.link.destroy()
|
||||||
|
self.link = None
|
||||||
|
|
||||||
# ReceiverEventHandler callbacks:
|
# ReceiverEventHandler callbacks:
|
||||||
|
|
||||||
def receiver_remote_closed(self, receiver_link, error):
|
def receiver_remote_closed(self, receiver_link, error):
|
||||||
self.link.close()
|
self.link.close()
|
||||||
|
|
||||||
def receiver_closed(self, receiver_link):
|
def receiver_closed(self, receiver_link):
|
||||||
self.link.destroy()
|
self.destroy()
|
||||||
self.link = None
|
|
||||||
|
|
||||||
def message_received(self, receiver_link, message, handle):
|
def message_received(self, receiver_link, message, handle):
|
||||||
"""Forward this message out the proper sending link."""
|
"""Forward this message out the proper sending link."""
|
||||||
@ -802,8 +826,11 @@ class FakeBroker(threading.Thread):
|
|||||||
|
|
||||||
# Shutting down
|
# Shutting down
|
||||||
self._my_socket.close()
|
self._my_socket.close()
|
||||||
for conn in self._connections.itervalues():
|
for conn in self._connections.values():
|
||||||
conn.destroy()
|
conn.destroy()
|
||||||
|
self._connections = None
|
||||||
|
self.container.destroy()
|
||||||
|
self.container = None
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
def add_route(self, address, link):
|
def add_route(self, address, link):
|
||||||
|
Loading…
Reference in New Issue
Block a user