promise: replace AtomicWaker with event-listener via Listener

Drop the atomic-waker dependency. Shared<T> now holds a Listener for
the notify side; Promise<T> holds a cloned Listener (sharing the same
Arc<Event>) for the arm/poll side. poll() follows the same loop pattern
as Consumer::poll_next. The manual RefUnwindSafe impl is kept since
event-listener uses UnsafeCell internally.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Keruspe

AGENTS.md

@@ -0,0 +1,94 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Commands
+
+```bash
+# Build and type-check (--all-features requires system deps for openssl; omit locally if Perl unavailable)
+cargo check --all --bins --examples --tests --all-features
+
+# Run all tests (requires a running RabbitMQ instance)
+cargo test
+
+# Run a single test file
+cargo test --test tokio           # tokio integration tests
+cargo test --test smol            # smol integration tests
+cargo test --test runtime_isolation --features=tokio
+
+# Lint
+cargo clippy --all-features -- -W clippy::all
+
+# Format check
+cargo fmt --all -- --check
+
+# Format (apply)
+cargo fmt --all
+
+# Docs
+RUSTDOCFLAGS=-D warnings cargo doc --no-deps --document-private-items --all-features
+
+# Regenerate protocol code (after changing templates/ or lapin.json)
+./regen-code.sh
+```
+
+Tests require RabbitMQ running locally (default port 5672). The CI uses a Docker `rabbitmq:latest` service.
+
+## Architecture
+
+Lapin is an async AMQP 0.9.1 client. The core abstraction is a `Connection` (one TCP socket) that multiplexes many lightweight `Channel`s. All I/O runs on a single background thread (the IO loop); user code interacts with it via lock-free channels.
+
+### Key layers
+
+**Connection & Channel (`connection.rs`, `channel.rs`)**
+- `Connection::connect()` / `Connection::connect_with_runtime()` are the entry points.
+- `connection.create_channel()` creates logical channels on the same socket.
+- `channel.rs` (~44 KB, hand-written) implements every AMQP operation (basic_publish, basic_consume, queue_declare, …). It delegates the heavy lifting to the generated layer.
+- `src/generated/channel.rs` (~73 KB, **do not edit by hand**) is produced by the codegen system and contains all option structs and low-level method implementations.
+
+**IO Loop (`io_loop.rs`)**
+- Spawned once per connection as a background thread via `runtime.rs`.
+- Owns the socket, read/write buffers, and the heartbeat timer.
+- Implements automatic reconnection with exponential backoff (`backon` crate).
+- State machine: `Initial → Connected → Stop`.
+
+**Internal RPC (`internal_rpc.rs`)**
+- `InternalRPCHandle` (cloneable `flume` sender) lets channels submit commands (Ack, Nack, Reject, CreateChannel, CloseChannel, …) to the IO loop without shared mutable state.
+- `InternalRPC` (receiver side) processes the command queue inside the IO loop.
+
+**Frame handling (`frames.rs`, `parsing.rs`)**
+- `Frames` manages the outbound frame queue and tracks expected replies.
+- `ExpectedReply` associates an outgoing frame with a promise resolver so callers can `await` the server response.
+- Frame serialization/deserialization is done by the `amq-protocol` crate.
+
+**Promise system (`promise.rs`)**
+- Thin async primitive used instead of `oneshot` channels for RPC-style request/response throughout the library.
+- `Shared<T>` holds the result behind a `Mutex<Option<Result<T>>>` (for ownership transfer on take) and a `Listener` (notify side only) for waking the waiting task.
+- `Promise<T>` holds its own `Listener` (cloned from `Shared` at construction, sharing the same `Arc<Event>`) for the arm/poll side.
+- `poll` uses the arm-then-recheck loop to close the lost-wakeup race window; the resolver drops the data lock before calling `notify()`.
+
+**Listener (`listener.rs`)**
+- `Listener` wraps `event_listener::{Event, EventListener}` and is used by `Promise`, `Consumer`'s stream impl, and `Notifier`.
+- Usage pattern: call `arm()` before checking the guarded condition, `disarm()` if it is met, otherwise `poll()` to register the waker; `notify()` wakes all clones sharing the same `Event`.
+
+**Topology & recovery (`topology.rs`, `channel_recovery_context.rs`)**
+- `Topology` tracks declared exchanges, queues, and bindings so the IO loop can replay them after a reconnect.
+- `ConnectionProperties::enable_auto_recover()` enables this behavior.
+
+### Code generation
+
+Templates live in `templates/channel.rs` (Handlebars) and `templates/lapin.json` (extra metadata). Running `./regen-code.sh` sets `LAPIN_CODEGEN_DIR=src/generated`, invokes `cargo build --features=codegen-internal`, and then formats the output. The generated file is committed; `build.rs` only regenerates it when the `codegen-internal` feature is active.
+
+When the AMQP spec or method signatures need to change, edit `templates/` and re-run `./regen-code.sh`, then commit both.
+
+### Feature flags
+
+| Category | Flags |
+|----------|-------|
+| Runtime (pick one) | `tokio` (default), `smol`, `async-global-executor` |
+| TLS | `rustls` (default), `native-tls`, `openssl`, `rustls-platform-verifier`, `rustls-native-certs`, `rustls-webpki-roots-certs` |
+| Rustls crypto | `rustls--aws_lc_rs` (default), `rustls--ring` (more portable) |
+| DNS | `hickory-dns` |
+| Codegen | `codegen` (user-facing), `codegen-internal` (build.rs only) |
+
+MSRV is **1.88.0** (Rust 2024 edition).

Cargo.lock

@@ -49,7 +49,6 @@ optional = true
 [dependencies]
 async-trait = "^0.1.42"
 cfg-if = "^1.0"
-atomic-waker = "^1.0"
 event-listener = "^5.0"
 futures-core = "^0.3"
 futures-io = "^0.3"

Cargo.toml

@@ -1,5 +1,4 @@
-use crate::{Error, Result};
-use atomic_waker::AtomicWaker;
+use crate::{Error, Result, listener::Listener};
 use std::{
     fmt,
     future::Future,
@@ -13,6 +12,7 @@ use tracing::{Level, level_enabled, trace};
 #[must_use = "Promise should be used or you can miss errors"]
 pub(crate) struct Promise<T> {
     shared: Arc<Shared<T>>,
+    listener: Listener,
 }
 
 impl<T> fmt::Debug for Promise<T> {
@@ -32,17 +32,19 @@ impl<T> Drop for Promise<T> {
 
 impl<T> Promise<T> {
     pub(crate) fn new(marker: &str) -> (Self, PromiseResolver<T>) {
-        let promise = Self {
-            shared: Shared::new(None, marker),
-        };
+        let promise = Self::build(marker, None);
         let resolver = promise.resolver();
         (promise, resolver)
     }
 
     pub(crate) fn new_with_data(marker: &str, data: Result<T>) -> Self {
-        Self {
-            shared: Shared::new(Some(data), marker),
-        }
+        Self::build(marker, Some(data))
+    }
+
+    fn build(marker: &str, data: Option<Result<T>>) -> Self {
+        let shared = Shared::new(data, marker);
+        let listener = shared.listener.clone();
+        Self { shared, listener }
     }
 
     pub(crate) fn try_wait(&self) -> Option<Result<T>> {
@@ -59,17 +61,17 @@ impl<T> Promise<T> {
 impl<T> Future for Promise<T> {
     type Output = Result<T>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        // Fast path: already resolved.
-        if let Some(data) = self.shared.take() {
-            return Poll::Ready(data);
-        }
-        // Register the waker before the second check so we don't miss a
-        // wakeup that arrives between the two take() calls.
-        self.shared.waker.register(cx.waker());
-        match self.shared.take() {
-            Some(data) => Poll::Ready(data),
-            None => Poll::Pending,
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        loop {
+            self.listener.arm();
+            if let Some(data) = self.shared.take() {
+                self.listener.disarm();
+                return Poll::Ready(data);
+            }
+            match self.listener.poll(cx) {
+                Poll::Ready(()) => {}
+                Poll::Pending => return Poll::Pending,
+            }
         }
     }
 }
@@ -122,21 +124,21 @@ impl<T> Cancelable for PromiseResolver<T> {
 
 struct Shared<T> {
     data: Mutex<Option<Result<T>>>,
-    waker: AtomicWaker,
+    listener: Listener,
     marker: Option<String>,
 }
 
-// AtomicWaker uses UnsafeCell internally, which opts out of RefUnwindSafe by
-// default. The only panic vector inside AtomicWaker is Waker::clone(); if that
-// panics, the waker's atomic state machine gets stuck. This is not a broke
-// invariant that could cause further unsoundness in code that catches the unwind.
+// Listener wraps event-listener::Event which uses UnsafeCell internally, opting
+// out of RefUnwindSafe by default. The only panic vector is Waker::wake() inside
+// notify(); if it panics the waiting task is not woken, but the data is already
+// written before notify() is called so a subsequent poll will find it.
 impl<T> RefUnwindSafe for Shared<T> where Result<T>: RefUnwindSafe {}
 
 impl<T> Shared<T> {
     fn new(data: Option<Result<T>>, marker: &str) -> Arc<Self> {
         Arc::new(Self {
             data: Mutex::new(data),
-            waker: AtomicWaker::new(),
+            listener: Listener::default(),
             marker: if level_enabled!(Level::TRACE) {
                 Some(marker.into())
             } else {
@@ -152,7 +154,7 @@ impl<T> Shared<T> {
             // Release the lock before waking to avoid the woken task
             // immediately blocking on it.
             drop(lock);
-            self.waker.wake();
+            self.listener.notify();
         }
     }