@copilot I dont like that this is separated out to AMQPReconnectingClient, implement it into the "default" client. We can do breaking changes and a new major release to ship this.

Integrated reconnection directly into AMQPClient and AMQPWebSocketClient in commit 824f876. The AMQPReconnectingClient wrapper class has been removed.

Reconnection is now configured via the constructor:

const client = new AMQPClient("amqp://localhost", undefined, undefined, {
  reconnectInterval: 1000,
  maxRetries: 0
})

Use client.subscribe() for automatic consumer recovery after reconnection.

Hey I looked at the proposed code and found the following issues/review:

• api is different now. The client provides the subscribe method.
  • Default approach is client -> channel -> queue -> subscribe or client -> channel -> basic-consume
• the subscribe method creates a channel per subscription and the user has no control of which channel is used for subscription
• reconnect logic it self seems ok (all relevant options (intervals, backoff, ...) are implemented)

In my opinion a channel should maintain the subscriptions. If the client reconnects, the client should automatically setup the previously used channels. This could be implemented as follows:

• A channel should implement a copy which returns all subscription callbacks.
• If connection is dropped -> trigger recreation of existing channels (and copy callbacks of the old elements) - use the channel id as base.
• The returned channel by the client should be like a proxy -> pointing to the actual "online" channel in the client
  • This allows e.g. storing a returned channel in variable.
  • Same should be the case for queues/consumers/(generators?) created by a channel

Open to that discussion:

• How to deal with ack-messages? messages can only be acknowledged on a specific channel.
• Subscribing to streams should although be handled differently. (perhaps the last processed offset should be stored, in the case of a reconnect use the last offset. otherwise message will be received multiple times.
  • A simple approach would to change all timebases and offset based offsets to a offset during resubscribing.

@anti-held-333 Thanks for the review! Some clarifications:

The different API is intentional - we're adding a high-level API alongside the existing low-level one, similar to https://github.com/cloudamqp/amqp-client.rb:

• Low-level (unchanged): client → channel → queue → subscribe - full control, no auto-reconnect
• High-level (new): client.subscribe() - convenience API with auto-reconnect

Both APIs coexist. The existing channel-based API is untouched.

Your Concerns

1. Channel per subscription - I dont think this is an issue. Clients normally don't have 1000s of consumers in a single process. If you do you have the low level api to work with. Or we can add pooling or something in the future.

2. No prefetch control
   Valid. Should add: client.subscribe(queue, params, callback, { prefetch: 10 })

3. Acknowledgments
   Correct - acks during reconnection fail. Unacked messages get redelivered (standard AMQP). Needs better docs and error handling.

4. Stream offsets
   Good point. Not currently handled.

Does this address your concerns?

I summarize my thoughts on that topic:

Perhaps it makes sense to use internally an topology description like:

export interface TopologyDefinition {
  channels?: {
    name: string; // Channel name/identifier (user-friendly)
    prefetch?: number; // QoS prefetch limit
  }[],
  exchanges?: {
    name: string;
    type: "direct" | "topic" | "fanout" | "headers";
  }[],
  queues?: {
    type: "queue";
    channel?: string;
    durable?: boolean;
    exclusive?: boolean;
    autoDelete?: boolean;
    bindings: {
       exchange: string;
       routingKey?: string;
       args?: Record<string, any>;
    }[]
  }[];
  streams?: {
    // See queues but with x-args
    // ...
  }[];
  subscriptions?: {
    id: string,
    queue: string,
    handler: (...args) => void | Promise<void>
  }
}

Using fallbacks it would be possible to provide the approach like described by you.

• It would require to add interfaces like: addChannel(name: string, prefetch: number) => ..., addQueue(...)
  • These would perform the required calls like creating a channel object etc. and manipulate the topology object.
• Functions like addQueue could reference to a channel created with addChannel
• That type of interface should make it possible to use the simple approach like addSubscription("hello", "hello_world_queue", (...args) => console.log(args)); without defining a channel etc! A power user should be able to call addChannel("demo", ...) and addQueue("hello_world", channel="demo") before to add some specifics (or define a custom channel to use). (I know it's not correct ts-syntax but I guess you get the point!)
• Subscriptions using a function like addSubscription could store the last offset.

@anti-held-333 Thanks for the proposal, we will consider it