From 134cd670ed5c2773eec8ab6774ea0a6bdb2e7528 Mon Sep 17 00:00:00 2001
From: bbouters <bbouters@redhat.com>
Date: Wed, 3 Aug 2016 19:23:28 +0000
Subject: [PATCH] Fixes Qpid file descriptor leak

This is already fixed in upstream Kombu, so this is a
port of the upstream fix to the version we carry.

The Pulp issue is:

https://pulp.plan.io/issues/2124
---
 tests/transport/test_qpid.py | 48 +++++++++++++++-----------------------------
 transport/qpid.py            | 19 +++++++++---------
 2 files changed, 26 insertions(+), 41 deletions(-)

diff --git a/tests/transport/test_qpid.py b/tests/transport/test_qpid.py
index e9272ff..fb7a831 100644
--- a/kombu/tests/transport/test_qpid.py
+++ b/kombu/tests/transport/test_qpid.py
@@ -1427,21 +1427,6 @@ class TestTransportInit(Case):
         Transport(m)
         self.mock_base_Transport__init__.assert_called_once_with(m)
 
-    def test_transport___init___calls_os_pipe(self):
-        Transport(Mock())
-        self.mock_os.pipe.assert_called_once_with()
-
-    def test_transport___init___saves_os_pipe_file_descriptors(self):
-        transport = Transport(Mock())
-        self.assertIs(transport.r, self.mock_r)
-        self.assertIs(transport._w, self.mock_w)
-
-    def test_transport___init___sets_non_blocking_behavior_on_r_fd(self):
-        Transport(Mock())
-        self.mock_fcntl.fcntl.assert_called_once_with(
-            self.mock_r,  self.mock_fcntl.F_SETFL,  self.mock_os.O_NONBLOCK,
-        )
-
 
 @case_no_python3
 @case_no_pypy
@@ -1813,6 +1798,7 @@ class TestTransportOnReadable(Case):
         self.patch_b = patch.object(Transport, 'drain_events')
         self.mock_drain_events = self.patch_b.start()
         self.transport = Transport(Mock())
+        self.transport.register_with_event_loop(Mock(), Mock())
 
     def tearDown(self):
         self.patch_a.stop()
@@ -1904,25 +1890,23 @@ class TestTransport(ExtraAssertionsMixin, Case):
         result_params = my_transport.default_connection_params
         self.assertDictEqual(correct_params, result_params)
 
-    @patch('os.close')
-    def test_del(self, close):
+    @patch(QPID_MODULE + '.os.close')
+    def test_del_sync(self, close):
+        my_transport = Transport(self.mock_client)
+        my_transport.__del__()
+        self.assertFalse(close.called)
+
+    @patch(QPID_MODULE + '.os.close')
+    def test_del_async(self, close):
         my_transport = Transport(self.mock_client)
+        my_transport.register_with_event_loop(Mock(), Mock())
         my_transport.__del__()
-        self.assertEqual(
-            close.call_args_list,
-            [
-                ((my_transport.r,), {}),
-                ((my_transport._w,), {}),
-            ])
-
-    @patch('os.close')
-    def test_del_failed(self, close):
+        self.assertTrue(close.called)
+
+    @patch(QPID_MODULE + '.os.close')
+    def test_del_async_failed(self, close):
         close.side_effect = OSError()
         my_transport = Transport(self.mock_client)
+        my_transport.register_with_event_loop(Mock(), Mock())
         my_transport.__del__()
-        self.assertEqual(
-            close.call_args_list,
-            [
-                ((my_transport.r,), {}),
-                ((my_transport._w,), {}),
-            ])
+        self.assertTrue(close.called)
diff --git a/transport/qpid.py b/transport/qpid.py
index aa0d8e9..a8e78c4 100644
--- a/kombu/transport/qpid.py
+++ b/kombu/transport/qpid.py
@@ -1393,9 +1393,6 @@ class Transport(base.Transport):
         """
         self.verify_runtime_environment()
         super(Transport, self).__init__(*args, **kwargs)
-        self.r, self._w = os.pipe()
-        if fcntl is not None:
-            fcntl.fcntl(self.r, fcntl.F_SETFL, os.O_NONBLOCK)
         self.use_async_interface = False
 
     def verify_runtime_environment(self):
@@ -1522,6 +1519,9 @@ class Transport(base.Transport):
         :param loop: A reference to the external loop.
         :type loop: kombu.async.hub.Hub
         """
+        self.r, self._w = os.pipe()
+        if fcntl is not None:
+            fcntl.fcntl(self.r, fcntl.F_SETFL, os.O_NONBLOCK)
         self.use_async_interface = True
         loop.add_reader(self.r, self.on_readable, connection, loop)
 
@@ -1691,9 +1691,10 @@ class Transport(base.Transport):
         """
         Ensure file descriptors opened in __init__() are closed.
         """
-        for fd in (self.r, self._w):
-            try:
-                os.close(fd)
-            except OSError:
-                # ignored
-                pass
+        if self.use_async_interface:
+            for fd in (self.r, self._w):
+                try:
+                    os.close(fd)
+                except OSError:
+                    # ignored
+                    pass
-- 
2.7.4