diff --git a/docs/examples/asynchronous_consumer_example.rst b/docs/examples/asynchronous_consumer_example.rst
index 2b6f9a7..88cdf5d 100644
--- a/docs/examples/asynchronous_consumer_example.rst
+++ b/docs/examples/asynchronous_consumer_example.rst
@@ -55,8 +55,7 @@ consumer.py::
             """
             LOGGER.info('Connecting to %s', self._url)
             return pika.SelectConnection(pika.URLParameters(self._url),
-                                         self.on_connection_open,
-                                         stop_ioloop_on_close=False)
+                                         self.on_connection_open)
 
         def on_connection_open(self, unused_connection):
             """This method is called by pika once the connection to RabbitMQ has
diff --git a/docs/examples/asynchronous_publisher_example.rst b/docs/examples/asynchronous_publisher_example.rst
index b310146..dac92e6 100644
--- a/docs/examples/asynchronous_publisher_example.rst
+++ b/docs/examples/asynchronous_publisher_example.rst
@@ -55,9 +55,7 @@ publisher.py::
         def connect(self):
             """This method connects to RabbitMQ, returning the connection handle.
             When the connection is established, the on_connection_open method
-            will be invoked by pika. If you want the reconnection to work, make
-            sure you set stop_ioloop_on_close to False, which is not the default
-            behavior of this adapter.
+            will be invoked by pika.
 
             :rtype: pika.SelectConnection
 
@@ -65,8 +63,7 @@ publisher.py::
             LOGGER.info('Connecting to %s', self._url)
             return pika.SelectConnection(pika.URLParameters(self._url),
                                          on_open_callback=self.on_connection_open,
-                                         on_close_callback=self.on_connection_closed,
-                                         stop_ioloop_on_close=False)
+                                         on_close_callback=self.on_connection_closed)
 
         def on_connection_open(self, unused_connection):
             """This method is called by pika once the connection to RabbitMQ has
diff --git a/examples/asynchronous_consumer_example.py b/examples/asynchronous_consumer_example.py
index 67f16bd..1744788 100644
--- a/examples/asynchronous_consumer_example.py
+++ b/examples/asynchronous_consumer_example.py
@@ -52,8 +52,7 @@ class ExampleConsumer(object):
         return pika.SelectConnection(parameters=pika.URLParameters(self._url),
                                      on_open_callback=self.on_connection_open,
                                      on_open_error_callback=self.on_connection_open_error,
-                                     on_close_callback=self.on_connection_closed,
-                                     stop_ioloop_on_close=False)
+                                     on_close_callback=self.on_connection_closed)
 
     def on_connection_open(self, unused_connection):
         """This method is called by pika once the connection to RabbitMQ has
diff --git a/examples/asynchronous_publisher_example.py b/examples/asynchronous_publisher_example.py
index 75143ce..d92e511 100644
--- a/examples/asynchronous_publisher_example.py
+++ b/examples/asynchronous_publisher_example.py
@@ -50,9 +50,7 @@ class ExamplePublisher(object):
     def connect(self):
         """This method connects to RabbitMQ, returning the connection handle.
         When the connection is established, the on_connection_open method
-        will be invoked by pika. If you want the reconnection to work, make
-        sure you set stop_ioloop_on_close to False, which is not the default
-        behavior of this adapter.
+        will be invoked by pika.
 
         :rtype: pika.SelectConnection
 
@@ -61,8 +59,7 @@ class ExamplePublisher(object):
         return pika.SelectConnection(pika.URLParameters(self._url),
                                      on_open_callback=self.on_connection_open,
                                      on_open_error_callback=self.on_connection_open_error,
-                                     on_close_callback=self.on_connection_closed,
-                                     stop_ioloop_on_close=False)
+                                     on_close_callback=self.on_connection_closed)
 
     def on_connection_open(self, unused_connection):
         """This method is called by pika once the connection to RabbitMQ has
diff --git a/pika/adapters/asyncio_connection.py b/pika/adapters/asyncio_connection.py
index dff7656..cc38c3f 100644
--- a/pika/adapters/asyncio_connection.py
+++ b/pika/adapters/asyncio_connection.py
@@ -159,7 +159,6 @@ class AsyncioConnection(base_connection.BaseConnection):
                  on_open_callback=None,
                  on_open_error_callback=None,
                  on_close_callback=None,
-                 stop_ioloop_on_close=False,
                  custom_ioloop=None):
         """ Create a new instance of the AsyncioConnection class, connecting
         to RabbitMQ automatically
@@ -169,7 +168,7 @@ class AsyncioConnection(base_connection.BaseConnection):
         :type on_open_callback: method
         :param on_open_error_callback: Method to call if the connection cant be opened
         :type on_open_error_callback: method
-        :param asyncio.AbstractEventLoop loop: By default asyncio.get_event_loop()
+        :param asyncio.AbstractEventLoop custom_ioloop: By default asyncio.get_event_loop()
 
         """
         self.sleep_counter = 0
@@ -177,10 +176,11 @@ class AsyncioConnection(base_connection.BaseConnection):
         self.ioloop = IOLoopAdapter(self.loop)
 
         super().__init__(
-            parameters, on_open_callback,
+            parameters,
+            on_open_callback,
             on_open_error_callback,
-            on_close_callback, self.ioloop,
-            stop_ioloop_on_close=stop_ioloop_on_close,
+            on_close_callback,
+            self.ioloop
         )
 
     def _adapter_connect(self):
diff --git a/pika/adapters/base_connection.py b/pika/adapters/base_connection.py
index 37dcc43..fdc456b 100644
--- a/pika/adapters/base_connection.py
+++ b/pika/adapters/base_connection.py
@@ -37,8 +37,7 @@ class BaseConnection(connection.Connection):
                  on_open_callback=None,
                  on_open_error_callback=None,
                  on_close_callback=None,
-                 ioloop=None,
-                 stop_ioloop_on_close=True):
+                 ioloop=None):
         """Create a new instance of the Connection object.
 
         :param pika.connection.Parameters parameters: Connection parameters
@@ -48,7 +47,6 @@ class BaseConnection(connection.Connection):
         :param method on_close_callback: Called when the connection is closed:
             on_close_callback(connection, reason_code, reason_text)
         :param object ioloop: IOLoop object to use
-        :param bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
         :raises: RuntimeError
         :raises: ValueError
 
@@ -64,7 +62,6 @@ class BaseConnection(connection.Connection):
         self.event_state = self.base_events
         self.ioloop = ioloop
         self.socket = None
-        self.stop_ioloop_on_close = stop_ioloop_on_close
         self.write_buffer = None
         super(BaseConnection,
               self).__init__(parameters, on_open_callback,
@@ -119,11 +116,7 @@ class BaseConnection(connection.Connection):
         :param str reply_text: The text reason for the close
 
         """
-        try:
-            super(BaseConnection, self).close(reply_code, reply_text)
-        finally:
-            if self.is_closed:
-                self._handle_ioloop_stop()
+        super(BaseConnection, self).close(reply_code, reply_text)
 
     def remove_timeout(self, timeout_id):
         """Remove the timeout from the IOLoop by the ID returned from
@@ -170,10 +163,7 @@ class BaseConnection(connection.Connection):
 
     def _adapter_disconnect(self):
         """Invoked if the connection is being told to disconnect"""
-        try:
-            self._cleanup_socket()
-        finally:
-            self._handle_ioloop_stop()
+        self._cleanup_socket()
 
     def _cleanup_socket(self):
         """Close the socket cleanly"""
@@ -299,14 +289,6 @@ class BaseConnection(connection.Connection):
         # called), etc., etc., etc.
         self._manage_event_state()
 
-    def _handle_ioloop_stop(self):
-        """Invoked when the connection is closed to determine if the IOLoop
-        should be stopped or not.
-
-        """
-        if self.stop_ioloop_on_close and self.ioloop:
-            self.ioloop.stop()
-
     def _handle_error(self, error_value):
         """Internal error handling method. Here we expect a socket error
         coming in and will handle different socket errors differently.
diff --git a/pika/adapters/blocking_connection.py b/pika/adapters/blocking_connection.py
index 8649d2e..2a1629f 100644
--- a/pika/adapters/blocking_connection.py
+++ b/pika/adapters/blocking_connection.py
@@ -369,8 +369,7 @@ class BlockingConnection(object):
             parameters=parameters,
             on_open_callback=self._opened_result.set_value_once,
             on_open_error_callback=self._open_error_result.set_value_once,
-            on_close_callback=self._closed_result.set_value_once,
-            stop_ioloop_on_close=False)
+            on_close_callback=self._closed_result.set_value_once)
 
         self._impl.ioloop.activate_poller()
 
diff --git a/pika/adapters/select_connection.py b/pika/adapters/select_connection.py
index f6e6f3e..3ae3eca 100644
--- a/pika/adapters/select_connection.py
+++ b/pika/adapters/select_connection.py
@@ -71,7 +71,6 @@ class SelectConnection(BaseConnection):
             on_open_callback=None,
             on_open_error_callback=None,
             on_close_callback=None,
-            stop_ioloop_on_close=True,
             custom_ioloop=None):
         """Create a new instance of the Connection object.
 
@@ -81,15 +80,17 @@ class SelectConnection(BaseConnection):
             be established: on_open_error_callback(connection, str|exception)
         :param method on_close_callback: Called when the connection is closed:
             on_close_callback(connection, reason_code, reason_text)
-        :param bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
         :param custom_ioloop: Override using the global IOLoop in Tornado
         :raises: RuntimeError
 
         """
         ioloop = custom_ioloop or IOLoop()
         super(SelectConnection, self).__init__(
-            parameters, on_open_callback, on_open_error_callback,
-            on_close_callback, ioloop, stop_ioloop_on_close)
+            parameters,
+            on_open_callback,
+            on_open_error_callback,
+            on_close_callback,
+            ioloop)
 
     def _adapter_connect(self):
         """Connect to the RabbitMQ broker, returning True on success, False
diff --git a/pika/adapters/tornado_connection.py b/pika/adapters/tornado_connection.py
index 309fb77..3c3c736 100644
--- a/pika/adapters/tornado_connection.py
+++ b/pika/adapters/tornado_connection.py
@@ -9,20 +9,7 @@ LOGGER = logging.getLogger(__name__)
 
 
 class TornadoConnection(base_connection.BaseConnection):
-    """The TornadoConnection runs on the Tornado IOLoop. If you're running the
-    connection in a web app, make sure you set stop_ioloop_on_close to False,
-    which is the default behavior for this adapter, otherwise the web app
-    will stop taking requests.
-
-    :param pika.connection.Parameters parameters: Connection parameters
-    :param on_open_callback: The method to call when the connection is open
-    :type on_open_callback: method
-    :param on_open_error_callback: Method to call if the connection cant
-                                   be opened
-    :type on_open_error_callback: method
-    :param bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
-    :param custom_ioloop: Override using the global IOLoop in Tornado
-
+    """The TornadoConnection runs on the Tornado IOLoop.
     """
 
     def __init__(self,
@@ -30,7 +17,6 @@ class TornadoConnection(base_connection.BaseConnection):
                  on_open_callback=None,
                  on_open_error_callback=None,
                  on_close_callback=None,
-                 stop_ioloop_on_close=False,
                  custom_ioloop=None):
         """Create a new instance of the TornadoConnection class, connecting
         to RabbitMQ automatically
@@ -42,16 +28,16 @@ class TornadoConnection(base_connection.BaseConnection):
             be established: on_open_error_callback(connection, str|exception)
         :param method on_close_callback: Called when the connection is closed:
             on_close_callback(connection, reason_code, reason_text)
-        :param bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
         :param custom_ioloop: Override using the global IOLoop in Tornado
 
         """
         self.sleep_counter = 0
         self.ioloop = custom_ioloop or ioloop.IOLoop.instance()
-        super(TornadoConnection, self).__init__(parameters, on_open_callback,
+        super(TornadoConnection, self).__init__(parameters,
+                                                on_open_callback,
                                                 on_open_error_callback,
-                                                on_close_callback, self.ioloop,
-                                                stop_ioloop_on_close)
+                                                on_close_callback,
+                                                self.ioloop)
 
     def _adapter_connect(self):
         """Connect to the remote socket, adding the socket to the IOLoop if
diff --git a/pika/adapters/twisted_connection.py b/pika/adapters/twisted_connection.py
index 3ce458b..72c7c2e 100644
--- a/pika/adapters/twisted_connection.py
+++ b/pika/adapters/twisted_connection.py
@@ -282,15 +282,13 @@ class TwistedConnection(base_connection.BaseConnection):
                  parameters=None,
                  on_open_callback=None,
                  on_open_error_callback=None,
-                 on_close_callback=None,
-                 stop_ioloop_on_close=False):
+                 on_close_callback=None):
         super(TwistedConnection, self).__init__(
             parameters=parameters,
             on_open_callback=on_open_callback,
             on_open_error_callback=on_open_error_callback,
             on_close_callback=on_close_callback,
-            ioloop=IOLoopReactorAdapter(self, reactor),
-            stop_ioloop_on_close=stop_ioloop_on_close)
+            ioloop=IOLoopReactorAdapter(self, reactor))
 
     def _adapter_connect(self):
         """Connect to the RabbitMQ broker"""
@@ -374,8 +372,7 @@ class TwistedProtocolConnection(base_connection.BaseConnection):
             on_open_callback=self.connectionReady,
             on_open_error_callback=self.connectionFailed,
             on_close_callback=on_close_callback,
-            ioloop=IOLoopReactorAdapter(self, reactor),
-            stop_ioloop_on_close=False)
+            ioloop=IOLoopReactorAdapter(self, reactor))
 
     def connect(self):
         # The connection is open asynchronously by Twisted, so skip the whole

Remove stop_ioloop_on_close arg from adapter constructors for the 1.0 release
Some time ago, the `stop_ioloop_on_close=True` constructor arg slipped into all the asynchronous adapters. I think that the presence of this arg is overly-opinionated. An asynchronous app is likely to have other logic besides a single pika connection running on the same IOLoop (tornado, asyncio, etc.), so having ioloop-stopping logic in pika adapters is too pika-centric, unnecessary, non-orthogonal, and likely error-prone. Especially when it's brain-dead easy for the app to pass `on_close_callback`- having the signature `on_close_callback(connection, reason_code, reason_text)` - to the adapter constructor and call `connection.ioloop.stop()` itself if needed.

I think we should remove the `stop_ioloop_on_close` args everywhere in pika as well as any logic dealing with stopping the ioloop in pika adapters.
diff --git a/tests/unit/blocking_connection_tests.py b/tests/unit/blocking_connection_tests.py
index 118aa2d..69a80e5 100644
--- a/tests/unit/blocking_connection_tests.py
+++ b/tests/unit/blocking_connection_tests.py
@@ -44,8 +44,7 @@ class BlockingConnectionTests(unittest.TestCase):
             parameters='params',
             on_open_callback=mock.ANY,
             on_open_error_callback=mock.ANY,
-            on_close_callback=mock.ANY,
-            stop_ioloop_on_close=mock.ANY)
+            on_close_callback=mock.ANY)
 
         self.assertEqual(connection._impl.ioloop.activate_poller.call_count, 1)
 
diff --git a/tests/unit/tornado_tests.py b/tests/unit/tornado_tests.py
index de78c60..7c36346 100644
--- a/tests/unit/tornado_tests.py
+++ b/tests/unit/tornado_tests.py
@@ -15,4 +15,4 @@ class TornadoConnectionTests(unittest.TestCase):
         obj = tornado_connection.TornadoConnection()
         mock_init.assert_called_once_with(
             None, None, None, None,
-            tornado_connection.ioloop.IOLoop.instance(), False)
+            tornado_connection.ioloop.IOLoop.instance())

pip install -e .[dev]
parse_log_pytest
requirements.txt
pytest pytest-asyncio
apt-get update apt-get install -y gcc apt-get install -y rabbitmq-server
3.9
test-requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
certifi==2025.1.31
charset-normalizer==3.4.1
codecov==2.1.13
coverage==7.8.0
exceptiongroup==1.2.2
idna==3.10
iniconfig==2.1.0
mock==5.2.0
nose==1.3.7
packaging==24.2
-e git+https://github.com/pika/pika.git@0e13941b26bfa2f5a074b6ee968b180bb5a13724#egg=pika
pluggy==1.5.0
pytest==8.3.5
pytest-asyncio==0.26.0
requests==2.32.3
tomli==2.2.1
tornado==6.4.2
Twisted==15.3.0
typing_extensions==4.13.0
urllib3==2.3.0
zope.interface==7.2

name: pika
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tzdata=2025a=h04d1e81_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - certifi==2025.1.31
      - charset-normalizer==3.4.1
      - codecov==2.1.13
      - coverage==7.8.0
      - exceptiongroup==1.2.2
      - idna==3.10
      - iniconfig==2.1.0
      - mock==5.2.0
      - nose==1.3.7
      - packaging==24.2
      - pluggy==1.5.0
      - pytest==8.3.5
      - pytest-asyncio==0.26.0
      - requests==2.32.3
      - tomli==2.2.1
      - tornado==6.4.2
      - twisted==15.3.0
      - typing-extensions==4.13.0
      - urllib3==2.3.0
      - zope-interface==7.2
prefix: /opt/conda/envs/pika

tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_constructor tests/unit/tornado_tests.py::TornadoConnectionTests::test_tornado_connection_call_parent
tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_channel tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_close_with_channel_closed_exception tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_connection_attempts_with_timeout tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output_server_initiated_error_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output_server_initiated_no_error_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output_user_initiated_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_process_io_for_connection_setup tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_process_io_for_connection_setup_fails_with_open_error tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_sleep
pika/adapters/blocking_connection.py pika/adapters/asyncio_connection.py pika/adapters/select_connection.py pika/adapters/tornado_connection.py pika/adapters/twisted_connection.py examples/asynchronous_publisher_example.py examples/asynchronous_consumer_example.py docs/examples/asynchronous_publisher_example.rst pika/adapters/base_connection.py docs/examples/asynchronous_consumer_example.rst
pika/adapters/blocking_connection.py pika/adapters/asyncio_connection.py pika/adapters/select_connection.py pika/adapters/tornado_connection.py pika/adapters/twisted_connection.py examples/asynchronous_publisher_example.py examples/asynchronous_consumer_example.py docs/examples/asynchronous_publisher_example.rst pika/adapters/base_connection.py docs/examples/asynchronous_consumer_example.rst