Pika 2

AMQP10-CLIENT-AUDIT.md

rabbitmq-amqp-python-client v0.7.0 - Code Audit

Audit of the official RabbitMQ AMQP 1.0 Python client. Performed 2026-06-01 against the local checkout.

Repository: https://github.com/rabbitmq/rabbitmq-amqp-python-client

Overview

• Python 3.10+, built on python-qpid-proton (C extension for AMQP 1.0)
• Sync + async API (async wraps sync via run_in_executor)
• Poetry-based build, no unit tests (integration tests only, require live broker)
• Vendors a modified fork of the qpid/proton Python layer (24 files)

---

Critical (correctness bugs, data loss risks, race conditions)

1. Consumer reconnection leaks a receiver link

consumer.py:85 - _update_connection() creates a fully-attached receiver via self._receiver = self._create_receiver(addr), then immediately overwrites self._receiver at lines 89/98/109 with a new one. The first receiver's AMQP link is never closed, leaking a link on the broker and potentially stealing messages during the brief window it exists.

2. Stream consumer reconnection crashes

consumer.py:97 - self._consumer_options.offset(self._handler.offset - 1) calls a nonexistent method. StreamConsumerOptions only has _offset() (private). Stream consumers cannot reconnect after disconnection - this is a crash bug (AttributeError) in the recovery path.

3. Environment.close() mutates list during iteration

environment.py:113 - Iterates self._connections directly while connection.close() (connection.py:359) removes the connection from the same list. This skips connections (every other one) or raises RuntimeError on some Python versions. Compare with connection.py:349 which correctly iterates a copy (self._publishers[:]).

4. Exchange-to-exchange unbind uses wrong address prefix

address_helper.py:151 - binding_path_with_exchange_exchange uses dstq= (destination queue) instead of dste= (destination exchange). Exchange-to-exchange bindings cannot be removed via the returned path.

5. Custom exceptions inherit BaseException instead of Exception

exceptions.py:1,10 - ValidationCodeException and ArgumentOutOfRangeException inherit from BaseException. They escape standard except Exception: handlers, propagating through cleanup code and surprising every user. Library exceptions must inherit Exception.

6. Async path omits 4.3.0 version check

asyncio/connection.py:256-262 - The async path passes versions 4.0.0 through 4.2.0 but omits 4.3.0. QuorumConsumerOptions.validate() checks versions.get("4.3.0", False) which will always be False in the async path. SAC state handler validation incorrectly rejects on any server version when using the async API.

---

Important (resource leaks, poor error handling, API foot-guns)

7. Reconnection handler blocks the reactor event loop

qpid/proton/_utils.py:670 - event.container.schedule(0, self._on_disconnection_handler()) - the parentheses call the handler immediately. Connection._on_disconnection calls time.sleep() in a loop. This blocks the reactor's process() call for potentially minutes. The schedule(0, None) after is a no-op. Then line 672 still raises ConnectionClosed, which propagates to the user even though reconnection may have succeeded.

8. Connection.close() does not close Management links

connection.py:348-353 - Publishers and consumers are closed, but self._managements links are never closed. If the user calls connection.close() without first calling management.close(), the management sender/receiver links leak.

9. 409 response code handling is dead code

management.py:413-414 - _validate_reponse_code unconditionally raises on 409 before the loop that checks expected codes. Both declare_exchange (line 207) and declare_queue (line 252) include 409 in their expected codes (suggesting idempotent declaration was intended), but it always throws.

10. No thread safety in synchronous API

Connection, Publisher, Consumer, and Management have no locking. Environment docstring mentions "Connection is not thread-safe" (line 81) in a buried comment, but there are no protections. Sharing a connection across threads corrupts internal state. The async wrapper correctly uses asyncio.Lock, but the sync API is completely unprotected.

11. Converter.bytes_to_string corrupts multi-byte UTF-8

utils.py:25 - "".join(map(chr, body)) treats each byte as a Unicode code point. For UTF-8 encoded text with non-ASCII characters, this produces garbage. Should use body.decode('utf-8').

12. AsyncPublisher.open() misleading comment about lock safety

asyncio/publisher.py:78-79 - Comment says "We don't need the lock here because Publisher.init doesn't send any network traffic". This is false - Publisher.__init__ calls _open() which calls _create_sender() which performs AMQP ATTACH. The code is safe only because AsyncConnection.publisher() already holds the lock, but the comment is dangerously wrong for anyone refactoring.

13. No context manager support for sync Connection/Publisher/Consumer

Only Environment supports with syntax. Users must manually manage close/cleanup for every other resource, making leaks easy. The async API does better with __aenter__/__aexit__ on several classes.

14. OAuth2 user hardcoded to "no"

connection.py:147 - Comment says "normally in case of oauth user should be '' but the internal library gives error". Hardcoding "no" as a username is a fragile undocumented workaround.

---

Minor (style issues, missing features, code smell)

15. RecoveryConfiguration.MaxReconnectAttempts uses PascalCase

entities.py:551 - Mixed naming: active_recovery and back_off_reconnect_interval are snake_case, but MaxReconnectAttempts is PascalCase. Violates PEP 8.

16. Typo: _validate_reponse_code (missing 's' in 'response')

management.py:410

17. Vendored qpid/proton code is a forked copy

The qpid/ directory contains 24 Python files that import cproton (the C extension from python-qpid-proton). The project declares python-qpid-proton = "^0.40.0" as a dependency AND vendors a modified copy of the Python layer. Modifications include _flow_properties.py (SAC support), _utils.py (disconnection handler), and _io.py.

This creates version-skew risk: if python-qpid-proton upgrades its C layer but the vendored Python is stale, things break silently. It's also a significant maintenance burden.

18. SenderOptionUnseattle class name is a typo

options.py:34 - Should be SenderOptionUnsettled.

19. Hardcoded defaults scattered across code

• Default credit of 10 in consumer.py:80,183 with no named constant
• Reconnect max delay hardcoded to timedelta(minutes=1) in connection.py:435
• Default timeout of 60 seconds in _utils.py:445
• Should be configurable or at least named constants

20. Test coverage gaps

All tests require a live RabbitMQ server - no unit tests (except test_server_validation.py which mocks). No tests for: concurrent access, error recovery paths (stream reconnection), exchange-to-exchange bindings, OAuth token refresh, edge cases in encode_path_segment, async consumer stop_processing path.

21. Publisher destination uses empty string as sentinel instead of None

connection.py:362, publisher.py passim - Mixed use of None vs "" as "no value". Consumer uses Optional[str] properly while publisher uses str = "".

22. Connection._index field declared but never used

connection.py:102

23. Exception classes don't call super().init()

exceptions.py - Custom exceptions store self.msg but don't pass it to BaseException.__init__(), so str(exc) and args attribute don't work as expected.

---

Architectural observations

1. The vendored proton fork is the biggest long-term risk. Maintaining 24 modified files from an upstream C-extension library is fragile. Changes in proton's C layer can silently break the vendored Python.

2. The sync API's threading model is fundamentally limited. The qpid-proton reactor is single-threaded; the sync API blocks on it. The async API wraps sync calls in run_in_executor, which is correct but means the async API's concurrency is limited to one operation at a time (serialized by the executor + lock). True async would require a native asyncio transport.

3. Recovery is the weakest area. The reconnection logic blocks the reactor, leaks links, crashes for streams, and has an unclear contract about what state is preserved vs. lost.

4. No unit tests means regressions are caught late. Every test requires a running broker, so CI is slow and failures are hard to reproduce locally.

API-DESIGN.md

Pika v.next API Design

Architecture: three layers

Layer 1: Protocol engines (internal, hidden)

Two protocol implementations, neither user-facing:

• pika._amqp091 - what exists today (SelectConnection, Channel, spec, frame)
• pika._amqp10 - forked from Azure _pyamqp (Connection, Session, Link, codec)

Users never import from these. They are the wire-format machinery.

Layer 2: The "pit of success" API (user-facing, new)

env = pika.Environment("amqp://localhost")
pub = env.publisher("/exchanges/my-exchange/my-key")
pub.publish(b"hello")

con = env.consumer("/queues/my-queue")
for msg in con:
    msg.ack()

Protocol auto-negotiated or explicit. Safe defaults baked in (confirms on, bounded outstanding, auto-recovery, heartbeats non-optional). This is the target for new code and the recommended API going forward.

Layer 3: ThreadSafeConnection compatibility (user-facing, transitional)

# This still works - same API as Pika 1.5
conn = pika.ThreadSafeConnection(pika.ConnectionParameters("localhost"))
ch = conn.channel()
ch.queue_declare(queue="my-queue")
ch.basic_publish(exchange="", routing_key="my-queue", body=b"hello")
conn.close()

Migration bridge. Existing Pika 1.x code runs unchanged. Users migrate at their own pace. Internally ThreadSafeConnection could eventually be re-implemented on top of the protocol engine, or just stay as-is for 0-9-1-only users.

Migration path from Pika 1.x

Step	What changes	Effort
1. Upgrade to Pika 2.0	Nothing breaks. ThreadSafeConnection is the default instead of BlockingConnection. Existing code works.	Zero (just upgrade)
2. Adopt Environment API (optional)	Simpler code, gains auto-recovery, bounded confirms, safe defaults. Still 0-9-1 under the hood.	Low (rewrite connection setup, simplify pub/sub)
3. Switch to AMQP 1.0 (optional)	Change the URL scheme or add protocol="amqp10". Gain 1.0 features. Topology declarations move to the management interface.	Medium (address format changes, confirm callback signature changes slightly)

Operation mapping across APIs

Pika 1.x (ThreadSafe)	Environment API	Under the hood (0-9-1)	Under the hood (1.0)
ch.basic_publish(exchange, rk, body)	pub.publish(body)	basic_publish frame	transfer on sender link
ch.basic_consume(queue, callback)	env.consumer(queue)	Basic.Consume + deliveries	attach receiver link + grant credit
ch.basic_ack(tag)	msg.ack()	Basic.Ack	disposition(accepted)
ch.basic_nack(tag, requeue=True)	msg.nack()	Basic.Nack(requeue=True)	disposition(released)
ch.basic_nack(tag, requeue=False)	msg.reject()	Basic.Nack(requeue=False)	disposition(rejected)
ch.confirm_delivery(cb)	default behavior	Confirm.Select	unsettled sender settle mode
ch.queue_declare(queue)	env.management.declare_queue(...)	Queue.Declare	management link request
ch.basic_qos(prefetch_count=N)	env.consumer(queue, prefetch=N)	Basic.Qos	link credit = N

Protocol-specific features (escape hatches)

These are available as kwargs or methods that raise NotImplementedError if used with the wrong protocol:

AMQP 1.0 only

• msg.modify(annotations={...}) - requeue with altered message annotations
• consumer = env.consumer(queue, filter_expression="color = 'red'") - server-side filtering
• Explicit link credit control (consumer.grant_credit(n))
• Mixed sender settle mode (per-message confirm decisions)
• Queue locality (publish/consume local to reduce intra-cluster traffic)

AMQP 0-9-1 only

• exchange_declare / exchange_bind as protocol operations (1.0 uses management link)
• Transactions (tx_select / tx_commit / tx_rollback)
• Channel interceptor plugins (sharding)

Safe defaults (rmqcpp philosophy)

Behavior	Default	Opt-out
Publisher confirms	Always on	publisher(..., confirms=False) for fire-and-forget
Bounded outstanding confirms	1000 (blocks on full)	publisher(..., max_unconfirmed=N)
Auto-recovery	Always on (topology redeclared)	Environment(..., recovery=False)
Heartbeats	Negotiated per spec, non-optional	Cannot disable
Consumer ack required	Always (no auto-ack)	consumer(..., auto_ack=True)
Separate pub/sub connections	Default (one connection per role)	Environment(..., shared_connection=True)
Message durability	Persistent delivery mode	pub.publish(body, persistent=False)

Message guard pattern (inspired by rmqcpp)

for msg in consumer:
    # If this block raises or the loop breaks without calling
    # msg.ack()/msg.nack()/msg.reject(), the message is automatically
    # nacked (requeued) when msg goes out of scope.
    process(msg.body)
    msg.ack()

Implemented via __del__ or a context manager:

for msg in consumer:
    with msg:  # auto-nacks on exception
        process(msg.body)
        msg.ack()

BlockingConnection vs ThreadSafeConnection

Fundamental architecture

Aspect	BlockingConnection	ThreadSafeConnection
Threading model	Single-threaded. The calling thread IS the IOLoop. Every operation polls the socket in a busy loop until the response arrives.	Two threads. Dedicated IOLoop thread handles all I/O. Calling threads submit work and wait on Events.
How "blocking" works	_flush_output() calls ioloop.poll() + ioloop.process_timeouts() in a while loop until the waiter fires	threading.Event.wait(timeout) - OS-level sleep, zero CPU
Heartbeat safety	Fragile. If user code blocks (DB query, HTTP call, sleep), heartbeats stop because the IOLoop isn't being polled. Users must call process_data_events() periodically.	Safe. Heartbeats run on the IOLoop thread regardless of what user code does.
Thread safety	None. Single-threaded by design. Calling from multiple threads corrupts state.	Full. Any method can be called from any thread. All operations route through add_callback_threadsafe.
Consumer callbacks	Run inline during process_data_events() / start_consuming(). A slow callback stalls heartbeats and all other I/O.	Run on a per-channel worker thread. IOLoop is never blocked by user code.
Reentrancy	Explicitly forbidden. start_consuming() raises ReentrancyError if called from a callback.	Not an issue. Callbacks run on a different thread from the IOLoop.

Publish with confirms

Aspect	BlockingConnection	ThreadSafeConnection
Mechanism	basic_publish polls until Basic.Ack arrives. Blocks both calling thread AND the IOLoop until the broker confirms.	Schedules work on the IOLoop thread and returns immediately. Confirm arrives asynchronously on the worker thread.
Throughput	One message in-flight at a time (publish, wait for confirm, repeat).	Many messages in-flight simultaneously. Soak validated 800 msg/s sustained across 8 threads with sub-1ms p99 latency.
Error reporting	Raises NackError or UnroutableError synchronously from basic_publish. Simple but slow.	Nack/return reported via ack_nack_callback on the worker thread. More complex but allows pipelining.

Consumer patterns

Aspect	BlockingConnection	ThreadSafeConnection
Slow consumer	Stalls heartbeats. Broker may kill the connection. Users must be careful about processing time.	Cannot stall heartbeats. User code runs on the worker thread; IOLoop thread is independent.
process_data_events()	Required. Long-lived connections must call this periodically or heartbeats timeout. The #1 footgun in pika.	Not needed. IOLoop thread handles heartbeats automatically.
time.sleep()	Dangerous. Stalls heartbeats. Must use connection.sleep() instead.	Safe. time.sleep() in user code is fine because heartbeats are on another thread.

Feature gap analysis

What BlockingConnection has that ThreadSafe doesn't (yet):*

• consume() generator interface (iterate messages with for)
• start_consuming() / stop_consuming() pattern
• Synchronous basic_publish that returns only after confirm
• basic_recover
• tx_select / tx_commit / tx_rollback (transactions)
• exchange_unbind
• get_waiting_message_count()
• call_later() / remove_timeout() timer API
• update_secret() for auth token refresh

What ThreadSafe has that BlockingConnection doesn't:*

• Thread safety
• Heartbeat safety (cannot stall)
• on_publish callback with delivery tag at write time
• next_publish_seq_no property
• Idempotent confirm_delivery
• add_on_cancel_callback (server-initiated consumer cancel)
• add_on_return_callback (mandatory message returns)
• add_on_connection_blocked_callback / unblocked (on worker thread)
• Per-channel worker thread isolation
• abort() on both connection and channel
• Context manager support

Migration assessment

The missing pieces in ThreadSafe* for BlockingConnection users fall into two categories:

1. Easy to add: exchange_unbind, basic_recover, update_secret, get_waiting_message_count - thin RPC wrappers following the existing _blocking_rpc pattern.

2. Better redesigned in the Environment API: consume() generator, start_consuming/stop_consuming, synchronous confirms, transactions. These patterns assume the calling thread is the IOLoop; the ThreadSafe model obsoletes them in favor of the iterator consumer and async confirms.

How much to hide: the spectrum

Level	What's exposed	Who it's for
Environment API	publish, consume, ack, topology management	95% of users. Protocol is invisible.
Protocol-specific kwargs	filter expressions, modified outcome, transactions	Users who need a specific 1.0 or 0-9-1 feature
ThreadSafeConnection (legacy)	Full AMQP 0-9-1 channel model	Existing pika users migrating gradually
Protocol engine internals	Frames, links, sessions, codecs	Library developers, debugging, extension

Open design questions

1. Protocol selection: Auto-negotiate based on broker capabilities? Explicit in URL (amqp:// vs amqp10://)? Explicit kwarg (protocol="amqp10")? Prefer 1.0 and fall back to 0-9-1?

2. Deprecation timeline: When (if ever) does ThreadSafeConnection get removed? rmqcpp would say "one API, no legacy." Pika's installed base argues for a long tail. Compromise: deprecation warning in 3.0, removal in 4.0?

3. Fire-and-forget: rmqcpp doesn't support it. Should we? High-throughput logging is a legitimate use case. Option: publisher(..., confirms=False) that explicitly opts out.

4. Multiple consumers per connection: In 0-9-1 this is natural (multiple channels). In 1.0 this is natural (multiple links on a session). The Environment API should make this easy without exposing channels/sessions.

5. Async API: Should Environment be sync-only with an AsyncEnvironment parallel, or should the base API be async with sync wrappers? Azure went sync-first with async wrappers. Modern Python argues async-first. Pika's user base is heavily sync.

6. Naming: Is pika still the right name for a dual-protocol client? Or does this become a new project that subsumes pika?

ASYNC-ARCHITECTURE.md

Async Architecture Options

Why adapters existed

In 2011-2016, Python had no standard async I/O story. If your application ran on Tornado's IOLoop, your AMQP client had to integrate into it. Same for Twisted's reactor. The constraints were:

1. One process, one event loop. You can't run two event loops in the same thread.
2. Threads were considered harmful. The Python community in 2011 was skeptical of threads (GIL, complexity, "shared mutable state is evil"). The convention was single-threaded callbacks.
3. No asyncio. It arrived in Python 3.4 (2014). Before that, Twisted, Tornado, and gevent were incompatible ecosystems.

What changed:

• asyncio became the standard. Tornado adopted it as its underlying loop (5.0, 2018). Twisted added asyncio reactor support.
• Threads became acceptable again. concurrent.futures (3.2), asyncio.to_thread (3.9), and the realization that the GIL doesn't matter for I/O-bound work.
• The "one event loop" constraint dissolved. A background thread running its own event loop doesn't conflict with the application's event loop.

The adapter zoo solved a problem that no longer exists. ThreadSafeConnection proves it: run SelectConnection's IOLoop on a background thread, dispatch via add_callback_threadsafe. The application can be Tornado, asyncio, Django, Flask, or plain synchronous - it doesn't matter.

Three approaches to async

Approach 1: Full duplication (Azure _pyamqp)

Azure wrote the protocol twice - once sync, once async:

Layer	Sync	Async
Transport	_transport.py (772 lines, blocking sockets)	_transport_async.py (539 lines, asyncio.open_connection)
Connection	_connection.py (886 lines)	_connection_async.py (879 lines)
Session	session.py (446 lines)	_session_async.py (448 lines)
Sender	sender.py (195 lines)	_sender_async.py (195 lines)
Receiver	receiver.py (144 lines)	_receiver_async.py (144 lines)
Client	client.py (1103 lines)	_client_async.py (991 lines)
Total	~3,500 lines	~3,200 lines

Shared (no I/O, no duplication): performatives, codec (_encode.py/_decode.py), constants, types, error, outcomes, message, endpoints - about 3,500 lines.

Pro: True async concurrency. Multiple sessions/links multiplexed on one connection, all driven by the event loop without thread boundaries. Native asyncio primitives (asyncio.Lock, asyncio.wait_for, StreamReader/StreamWriter).

Con: ~3,200 lines of duplicated logic to maintain. Every bug fix, protocol change, or feature must be applied twice. The two implementations can (and do) drift.

Approach 2: run_in_executor wrapper (RabbitMQ AMQP 1.0 Python client)

One sync implementation. The async layer wraps every blocking call:

async def publish(self, message):
    async with self._lock:
        await asyncio.get_event_loop().run_in_executor(
            None, self._sync_publisher.publish, message)

Pro: Zero code duplication. One implementation to maintain.

Con: True concurrency is serialized through the executor + lock. You can't do multiple I/O operations simultaneously on the same connection. Every "async" call actually blocks a thread pool thread. It's async in API shape only - not in execution model.

Approach 3: Dedicated IOLoop thread + cross-thread futures (ThreadSafeConnection model)

One event-loop-based implementation running on a dedicated background thread. The async surface submits work via add_callback_threadsafe and awaits an asyncio.Future that gets resolved by a callback from the IOLoop thread:

# Conceptual - not actual implementation
async def publish(self, body):
    future = asyncio.get_event_loop().create_future()
    def _on_ioloop():
        self._sender.send(body)
        # When confirm arrives:
        loop.call_soon_threadsafe(future.set_result, delivery_tag)
    self._ioloop.add_callback_threadsafe(_on_ioloop)
    return await future

Pro: One implementation (the IOLoop-thread protocol engine). No code duplication. Real concurrency (the IOLoop thread handles multiplexing). Sync and async surfaces are thin - just different ways to submit work and wait for results. This is proven to work at scale (ThreadSafeConnection with 11.5M messages over 4 hours, zero errors).

Con: Cross-thread coordination adds latency (thread wakeup, context switch). Not zero-copy. The IOLoop thread is a single point of serialization (though this is true of any single-connection AMQP client regardless of approach).

Recommendation for Pika v.next

Approach 3 - one protocol engine on a dedicated thread, with both sync and async surfaces as thin submission layers.

Rationale:

1. Already proven. ThreadSafeConnection validates this model for AMQP 0-9-1. The same architecture works for AMQP 1.0.
2. No duplication. The protocol state machine (connection, session, link, flow control) is written once. Bug fixes apply once.
3. Sync surface is trivial. Block on an Event or Future.result(). This is what ThreadSafeConnection already does.
4. Async surface is thin. Bridge asyncio.Future to the IOLoop thread's callback. Maybe 100-200 lines of glue per major operation category.
5. Performance is sufficient. The cross-thread latency is microseconds. AMQP operations are network-bound (milliseconds). The overhead is invisible.
6. Eliminates the adapter zoo. Any Python application, regardless of its concurrency framework, uses the same client. No Tornado adapter, no Twisted adapter, no asyncio adapter. One implementation, two surfaces.

The only case where full duplication (Approach 1) wins is extreme high-throughput with many multiplexed links on a single connection where cross-thread wakeup overhead dominates. In practice, for a RabbitMQ client, this doesn't arise - the broker is the bottleneck long before the client's thread-crossing overhead matters.

What this looks like in practice

┌─────────────────────────────────────────────────────┐
│  User code (any thread, any framework)              │
├─────────────────────────────────────────────────────┤
│  Sync surface          │  Async surface             │
│  (Event.wait)          │  (asyncio.Future)          │
├─────────────────────────────────────────────────────┤
│  add_callback_threadsafe (cross-thread boundary)    │
├─────────────────────────────────────────────────────┤
│  Protocol engine (single IOLoop thread)             │
│  - AMQP 0-9-1 (SelectConnection internals)         │
│  - AMQP 1.0 (forked from Azure _pyamqp sync)       │
│  - Frame encode/decode                              │
│  - Connection/Session/Link state machines           │
│  - Flow control, heartbeats                         │
├─────────────────────────────────────────────────────┤
│  Socket I/O (owned by IOLoop thread)                │
└─────────────────────────────────────────────────────┘

Both protocols share the same architecture. The protocol engine is pluggable - selected at connection time based on what the broker supports or what the user requests.

PIKA2.md

Pika 2.0 Design Notes

Private working document. Not for commit until we're ready to go public.

Guiding principles

• Thread-safe by default. Users should not need to understand IOLoop internals to publish and consume safely from multiple threads.
• Fewer adapters, fewer choices, fewer footguns.
• Modern Python only (3.10+ minimum gives us union types, match statements, and structural pattern matching without future imports).
• Align behavior with the RabbitMQ Java and .NET clients where it makes sense for a Python library.

Major changes under consideration

1. ThreadSafeConnection becomes the primary API

• Rename to Connection (or keep ThreadSafeConnection as an alias).
• SelectConnection becomes an internal implementation detail, not a public class users instantiate.
• BlockingConnection is removed or becomes a thin compatibility shim that delegates to the new Connection.
• AsyncioConnection / TornadoConnection - evaluate whether to keep, deprecate, or replace with a native asyncio adapter on top of the same IOLoop-thread model.

2. Drop legacy adapters

• Remove TornadoConnection (Tornado is effectively dead).
• Evaluate whether AsyncioConnection stays as a first-class adapter or whether ThreadSafeConnection + asyncio.to_thread is sufficient.

3. Minimum Python version: 3.10+

• Removes need for from __future__ import annotations everywhere.
• Enables X | Y union syntax, match statements, ParamSpec, etc.
• Drops maintenance burden for 3.7/3.8/3.9 edge cases.

4. Publisher confirms: built-in tracking

• Internal unconfirmed set (like Java's unconfirmedSet).
• wait_for_confirms(timeout) blocks until all outstanding publishes are confirmed or nacked.
• wait_for_confirms_or_die(timeout) - closes the channel on nack.
• Keeps the on_publish callback and ack_nack_callback for users who want async-style notification.

5. Connection recovery / reconnect

• Java and .NET clients have built-in auto-recovery (topology redeclaration, consumer re-registration).
• Pika has never provided this - users must implement their own retry loop.
• Evaluate: should Pika 2 include auto-recovery, or is that better left to a higher-level library?

6. Type annotations everywhere

• Full py.typed marker, strict mypy compliance.
• Public API fully annotated; internal code annotated where it aids comprehension.

7. API cleanup

• Remove deprecated methods and parameters.
• Consistent naming (basic_publish not publish, etc. - already done).
• Consider whether Channel should be the user-facing class name (currently ThreadSafeChannel).

Open questions

• Do we maintain a pika 1.x branch for security fixes, or is 2.0 a clean break?
• What's the migration story for existing BlockingConnection users? (Likely: Connection is a drop-in replacement for most patterns.)
• Should Connection support a with statement that also opens a channel? (e.g., with Connection(params) as conn, conn.channel() as ch:)
• How much of the AMQP 0-9-1 spec do we want to expose vs. abstract? (e.g., do users need exchange_bind or is that power-user territory?)

Timeline

• Pika 1.5.0: ships ThreadSafeConnection as a new adapter alongside existing adapters. No breaking changes.
• Pika 2.0: ThreadSafeConnection becomes the default. Breaking changes are acceptable.

GitHub milestone issues (2.0.0) - categorized

Already addressed by ThreadSafe* in 1.5.0

• #1240 ChannelWrongStateError on publish. ThreadSafe* raises the actual close reason immediately via _check_not_closed().
• #1390 SelectConnection catches all errors, hides details. Users no longer touch SelectConnection directly.
• #1197 Performance. Threading model eliminates BlockingConnection's process_data_events polling overhead.

Must-do for 2.0

• #1600 Bound work queues. Replace ThreadPoolExecutor with bounded queue.Queue(maxsize=1000). Blocking enqueue on the IOLoop thread is the back-pressure mechanism (stops reading frames, TCP window fills, broker slows down). Critical for production safety under slow consumers.
• #1456 Heartbeat negotiation. Must use min(client, server) per AMQP 0-9-1 spec section 2.3.3, matching Java/.NET behavior.
• #1208 Don't cancel consumers before channel close. RabbitMQ cleans them up; extra Basic.Cancel frames broke guarantees in RabbitMQ 3.8+.
• #1389 Remove NACK during stop_consuming. Same spirit as #1208.
• #1043 Replace backpressure detection heuristic. With ThreadSafe*, the bounded queue (#1600) is the natural mechanism. Remove the old heuristic entirely.
• #601 Table encoder signed/unsigned mixup. Correctness bug.
• #599 Decoded floats cast to long ints. Correctness bug.

Bigger architectural decisions

• #645 Move to pamqp for core AMQP work. Replaces pika.spec, pika.frame, pika.data with a maintained external library. High value (less code to maintain, better-tested codec) but high effort.
• #1339 Auto-reconnect / connection recovery. Java/.NET have this built in. Decide: in-scope for 2.0 or defer to 2.1+?

Drop or defer

• #1282 Twisted adapter AlreadyCalledError. Moot if Twisted adapter is dropped.
• #1407 GeventConnection on Windows. Already closed. Gevent adapter should be dropped.
• #1413 TLS/X509 docs update. Docs task, not blocked on 2.0 code.
• #1189 Better TLS logging. Nice-to-have, straightforward.
• #1363 Release checklist. Mostly done (CI lint, pyproject.toml, examples).

AMQP 1.0 landscape

Protocol direction

RabbitMQ is investing in AMQP 1.0 as the future protocol (native since RabbitMQ 4.0, ISO/OASIS standard). New features (fine-grained flow control, filter expressions, queue locality, modified outcome, mixed settle mode) are AMQP 1.0-only. AMQP 0-9-1 is not deprecated but receives no new features.

Python AMQP 1.0 client libraries

Library	Type	Status	Notes
python-qpid-proton	C wrapper (libqpid-proton)	Maintained (v0.40.0)	Apache Qpid project. Requires compiling C extension.
rabbitmq-amqp-python-client	Wraps qpid-proton + vendored Python fork	Active (v0.7.0)	Official RabbitMQ client. Significant bugs in recovery path. See AMQP10-CLIENT-AUDIT.md.
uamqp (Azure)	C wrapper (Cython + vendored C libs)	Deprecated Q1 2025	Replaced by pure Python internal implementation.
Azure _pyamqp	Pure Python	Active, production	Internal to azure-eventhub / azure-servicebus. ~25 modules. Complete AMQP 1.0 stack (connection, session, link, codec, SASL, TLS, async). MIT licensed. Not published as standalone package.
amqp10-codec	Pure Python	Abandoned (v0.1.7)	Codec only, not a client.

Key observation

There is no standalone, maintained, pure Python AMQP 1.0 client library on PyPI today. Azure has a complete one but keeps it internal (_pyamqp underscore prefix). The RabbitMQ team's Python client is built on a C dependency and maintains a fragile 24-file fork of the Python binding layer.

Bloomberg rmqcpp - design model

Bloomberg's rmqcpp (Apache-2.0, C++) is a "pit of success" AMQP 0-9-1 client built from years of production experience. Its philosophy is: make the safe thing the default, make the unsafe thing hard. Key design choices:

Principle	rmqcpp approach
Never silently drop messages	Confirms on by default, mandatory on by default, consumer acks required
Always reconnect	Built-in, transparent, topology redeclared on recovery
Heartbeats handled automatically	Users cannot opt out or forget
Separate pub/sub connections	Prevents backpressure starvation cycles
Topology as first-class concept	Declared as part of connection setup, re-applied on reconnect
MessageGuard pattern	Auto-nacks if user forgets to ack (RAII safety)
Bounded outstanding confirms	send blocks when limit is hit (API-level backpressure)

This is the design model for Pika's future API layer - not a port, but the philosophy applied to Python.

Implications for Pika

The ecosystem has three pieces sitting on the table:

1. Protocol layer: Azure's _pyamqp (MIT, pure Python, production-proven at Event Hubs/Service Bus scale). Complete AMQP 1.0 stack ready to be extracted as standalone.
2. API philosophy: Bloomberg's rmqcpp - opinionated, safe-by-default, production-hardened. The design to port to Python.
3. Gap: No standalone pure Python AMQP 1.0 client exists anywhere. The RabbitMQ team's offering is hobbled by a C dependency and has serious bugs. Azure's is locked internal.

Options (not mutually exclusive, represent a progression):

1. Pika 1.5 (now): Ship ThreadSafeConnection as a new adapter alongside existing adapters. No breaking changes. AMQP 0-9-1 only.

2. Pika 2.0: ThreadSafe becomes the default, cleanup, modern Python 3.10+. Best-in-class AMQP 0-9-1 client. Apply rmqcpp philosophy (bounded queues, built-in confirm tracking, safe defaults).

3. Pure Python AMQP 1.0 client (Pika 3.0 or new project): Fork Azure's _pyamqp as the protocol engine, wrap it with rmqcpp-style API design. Safe defaults, auto-recovery, topology management, thread-safe, zero C dependencies. Fills the ecosystem gap at the right time (RabbitMQ 4.0+ pushing everyone toward 1.0).

Open questions for the AMQP 1.0 direction

• Does this live under the Pika name (Pika 3.0) or as a new project?
• Should the 0-9-1 and 1.0 protocols share a high-level API, with the protocol as an implementation detail? Or are the semantics different enough that they deserve separate APIs?
• Is it worth discussing with the RabbitMQ team? Their Python 1.0 client needs help, and a pure Python alternative to qpid-proton would benefit them too.
• What's the relationship between Pika (community-maintained, BSD) and a hypothetical new client? Same repo? Same maintainers? Separate governance?

Discussion log

2026-06-01: Initial design session

Compared ThreadSafeChannel with the RabbitMQ Java client (ChannelN.java). Found and fixed two behavioral gaps:

1. Publish sequence number now always increments when confirms are enabled, regardless of whether on_publish is provided (commit 78fd732). Matches Java's getNextPublishSeqNo() behavior.
2. confirm_delivery is now idempotent - second call returns cached SelectOk without resending the frame or resetting the counter (commit b3ee8c1). Matches Java's confirmSelect().

Documented deliberate divergences from Java (confirm listeners, return listeners, and blocked/unblocked listeners dispatch on worker threads instead of the I/O thread - prevents heartbeat stalling).

Identified remaining gaps vs. Java not yet implemented:

• waitForConfirms / waitForConfirmsOrDie (internal unconfirmed set)
• Multiple confirm listeners (Java uses a list; pika takes one callback)
• messageCount / consumerCount convenience methods

2026-06-01: AMQP 1.0 landscape assessment

RabbitMQ is investing in AMQP 1.0 as the future protocol (native since RabbitMQ 4.0, ISO/OASIS standard, new features are 1.0-only). However AMQP 0-9-1 is not deprecated and will be supported for the foreseeable future.

The RabbitMQ team has an official AMQP 1.0 Python client (rabbitmq-amqp-python-client, v0.7.0) built on python-qpid-proton. Audited it and found significant issues in recovery, thread safety, and correctness (see AMQP10-CLIENT-AUDIT.md).

Surveyed all Python AMQP 1.0 options. Key finding: Azure deprecated their C-wrapper (uamqp) and replaced it with a pure Python AMQP 1.0 stack (_pyamqp, MIT licensed, ~25 modules) embedded inside azure-eventhub. It's production-grade but not published as standalone. No one has filled that gap.

2026-06-01: Bloomberg rmqcpp as design model

Reviewed Bloomberg's rmqcpp (Apache-2.0, C++ AMQP 0-9-1 client). Its "pit of success" philosophy - never drop messages, always reconnect, safe defaults, bounded confirms, topology-as-first-class - is exactly what Pika 2 should embody for 0-9-1, and what a future pure Python AMQP 1.0 client should embody from day one.

The convergence: fork Azure's _pyamqp for the protocol engine, apply rmqcpp's API philosophy on top, ship as a standalone pure Python AMQP 1.0 client with zero C dependencies. This fills the biggest gap in the Python messaging ecosystem at the exact moment RabbitMQ 4.0+ is pushing the industry toward AMQP 1.0.