feat: declare exchanges declaratively with WithExchangeConfig - #24
Conversation
Consumers and publishers can now declare exchanges as part of their configuration; the exchanges are declared on every channel setup, initially and after each reconnect, and for consumers before the queue and its bindings. This closes a cold-start hazard. WithBinding (and BindQueue) required the exchange to already exist: binding to a missing exchange fails with NOT_FOUND, and because that is a channel-level exception the broker closes the channel, so a consumer that started before whichever service owned the exchange could not bind at all. Declaring the exchange as part of the consumer's own topology removes the ordering requirement, on first start and on every reconnect. NewConsumer and NewPublisher reject an exchange config with an empty name (ErrInvalidConfig) rather than attempting to declare the default exchange, which the broker refuses. Consumer.DeclareExchange now defaults an unset type to direct, and the imperative declare/bind helpers document that a failed call closes the channel they share with consuming/publishing.
|
Warning Review limit reached
Next review available in: 32 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Plus Run ID: 📒 Files selected for processing (4)
WalkthroughThe library adds declarative exchange configuration for consumers and publishers. Configured exchanges are validated and declared before dependent topology during initial setup and reconnects. Tests and documentation cover declaration order, recovery, validation, and imperative channel failures. ChangesDeclarative exchange topology
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@consumer.go`:
- Around line 1141-1185: Document on the public ExchangeConfig.Type field that
an empty value defaults to the direct exchange type, and update
WithExchangeConfig documentation if needed to expose the same behavior. Also add
integration coverage in integration_test.go for an unset Type, verifying the
broker accepts the declaration as a direct exchange.
In `@publisher.go`:
- Around line 172-179: Update Publisher.handleReconnect to retry publisher setup
after setupChannel fails, including failures from declareExchange, instead of
waiting only for the next reconnectCh signal; preserve the existing reconnect
flow and ensure retries do not proceed without a successfully established
channel.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: bf9e2461-a366-4557-8c8b-1095207a92de
📒 Files selected for processing (8)
CHANGELOG.mdREADME.mdconsumer.goconsumer_test.godoc.goexamples/publisher/main.gointegration_test.gopublisher.go
Addresses PR review feedback. handleReconnect now retries setupChannel on a timer instead of waiting for the next reconnect signal. Setup can fail while the connection is healthy — a configured exchange may not be declarable yet, or may conflict with an existing one — and no further signal is coming in that case, so a single failed attempt left the publisher without a usable channel indefinitely. Retries reuse the reconnect goroutine and only ever install a channel that completed setup; a setup finishing after Close now discards its channel rather than leaking it, mirroring the consumer. ExchangeConfig.Type documents that an empty value declares a direct exchange, the AMQP default type, which declareExchange has relied on since it was introduced but only stated on the unexported helper. Integration coverage for both: an untyped ExchangeConfig is accepted by the broker and routes with direct semantics, and a publisher recovers on its own after a failing declaration stops conflicting.
Greptile SummaryThis PR adds declarative exchange declaration (
Confidence Score: 5/5Safe to merge — the changes are well-scoped, correctly handle the Close/retry race, and are backed by both unit and integration tests. The core logic — copy-before-append slice semantics, early validation of exchange names, declaring exchanges in applyTopology before queue/bindings, the p.closed guard in setupChannel, and the nil-channel select arm in handleReconnect — is all correct. The timer-based retry in handleReconnect cleanly handles connection-healthy / channel-broken scenarios without leaking goroutines or dropping close signals. Integration tests cover cold-start, reconnect recovery, type defaulting, conflict retry, and the failure baseline. Files Needing Attention: No files require special attention.
|
| Filename | Overview |
|---|---|
| publisher.go | Adds Exchanges slice to PublisherConfig with WithExchangeConfig (copy-before-append), declares configured exchanges in setupChannel(), and refactors handleReconnect to retry on a timer (via a nil-channel select arm) when setup fails. The p.closed check inside setupChannel() correctly handles the race between Close() and an in-flight timer retry. |
| consumer.go | Adds Exchanges slice to ConsumerConfig with WithExchangeConfig (copy-before-append); applyTopology now declares configured exchanges before the queue/bindings; declareExchange helper defaults empty Type to ExchangeDirect; validateExchanges rejects empty names at construction time. |
| consumer_test.go | Adds unit tests for WithExchangeConfig copy semantics, validateExchanges, and construction-time rejection of unnamed exchanges for both consumer and publisher. |
| integration_test.go | Adds six integration tests covering cold-start, reconnect recovery, unset-type defaulting, missing-exchange failure, and publisher setup-retry; helper functions waitForExchange and declareExchangeRaw are well-scoped. |
| CHANGELOG.md | New 0.11.0 entry documents WithExchangeConfig, the DeclareExchange type-defaulting fix, PublisherConfig comparability change, and the publisher setup-retry fix, all in proper Keep-a-Changelog format. |
| examples/publisher/main.go | Example updated to use declarative WithExchangeConfig instead of the imperative DeclareExchange call after construction. |
| doc.go | Package-level doc adds a 'Declarative topology' section showing WithExchangeConfig usage and explaining the cold-start hazard it solves. |
| README.md | Updates the declarative topology section with WithExchangeConfig examples for both consumer and publisher, and adds a callout warning about imperative declare/bind sharing the consuming channel. |
Sequence Diagram
sequenceDiagram
participant App
participant Publisher
participant handleReconnect
participant Connection
participant Broker
App->>Publisher: NewPublisher(conn, cfg.WithExchangeConfig(...))
Publisher->>Publisher: validateExchanges(cfg.Exchanges)
Publisher->>Publisher: setupChannel()
Publisher->>Connection: conn.Channel()
Publisher->>Broker: ExchangeDeclare("events", "topic", ...)
Broker-->>Publisher: OK (idempotent)
Publisher->>Publisher: install p.channel
Publisher->>handleReconnect: go handleReconnect()
Note over handleReconnect: Waits on reconnectCh or retry timer
Broker--xConnection: connection drop
Connection->>Connection: reconnect with backoff
Connection->>handleReconnect: reconnectCh signal
handleReconnect->>Publisher: setupChannel()
Publisher->>Broker: ExchangeDeclare("events", "topic", ...)
alt Exchange conflict (PRECONDITION_FAILED)
Broker--xPublisher: channel closed
Publisher-->>handleReconnect: error
handleReconnect->>handleReconnect: "retry = time.After(5s)"
Note over handleReconnect: waits 5s, then retries
handleReconnect->>Publisher: setupChannel() again
Publisher->>Broker: ExchangeDeclare("events", "topic", ...)
Broker-->>Publisher: OK
else Successful setup
Broker-->>Publisher: OK
end
Publisher->>Publisher: install p.channel
handleReconnect-->>handleReconnect: log channel re-established
Reviews (2): Last reviewed commit: "fix: distinguish retry from reconnect in..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
A timer-driven retry and a genuine reconnect signal both logged "re-establishing channel after reconnect", so logs could not show whether an attempt followed a connection recovery or a previous setup failure. Record which select arm woke the loop and log accordingly.
2 participants
Summary
Consumers and publishers can now declare exchanges as part of their configuration; the exchanges are declared on every channel setup, initially and after each reconnect, and for consumers before the queue and its bindings.
This closes a cold-start hazard. WithBinding (and BindQueue) required the exchange to already exist: binding to a missing exchange fails with NOT_FOUND, and because that is a channel-level exception the broker closes the channel, so a consumer that started before whichever service owned the exchange could not bind at all. Declaring the exchange as part of the consumer's own topology removes the ordering requirement, on first start and on every reconnect.
NewConsumer and NewPublisher reject an exchange config with an empty name (ErrInvalidConfig) rather than attempting to declare the default exchange, which the broker refuses. Consumer.DeclareExchange now defaults an unset type to direct, and the imperative declare/bind helpers document that a failed call closes the channel they share with consuming/publishing.
Motivation
Fixes #
Changes
Checklist
make all)Summary by CodeRabbit
New Features
Bug Fixes