### Example Metric Names for RabbitMQ AMQP Go Client

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Illustrates good and bad examples of metric naming conventions for clarity and consistency, following framework guidelines.

```text
- Good: `rabbitmq_amqp_connections`, `rabbitmq_amqp_published_total`
- Bad: `conn`, `msgs`, `pub`
```

--------------------------------

### Python Prometheus Client Example

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Example implementation of a metrics collector using Python's Prometheus client library. Demonstrates initialization of Gauge and Counter metrics for connections and published messages.

```python
from prometheus_client import Counter, Gauge
from typing import Protocol

class MetricsCollector(Protocol):
    def open_connection(self) -> None: ...
    def close_connection(self) -> None: ...
    # ... etc

class PrometheusMetricsCollector:
    def __init__(self, prefix: str = "rabbitmq_amqp"):
        self.connections = Gauge(
            f"{prefix}_connections",
            "Current number of open connections"
        )
        self.published = Counter(
            f"{prefix}_published_total",
            "Total published messages"
        )
        # ... initialize other metrics
    
    def open_connection(self):
        self.connections.inc()
    
    def close_connection(self):
        self.connections.dec()
    
    def publish(self):
        self.published.inc()
    
    # ... implement other methods
```

--------------------------------

### Initialize Metric Instruments (Micrometer Example)

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Initialize metric instruments in the constructor of a concrete `MetricsCollector`, associating them with a registry, prefix, and tags.

```pseudocode
constructor(registry, prefix, tags):
  connectionsGauge = registry.gauge(prefix + ".connections", tags)
  publishersGauge = registry.gauge(prefix + ".publishers", tags)
  // ... etc
```

--------------------------------

### Environment Setup with Metrics Collector

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Inject the MetricsCollector implementation during client initialization. Defaults to NoOpMetricsCollector if none is provided.

```pseudocode
environment = EnvironmentBuilder()
  .metricsCollector(metricsCollectorInstance)
  .build()
```

--------------------------------

### Implement Metric Interface Methods (Micrometer Example)

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Implement the `MetricsCollector` interface methods by interacting with the stored metric instruments, such as incrementing counters or gauges.

```pseudocode
function openConnection():
  connectionsGauge.increment()
  
function closeConnection():
  connectionsGauge.decrement()
  
function publish():
  publishedCounter.increment()
  
function publishDisposition(disposition):
  switch disposition:
    case ACCEPTED: publishedAcceptedCounter.increment()
    case REJECTED: publishedRejectedCounter.increment()
    case RELEASED: publishedReleasedCounter.increment()
  
  // ... similar for all other methods
```

--------------------------------

### Complete RabbitMQ Publish/Consume Lifecycle in Go

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Demonstrates the full workflow including environment setup, connection monitoring, topology declaration (exchange, queue, binding), publishing messages, consuming with acknowledgment, and cleanup operations like unbinding and deleting resources. Ensure RabbitMQ is running and accessible at amqp://guest:guest@localhost:5672/.

```go
package main

import (
    "context"
    "errors"
    "fmt"
    "time"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    // 1. Status monitoring
    stateChanged := make(chan *rmq.StateChanged, 1)
    go func() {
        for sc := range stateChanged {
            rmq.Info("[conn]", "state", fmt.Sprintf("%T->%T", sc.From, sc.To))
        }
    }()

    // 2. Environment & connection
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, err := env.NewConnection(context.TODO())
    if err != nil {
        panic(err)
    }
    conn.NotifyStatusChange(stateChanged)

    // 3. Topology
    mgmt := conn.Management()
    exInfo, _ := mgmt.DeclareExchange(context.TODO(), &rmq.TopicExchangeSpecification{Name: "demo.exchange"})
    qInfo, _ := mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{Name: "demo.queue"})
    bindPath, _ := mgmt.Bind(context.TODO(), &rmq.ExchangeToQueueBindingSpecification{
        SourceExchange:   "demo.exchange",
        DestinationQueue: "demo.queue",
        BindingKey:       "demo.#",
    })

    // 4. Consumer (goroutine)
    consumer, _ := conn.NewConsumer(context.TODO(), "demo.queue", nil)
    ctx, cancel := context.WithCancel(context.TODO())
    done := make(chan struct{})
    go func() {
        defer close(done)
        for {
            dc, err := consumer.Receive(ctx)
            if errors.Is(err, context.Canceled) {
                return
            }
            if err != nil {
                rmq.Error("receive error", err)
                return
            }
            rmq.Info("received", "body", string(dc.Message().GetData()))
            dc.Accept(context.TODO())
        }
    }()

    // 5. Publisher
    pub, _ := conn.NewPublisher(context.TODO(), &rmq.ExchangeAddress{
        Exchange: "demo.exchange", Key: "demo.event",
    }, nil)
    for i := 0; i < 10; i++ {
        result, err := pub.Publish(context.TODO(), rmq.NewMessage([]byte(fmt.Sprintf(`{"seq":%d}`, i))))
        if err != nil {
            rmq.Error("publish error", err)
            continue
        }
        switch result.Outcome.(type) {
        case *rmq.StateAccepted:
            rmq.Info("accepted", "seq", i)
        }
    }

    // 6. Graceful shutdown
    time.Sleep(200 * time.Millisecond) // let consumer drain
    cancel()
    <-done
    pub.Close(context.TODO())
    consumer.Close(context.TODO())
    mgmt.Unbind(context.TODO(), bindPath)
    mgmt.DeleteExchange(context.TODO(), exInfo.Name())
    purged, _ := mgmt.PurgeQueue(context.TODO(), qInfo.Name())
    fmt.Printf("purged %d remaining messages\n", purged)
    mgmt.DeleteQueue(context.TODO(), qInfo.Name())
    env.CloseConnections(context.TODO())
    close(stateChanged)
}
```

--------------------------------

### Store Metric Instruments (Micrometer Example)

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Define metric instruments like gauges and counters within a concrete `MetricsCollector` implementation, such as one using Micrometer.

```pseudocode
class MicrometerMetricsCollector implements MetricsCollector:
  // Gauge instruments (up/down counters)
  private connectionsGauge
  private publishersGauge
  private consumersGauge
  
  // Counter instruments (monotonic increase)
  private publishedCounter
  private publishedAcceptedCounter
  private publishedRejectedCounter
  private publishedReleasedCounter
  private consumedCounter
  private consumedAcceptedCounter
  private consumedRequeuedCounter
  private consumedDiscardedCounter
  private writtenBytesCounter
  private readBytesCounter
```

--------------------------------

### Declare RabbitMQ Exchanges in Go

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Use `DeclareExchange` with different specification types (Topic, Direct, FanOut) to create exchanges. Ensure proper connection setup and cleanup.

```go
package main

import (
    "context"
    "fmt"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()

    // Topic exchange (routing by wildcard key patterns)
    topicInfo, err := mgmt.DeclareExchange(context.TODO(), &rmq.TopicExchangeSpecification{
        Name:         "events",
        IsAutoDelete: false,
    })
    if err != nil {
        panic(err)
    }
    fmt.Println("declared topic exchange:", topicInfo.Name())

    // Direct exchange
    _, err = mgmt.DeclareExchange(context.TODO(), &rmq.DirectExchangeSpecification{
        Name: "orders",
    })
    if err != nil {
        panic(err)
    }

    // FanOut exchange
    _, err = mgmt.DeclareExchange(context.TODO(), &rmq.FanOutExchangeSpecification{
        Name: "broadcast",
    })
    if err != nil {
        panic(err)
    }

    // Cleanup
    mgmt.DeleteExchange(context.TODO(), "events")
    mgmt.DeleteExchange(context.TODO(), "orders")
    mgmt.DeleteExchange(context.TODO(), "broadcast")
}
```

--------------------------------

### Declare RabbitMQ Queues in Go

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Declare queues using `DeclareQueue` with various specifications like Quorum, Classic, Stream, and auto-generated. Quorum queues are recommended for durability. Ensure proper connection setup and cleanup.

```go
package main

import (
    "context"
    "fmt"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()

    // Quorum queue (recommended for durability)
    qi, err := mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{
        Name:             "orders.quorum",
        DeadLetterExchange: "dlx",
        MaxLength:        10_000,
        DeliveryLimit:    5,
    })
    if err != nil {
        panic(err)
    }
    fmt.Printf("quorum queue %q, type=%s\n", qi.Name(), qi.Type())

    // Classic queue with TTL
    _, err = mgmt.DeclareQueue(context.TODO(), &rmq.ClassicQueueSpecification{
        Name:         "temp.classic",
        IsAutoDelete: true,
        IsExclusive:  false,
        MessageTTL:   60_000, // 60 s
    })
    if err != nil {
        panic(err)
    }

    // Stream queue with retention
    _, err = mgmt.DeclareQueue(context.TODO(), &rmq.StreamQueueSpecification{
        Name:           "events.stream",
        MaxLengthBytes: rmq.CapacityGB(10),
    })
    if err != nil {
        panic(err)
    }

    // Auto-named exclusive queue (useful for RPC/fanout)
    autoQ, err := mgmt.DeclareQueue(context.TODO(), &rmq.AutoGeneratedQueueSpecification{
        IsAutoDelete: true,
        IsExclusive:  true,
    })
    if err != nil {
        panic(err)
    }
    fmt.Println("auto queue:", autoQ.Name())

    // Cleanup
    mgmt.DeleteQueue(context.TODO(), "orders.quorum")
    mgmt.DeleteQueue(context.TODO(), "temp.classic")
    mgmt.DeleteQueue(context.TODO(), "events.stream")
    mgmt.DeleteQueue(context.TODO(), autoQ.Name())
}
```

--------------------------------

### Implement MetricsCollector Interface in Go

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Implement the MetricsCollector interface to track connection, publisher, consumer, and message events. This example shows a basic implementation that counts published and consumed messages using atomic integers.

```go
package main

import (
    "context"
    "fmt"
    "sync/atomic"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

// SimpleMetrics is a minimal MetricsCollector that counts published messages.
type SimpleMetrics struct {
    publishCount atomic.Int64
    consumeCount atomic.Int64
}

func (m *SimpleMetrics) OpenConnection()      { fmt.Println("[metrics] connection opened") }
func (m *SimpleMetrics) CloseConnection()     { fmt.Println("[metrics] connection closed") }
func (m *SimpleMetrics) OpenPublisher()       {}
func (m *SimpleMetrics) ClosePublisher()      {}
func (m *SimpleMetrics) OpenConsumer()        {}
func (m *SimpleMetrics) CloseConsumer()       {}
func (m *SimpleMetrics) Publish(_ rmq.PublishContext) { m.publishCount.Add(1) }
func (m *SimpleMetrics) PublishDisposition(d rmq.PublishDisposition, ctx rmq.PublishContext) {
    fmt.Printf("[metrics] publish disposition=%d dest=%s\n", d, ctx.DestinationName)
}
func (m *SimpleMetrics) Consume(_ rmq.ConsumeContext) { m.consumeCount.Add(1) }
func (m *SimpleMetrics) ConsumeDisposition(d rmq.ConsumeDisposition, ctx rmq.ConsumeContext) {
    fmt.Printf("[metrics] consume disposition=%d queue=%s\n", d, ctx.DestinationName)
}

func main() {
    metrics := &SimpleMetrics{}

    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil,
        rmq.WithMetricsCollector(metrics),
    )
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()
    mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{Name: "metrics.queue"})

    pub, _ := conn.NewPublisher(context.TODO(), &rmq.QueueAddress{Queue: "metrics.queue"}, nil)
    pub.Publish(context.TODO(), rmq.NewMessage([]byte("hello")))
    pub.Close(context.TODO())

    fmt.Printf("total published: %d\n", metrics.publishCount.Load())
    mgmt.DeleteQueue(context.TODO(), "metrics.queue")
}
```

--------------------------------

### Minimal Custom Metrics Collector Implementation

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

A basic implementation of the MetricsCollector interface for tracking connection, publisher, and consumer metrics, along with byte counts. Use this as a starting point for custom metric collection.

```pseudocode
class SimpleMetricsCollector implements MetricsCollector:
  private connections: AtomicInteger = 0
  private publishers: AtomicInteger = 0
  private consumers: AtomicInteger = 0
  private published: AtomicLong = 0
  private consumed: AtomicLong = 0
  private writtenBytes: AtomicLong = 0
  private readBytes: AtomicLong = 0
  
  function openConnection():
    connections.incrementAndGet()
    
  function closeConnection():
    connections.decrementAndGet()
    
  function openPublisher():
    publishers.incrementAndGet()
    
  function closePublisher():
    publishers.decrementAndGet()
    
  function openConsumer():
    consumers.incrementAndGet()
    
  function closeConsumer():
    consumers.decrementAndGet()
    
  function publish():
    published.incrementAndGet()
    
  function publishDisposition(disposition):
    // Optional: track dispositions separately
    pass
    
  function consume():
    consumed.incrementAndGet()
    
  function consumeDisposition(disposition):
    // Optional: track dispositions separately
    pass
    
  function writtenBytes(count: int):
    writtenBytes.addAndGet(count)
    
  function readBytes(count: int):
    readBytes.addAndGet(count)
    
  // Accessor methods for reporting
  function getMetrics():
    return {
      "connections": connections.get(),
      "publishers": publishers.get(),
      "consumers": consumers.get(),
      "published": published.get(),
      "consumed": consumed.get(),
      "written_bytes": writtenBytes.get(),
      "read_bytes": readBytes.get()
    }
```

--------------------------------

### Stream Consumer Replaying Messages from Offset

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Consume messages from a stream queue starting at a specific offset. Supported offsets include `OffsetFirst`, `OffsetLast`, `OffsetNext`, and `OffsetValue{Offset: N}`. Ensure `InitialCredits` are set appropriately.

```go
package main

import (
    "context"
    "fmt"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()
    _, err := mgmt.DeclareQueue(context.TODO(), &rmq.StreamQueueSpecification{
        Name:           "events.log",
        MaxLengthBytes: rmq.CapacityGB(5),
    })
    if err != nil {
        panic(err)
    }

    // Publish messages
    pub, _ := conn.NewPublisher(context.TODO(), &rmq.QueueAddress{Queue: "events.log"}, nil)
    for i := 0; i < 10; i++ {
        pub.Publish(context.TODO(), rmq.NewMessage([]byte(fmt.Sprintf("event-%d", i))))
    }
    pub.Close(context.TODO())

    // Consume from the beginning of the stream
    consumer, err := conn.NewConsumer(context.TODO(), "events.log", &rmq.StreamConsumerOptions{
        Offset:         &rmq.OffsetFirst{},
        InitialCredits: 100,
    })
    if err != nil {
        panic(err)
    }

    for i := 0; i < 10; i++ {
        dc, err := consumer.Receive(context.TODO())
        if err != nil {
            panic(err)
        }
        fmt.Printf("stream msg: %s\n", dc.Message().GetData())
        dc.Accept(context.TODO())
    }

    consumer.Close(context.TODO())
    mgmt.DeleteQueue(context.TODO(), "events.log")
}
```

--------------------------------

### Create Single-Node RabbitMQ Environment

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Use `NewEnvironment` to create a top-level `Environment` for a single RabbitMQ broker. It accepts an AMQP URI and optional connection options. You can configure endpoint selection strategy and metrics backend using functional options.

```go
package main

import (
    "context"
    "fmt"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    // Basic single-node environment (anonymous SASL, default recovery)
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)

    // With explicit options: sequential endpoint selection + custom metrics
    // env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil,
    //     rmq.WithStrategy(rmq.StrategySequential),
    //     rmq.WithMetricsCollector(myOtelCollector),
    // )

    conn, err := env.NewConnection(context.TODO())
    if err != nil {
        rmq.Error("connection failed", "error", err)
        return
    }
    defer env.CloseConnections(context.TODO())

    fmt.Println("connected, id:", conn.Id())
}
```

--------------------------------

### Asynchronous High-Throughput Publishing with Publisher.PublishAsync

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Ideal for high-throughput scenarios where blocking is not desired. Use `MaxInFlight` to limit concurrency and `PublishTimeout` for per-message wait times. Confirmations are handled via a callback.

```go
package main

import (
    "context"
    "fmt"
    "sync"
    "sync/atomic"
    "time"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()
    mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{Name: "async.queue"})

    publisher, err := conn.NewPublisher(context.TODO(),
        &rmq.QueueAddress{Queue: "async.queue"},
        &rmq.PublisherOptions{
            MaxInFlight:    100,              // max 100 concurrent in-flight confirmations
            PublishTimeout: 10 * time.Second,
        },
    )
    if err != nil {
        panic(err)
    }

    const total = 10_000
    var accepted, rejected, failed atomic.Int32
    var wg sync.WaitGroup
    wg.Add(total)

    for i := 0; i < total; i++ {
        msg := rmq.NewMessage([]byte(fmt.Sprintf("msg-%d", i)))
        err = publisher.PublishAsync(context.TODO(), msg, func(result *rmq.PublishResult, cbErr error) {
            defer wg.Done()
            if cbErr != nil {
                failed.Add(1)
                return
            }
            switch result.Outcome.(type) {
            case *rmq.StateAccepted:
                accepted.Add(1)
            case *rmq.StateRejected:
                rejected.Add(1)
            }
        })
        if err != nil {
            wg.Done()
            fmt.Println("publish error:", err)
        }
    }

    wg.Wait()
    publisher.Close(context.TODO())
    fmt.Printf("accepted=%d rejected=%d failed=%d\n",
        accepted.Load(), rejected.Load(), failed.Load())
    mgmt.PurgeQueue(context.TODO(), "async.queue")
    mgmt.DeleteQueue(context.TODO(), "async.queue")
}
```

--------------------------------

### Open Connection Metrics Integration

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Call `openConnection()` on the metrics collector after a successful connection establishment.

```pseudocode
function openConnection():
    nativeConnection = establishConnection()
    validateConnection(nativeConnection)
    setState(OPEN)
    environment.metricsCollector.openConnection()  // <-- HERE
    return this
```

--------------------------------

### NewEnvironment — Create a single-endpoint environment

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Creates the top-level `Environment` that manages one or more connections to a single RabbitMQ broker. Accepts a standard AMQP URI and optional `AmqpConnOptions`. Optional functional options (`WithStrategy`, `WithMetricsCollector`) configure the endpoint selection strategy and metrics backend.

```APIDOC
## NewEnvironment

### Description
Creates the top-level `Environment` that manages one or more connections to a single RabbitMQ broker. Accepts a standard AMQP URI and optional `AmqpConnOptions`. Optional functional options (`WithStrategy`, `WithMetricsCollector`) configure the endpoint selection strategy and metrics backend.

### Method
`NewEnvironment(uri string, options *AmqpConnOptions, opts ...EnvironmentOption) *Environment`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
package main

import (
    "context"
    "fmt"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    // Basic single-node environment (anonymous SASL, default recovery)
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)

    // With explicit options: sequential endpoint selection + custom metrics
    // env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil, 
    //     rmq.WithStrategy(rmq.StrategySequential),
    //     rmq.WithMetricsCollector(myOtelCollector),
    // )

    conn, err := env.NewConnection(context.TODO())
    if err != nil {
        rmq.Error("connection failed", "error", err)
        return
    }
    defer env.CloseConnections(context.TODO())

    fmt.Println("connected, id:", conn.Id())
}
```

### Response
#### Success Response (200)
- `*Environment`: A new environment instance.

#### Response Example
None provided.
```

--------------------------------

### Pseudocode for Recording RabbitMQ Metrics

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Illustrates how to record publish, consume, and disposition metrics with OpenTelemetry attributes for RabbitMQ. Ensure correct attribute population based on message and destination details.

```pseudocode
Pseudocode:
  function recordPublishMetric(message, destination):
    attributes = {
      "messaging.system": "rabbitmq",
      "messaging.destination.name": buildDestinationName(exchange, routingKey),
      "messaging.operation.name": "send",
      "messaging.operation.type": "send",
      "server.address": connectionHost,
      "server.port": connectionPort
    }
    
    if routingKey != "":
      attributes["messaging.rabbitmq.destination.routing_key"] = routingKey
    
    if message.messageId != "":
      attributes["messaging.message.id"] = message.messageId
    
    publishedCounter.add(1, attributes)
    
  function recordConsumeMetric(message, queue):
    attributes = {
      "messaging.system": "rabbitmq",
      "messaging.destination.name": queue,
      "messaging.operation.name": "receive",
      "messaging.operation.type": "receive",
      "server.address": connectionHost,
      "server.port": connectionPort
    }
    
    if message.messageId != "":
      attributes["messaging.message.id"] = message.messageId
    
    consumedCounter.add(1, attributes)
    
  function recordDispositionMetric(disposition, destination):
    attributes = {
      "messaging.system": "rabbitmq",
      "messaging.destination.name": destination,
      "messaging.operation.type": "settle"
    }
    
    switch disposition:
      case ACCEPTED:
        attributes["messaging.operation.name"] = "ack"
      case DISCARDED:
        attributes["messaging.operation.name"] = "nack"
      case REQUEUED:
        attributes["messaging.operation.name"] = "requeue"
    
    dispositionCounter.add(1, attributes)
```

--------------------------------

### Create Multi-Node RabbitMQ Cluster Environment

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Use `NewClusterEnvironment` to create an `Environment` that spans multiple RabbitMQ broker endpoints. The library automatically selects an endpoint based on the configured strategy (`StrategyRandom` or `StrategySequential`) on each connection attempt.

```go
package main

import (
    "context"

    "github.com/Azure/go-amqp"
    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    endpoints := []rmq.Endpoint{
        {Address: "amqp://rabbit1:5672/", Options: &rmq.AmqpConnOptions{
            SASLType: amqp.SASLTypePlain("guest", "guest"),
        }},
        {Address: "amqp://rabbit2:5672/", Options: &rmq.AmqpConnOptions{
            SASLType: amqp.SASLTypePlain("guest", "guest"),
        }},
    }

    env := rmq.NewClusterEnvironment(endpoints, rmq.WithStrategy(rmq.StrategySequential))

    conn, err := env.NewConnection(context.TODO())
    if err != nil {
        panic(err)
    }
    defer env.CloseConnections(context.TODO())

    _ = conn
}
```

--------------------------------

### NewClusterEnvironment — Create a multi-node cluster environment

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Creates an `Environment` over multiple broker endpoints. On each `NewConnection` call the library picks an endpoint according to `StrategyRandom` (default) or `StrategySequential`, cycling through all nodes until one succeeds.

```APIDOC
## NewClusterEnvironment

### Description
Creates an `Environment` over multiple broker endpoints. On each `NewConnection` call the library picks an endpoint according to `StrategyRandom` (default) or `StrategySequential`, cycling through all nodes until one succeeds.

### Method
`NewClusterEnvironment(endpoints []Endpoint, opts ...EnvironmentOption) *Environment`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
package main

import (
    "context"

    "github.com/Azure/go-amqp"
    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    endpoints := []rmq.Endpoint{
        {Address: "amqp://rabbit1:5672/", Options: &rmq.AmqpConnOptions{
            SASLType: amqp.SASLTypePlain("guest", "guest"),
        }},
        {Address: "amqp://rabbit2:5672/", Options: &rmq.AmqpConnOptions{
            SASLType: amqp.SASLTypePlain("guest", "guest"),
        }},
    }

    env := rmq.NewClusterEnvironment(endpoints, rmq.WithStrategy(rmq.StrategySequential))

    conn, err := env.NewConnection(context.TODO())
    if err != nil {
        panic(err)
    }
    defer env.CloseConnections(context.TODO())

    _ = conn
}
```

### Response
#### Success Response (200)
- `*Environment`: A new cluster environment instance.

#### Response Example
None provided.
```

--------------------------------

### Track Network Bytes via Transport Options

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Configure transport options with callbacks to track bytes read and written at the transport layer during connection creation.

```pseudocode
// In Environment initialization:
readBytesCallback = (count) => metricsCollector.readBytes(count)
writtenBytesCallback = (count) => metricsCollector.writtenBytes(count)

// In Connection creation:
transportOptions = TransportOptions()
  .readBytesCallback(readBytesCallback)
  .writtenBytesCallback(writtenBytesCallback)

connection = createConnection(address, transportOptions)
```

--------------------------------

### Open Publisher Metrics Integration

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Call `openPublisher()` on the metrics collector after successful publisher creation.

```pseudocode
function createPublisher(config):
    nativeSender = session.createSender(config.address)
    nativeSender.open()
    setState(OPEN)
    metricsCollector.openPublisher()  // <-- HERE
    return publisher
```

--------------------------------

### Open Consumer Metrics Integration

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Call `openConsumer()` on the metrics collector after successful consumer creation.

```pseudocode
function createConsumer(config):
    nativeReceiver = session.createReceiver(config.queue)
    nativeReceiver.setHandler(internalHandler)
    nativeReceiver.open()
    nativeReceiver.addCredits(initialCredits)
    setState(OPEN)
    metricsCollector.openConsumer()  // <-- HERE
    return consumer
```

--------------------------------

### Synchronous Message Publishing with Publisher.Publish

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Use this for sending single messages and waiting for broker confirmation. Messages are durable by default. Can target an exchange with a routing key or a queue directly.

```go
package main

import (
    "context"
    "fmt"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()
    mgmt.DeclareExchange(context.TODO(), &rmq.TopicExchangeSpecification{Name: "orders"})
    mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{Name: "orders.created"})
    mgmt.Bind(context.TODO(), &rmq.ExchangeToQueueBindingSpecification{
        SourceExchange: "orders", DestinationQueue: "orders.created", BindingKey: "order.created",
    })

    // Publisher targeting an exchange with routing key
    publisher, err := conn.NewPublisher(context.TODO(), &rmq.ExchangeAddress{
        Exchange: "orders",
        Key:      "order.created",
    }, nil)
    if err != nil {
        panic(err)
    }
    defer publisher.Close(context.TODO())

    for i := 0; i < 5; i++ {
        msg := rmq.NewMessage([]byte(fmt.Sprintf(`{"id":%d,"status":"created"}`, i)))
        result, err := publisher.Publish(context.TODO(), msg)
        if err != nil {
            panic(err)
        }
        switch result.Outcome.(type) {
        case *rmq.StateAccepted:
            fmt.Printf("msg %d accepted\n", i)
        case *rmq.StateReleased:
            fmt.Printf("msg %d not routed\n", i)
        case *rmq.StateRejected:
            r := result.Outcome.(*rmq.StateRejected)
            fmt.Printf("msg %d rejected: %v\n", i, r.Error)
            if result.MessageRejectedError != nil {
                fmt.Printf("  rejected by queue %q: %s\n",
                    result.MessageRejectedError.RejectedBy,
                    result.MessageRejectedError.Reason)
            }
        }
    }

    mgmt.DeleteQueue(context.TODO(), "orders.created")
    mgmt.DeleteExchange(context.TODO(), "orders")
}
```

--------------------------------

### RPC Pattern: NewResponder / NewRequester

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Implements the AMQP RPC pattern for request-reply communication. NewResponder acts as a server, and NewRequester acts as a client. Ensure RabbitMQ version is 4.2 or higher for direct-reply-to support.

```go
package main

import (
    "context"
    "fmt"
    "time"

    "github.com/Azure/go-amqp"
    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()
    mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{Name: "rpc.requests"})

    // --- Server side ---
    responder, err := conn.NewResponder(context.TODO(), rmq.ResponderOptions{
        RequestQueue: "rpc.requests",
        Handler: func(ctx context.Context, req *amqp.Message) (*amqp.Message, error) {
            input := string(req.GetData())
            return rmq.NewMessage([]byte("PONG: " + input)), nil
        },
    })
    if err != nil {
        panic(err)
    }
    defer responder.Close(context.TODO())

    // --- Client side ---
    requester, err := conn.NewRequester(context.TODO(), &rmq.RequesterOptions{
        RequestQueueName: "rpc.requests",
        RequestTimeout:   5 * time.Second,
    })
    if err != nil {
        panic(err)
    }
    defer requester.Close(context.TODO())

    // Send request and await reply
    req := requester.Message([]byte("PING"))
    replyCh, err := requester.Publish(context.TODO(), req)
    if err != nil {
        panic(err)
    }

    select {
    case reply, ok := <-replyCh:
        if !ok {
            fmt.Println("request timed out")
        } else {
            fmt.Printf("reply: %s\n", reply.GetData()) // "PONG: PING"
        }
    case <-time.After(10 * time.Second):
        fmt.Println("timed out waiting for reply")
    }

    mgmt.DeleteQueue(context.TODO(), "rpc.requests")
}
```

--------------------------------

### Enable Automatic Connection Recovery

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Configure automatic reconnection with exponential back-off and jitter for unexpected disconnects. Topology is re-declared and publishers/consumers are restarted upon successful reconnection.

```go
package main

import (
    "context"
    "time"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    recoveryConfig := &rmq.RecoveryConfiguration{
        ActiveRecovery:           true,
        BackOffReconnectInterval: 5 * time.Second, // doubles each attempt, up to 1 min
        MaxReconnectAttempts:     10,
    }

    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", &rmq.AmqpConnOptions{
        RecoveryConfiguration:   recoveryConfig,
        TopologyRecoveryOptions: rmq.TopologyRecoveryAllEnabled, // redeclare all exchanges/queues/bindings
    })

    conn, err := env.NewConnection(context.TODO())
    if err != nil {
        panic(err)
    }
    defer env.CloseConnections(context.TODO())

    _ = conn
}
```

--------------------------------

### Dynamic Target Publishing with MessagePropertyToAddress

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Use this pattern for fan-out scenarios where message destinations vary. It requires setting the `To` property on each message individually using `MessagePropertyToAddress`.

```go
package main

import (
    "context"
    "fmt"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()
    mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{Name: "alpha"})
    mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{Name: "beta"})

    // Publisher with no fixed destination
    publisher, _ := conn.NewPublisher(context.TODO(), nil, nil)
    defer publisher.Close(context.TODO())

    targets := []rmq.ITargetAddress{
        &rmq.QueueAddress{Queue: "alpha"},
        &rmq.QueueAddress{Queue: "beta"},
    }

    for i, target := range targets {
        msg := rmq.NewMessage([]byte(fmt.Sprintf("payload-%d", i)))
        if err := rmq.MessagePropertyToAddress(msg, target); err != nil {
            panic(err)
        }
        result, err := publisher.Publish(context.TODO(), msg)
        if err != nil {
            panic(err)
        }
        switch result.Outcome.(type) {
        case *rmq.StateAccepted:
            fmt.Printf("sent to %v\n", target)
        }
    }

    mgmt.DeleteQueue(context.TODO(), "alpha")
    mgmt.DeleteQueue(context.TODO(), "beta")
}
```

--------------------------------

### Publisher.PublishAsync

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Sends messages asynchronously without blocking the caller. Broker confirmations are received via a callback. Options like `MaxInFlight` and `PublishTimeout` control concurrency and wait times, making it suitable for high-throughput scenarios.

```APIDOC
## Publisher.PublishAsync — Asynchronous high-throughput publishing

Sends messages without blocking the caller for broker confirmation. Confirmations arrive via a callback in a background goroutine. `MaxInFlight` limits concurrency; `PublishTimeout` controls per-message wait time. Suitable for high-throughput pipelines.

### Method
Publisher.PublishAsync(context.Context, *rmq.Message, func(*rmq.PublishResult, error))

### Parameters
#### Context
- **ctx** (context.Context) - The context for the operation.

#### Message
- **msg** (*rmq.Message) - The message to publish.

#### Callback Function
- **callback** (func(*rmq.PublishResult, error)) - A function to be called with the publish result or an error.
  - **result** (*rmq.PublishResult) - Contains the outcome of the publish operation if successful.
  - **cbErr** (error) - An error if the publish operation failed before confirmation.

### Publisher Options (when creating publisher)
- **MaxInFlight** (int) - Limits the maximum number of concurrent in-flight confirmations.
- **PublishTimeout** (time.Duration) - Controls the maximum wait time for each message confirmation.
```

--------------------------------

### Prometheus Metrics Collector (Direct)

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Implement a Prometheus metrics collector by directly using Prometheus client library primitives like Gauge and Counter. Define names and documentation strings for each metric.

```pseudocode
import prometheus

class PrometheusMetricsCollector:
  constructor(prefix: String):
    this.connectionsGauge = prometheus.Gauge(
      name: prefix + "_connections",
      documentation: "Current number of open connections"
    )
    
    this.publishedCounter = prometheus.Counter(
      name: prefix + "_published_total",
      documentation: "Total published messages"
    )
```

--------------------------------

### Tagging/Labeling Metrics with a Registry

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

When initializing a metrics collector, apply tags or labels to the registry to add dimensionality to all collected metrics. Tags should be set at initialization and remain constant to avoid high cardinality issues.

```pseudocode
constructor(registry, prefix, tags):
  // tags might be: { "application": "my-app", "environment": "prod" }
  this.tags = tags
  
  // Apply tags to all metrics
  connectionsGauge = registry.gauge(
    name: prefix + ".connections",
    tags: this.tags
  )
```

--------------------------------

### Bind and Unbind Exchange to Queue/Exchange

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Use `Bind` to create bindings between exchanges and queues or other exchanges, and `Unbind` to remove them. The `Bind` operation returns a binding path for subsequent `Unbind` calls.

```go
package main

import (
    "context"
    "fmt"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()
    mgmt.DeclareExchange(context.TODO(), &rmq.TopicExchangeSpecification{Name: "events"})
    mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{Name: "events.handler"})

    // Exchange → Queue binding
    bindPath, err := mgmt.Bind(context.TODO(), &rmq.ExchangeToQueueBindingSpecification{
        SourceExchange:   "events",
        DestinationQueue: "events.handler",
        BindingKey:       "user.#",
    })
    if err != nil {
        panic(err)
    }
    fmt.Println("binding path:", bindPath)

    // Exchange → Exchange binding
    mgmt.DeclareExchange(context.TODO(), &rmq.DirectExchangeSpecification{Name: "dlx"})
    _, err = mgmt.Bind(context.TODO(), &rmq.ExchangeToExchangeBindingSpecification{
        SourceExchange:      "events",
        DestinationExchange: "dlx",
        BindingKey:          "dead",
    })
    if err != nil {
        panic(err)
    }

    // Unbind using the path returned by Bind
    mgmt.Unbind(context.TODO(), bindPath)

    // Cleanup
    mgmt.DeleteQueue(context.TODO(), "events.handler")
    mgmt.DeleteExchange(context.TODO(), "events")
    mgmt.DeleteExchange(context.TODO(), "dlx")
}
```

--------------------------------

### Performance Benchmarking for Metrics Overhead

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Compares the performance of publishing messages with and without metrics collection enabled. This helps in assessing the overhead introduced by the metrics collector.

```pseudocode
function benchmarkPublishWithMetrics():
  collector = MicrometerMetricsCollector(registry)
  environment = Environment(metricsCollector: collector)
  
  startTime = now()
  for i in 1..100000:
    publisher.publish(message)
  duration = now() - startTime
  
function benchmarkPublishWithoutMetrics():
  environment = Environment(metricsCollector: NoOpMetricsCollector.INSTANCE)
  
  startTime = now()
  for i in 1..100000:
    publisher.publish(message)
  duration = now() - startTime
  
// Overhead should be < 5%
assert (withMetricsDuration - withoutMetricsDuration) / withoutMetricsDuration < 0.05
```

--------------------------------

### Connect via WebSocket

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Utilize `ws://` or `wss://` schemes to connect to RabbitMQ over WebSocket. The library automatically handles the AMQP WebSocket sub-protocol and Basic Auth headers.

```go
package main

import (
    "context"

    "github.com/Azure/go-amqp"
    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("ws://127.0.0.1:15678/ws", &rmq.AmqpConnOptions{
        SASLType: amqp.SASLTypePlain("guest", "guest"),
    })

    conn, err := env.NewConnection(context.TODO())
    if err != nil {
        panic(err)
    }
    defer conn.Close(context.TODO())

    _, err = conn.Management().DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{
        Name: "ws-test-queue",
    })
    if err != nil {
        panic(err)
    }

    publisher, err := conn.NewPublisher(context.TODO(), &rmq.QueueAddress{Queue: "ws-test-queue"}, nil)
    if err != nil {
        panic(err)
    }
    result, err := publisher.Publish(context.TODO(), rmq.NewMessage([]byte("hello over ws")))
    if err != nil {
        panic(err)
    }
    switch result.Outcome.(type) {
    case *rmq.StateAccepted:
        rmq.Info("message accepted")
    }
    publisher.Close(context.TODO())
    conn.Management().DeleteQueue(context.TODO(), "ws-test-queue")
}
```

--------------------------------

### OpenTelemetry Recommended Attributes for RabbitMQ

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Lists recommended attributes for enriching RabbitMQ telemetry data with OpenTelemetry, providing more context for messaging operations.

```text
| Attribute | Description | Example Values |
|-----------|-------------|----------------|
| `messaging.message.conversation_id` | Message correlation ID property | `MyConversationId` |
| `messaging.message.id` | Message identifier as a string | `452a7c7c7c7048c2f887f61572b18fc2` |
| `network.peer.address` | Peer address of the network connection | `10.1.2.80`, `/tmp/my.sock` |
| `network.peer.port` | Peer port number of the network connection | `65123` |
| `server.port` | Server port number | `80`, `8080`, `443`, `5672` |
```

--------------------------------

### Define PublishContext Structure

Source: https://github.com/rabbitmq/rabbitmq-amqp-go-client/blob/main/docs/development/METRICS_IMPLEMENTATION_GUIDE.md

Defines the context structure for passing semantic convention data to publish metric collectors. Includes server address, port, destination, routing key, and message ID.

```go
type PublishContext struct {
    ServerAddress   string // Server hostname or IP (server.address)
    ServerPort      int    // Server port (server.port)
    DestinationName string // Exchange:routing_key or queue name
    RoutingKey      string // Routing key if applicable
    MessageID       string // Message ID if available
}
```

--------------------------------

### At-least-once Consumption with Explicit Settlement

Source: https://context7.com/rabbitmq/rabbitmq-amqp-go-client/llms.txt

Use `Consumer.Receive` for receiving messages one at a time, ensuring at-least-once delivery. Each message requires explicit settlement via `Accept`, `Discard`, or `Requeue`.

```go
package main

import (
    "context"
    "errors"
    "fmt"

    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

func main() {
    env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
    conn, _ := env.NewConnection(context.TODO())
    defer env.CloseConnections(context.TODO())

    mgmt := conn.Management()
    mgmt.DeclareQueue(context.TODO(), &rmq.QuorumQueueSpecification{Name: "work.queue"})

    // Publish a few test messages
    pub, _ := conn.NewPublisher(context.TODO(), &rmq.QueueAddress{Queue: "work.queue"}, nil)
    for i := 0; i < 3; i++ {
        pub.Publish(context.TODO(), rmq.NewMessage([]byte(fmt.Sprintf("job-%d", i))))
    }
    pub.Close(context.TODO())

    // Consumer with custom initial credits
    consumer, err := conn.NewConsumer(context.TODO(), "work.queue", &rmq.ConsumerOptions{
        InitialCredits: 10,
    })
    if err != nil {
        panic(err)
    }

    ctx, cancel := context.WithCancel(context.TODO())
    defer cancel()

    for i := 0; i < 3; i++ {
        dc, err := consumer.Receive(ctx)
        if errors.Is(err, context.Canceled) {
            break
        }
        if err != nil {
            fmt.Println("receive error:", err)
            return
        }

        body := dc.Message().GetData()
        fmt.Printf("received: %s\n", body)

        // Accept (ack), Discard (reject+dlq), or Requeue (nack)
        if err = dc.Accept(context.TODO()); err != nil {
            fmt.Println("accept error:", err)
        }
    }

    consumer.Close(context.TODO())
    mgmt.DeleteQueue(context.TODO(), "work.queue")
}
```