### Closing Example

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-publisher.md

Example of closing the publisher connection.

```go
if err := publisher.Close(); err != nil {
    log.Printf("close error: %v", err)
}
```

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

### Configure and Create Publisher

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-publisher.md

Example showing how to initialize a publisher with a specific configuration.

```go
config := jetstream.PublisherConfig{
    URL: "nats://localhost:4222",
    Logger: watermill.NewStdLogger(false, false),
    TrackMessageID: true,
}

pub, err := jetstream.NewPublisher(config)
if err != nil {
    panic(err)
}
defer pub.Close()
```

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

### Publishing Example

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-publisher.md

Example of creating a message and publishing it to a JetStream topic.

```go
msg := message.NewMessage(watermill.NewUUID(), []byte("event data"))
msg.Metadata.Set("user-id", "42")

if err := publisher.Publish("events.user-created", msg); err != nil {
    log.Printf("publish failed: %v", err)
}
```

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

### Initialize Subscriber with New Connection

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-nats-subscriber.md

Example of configuring and creating a subscriber instance with a new NATS connection.

```go
config := nats.SubscriberConfig{
    URL:              "nats://localhost:4222",
    QueueGroupPrefix: "my-service",
    SubscribersCount: 4,
    AckWaitTimeout:   30 * time.Second,
    CloseTimeout:     30 * time.Second,
    Unmarshaler:      &nats.NATSMarshaler{},
    JetStream:        nats.JetStreamConfig{Disabled: true},
}
logger := watermill.NewStdLogger(false, false)

sub, err := nats.NewSubscriber(config, logger)
if err != nil {
    panic(err)
}
defer sub.Close()
```

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

### Implement Grouped Consumer with Custom Configuration

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-consumer.md

A complete example demonstrating stream and consumer configuration alongside a grouped consumer initializer.

```go
package main

import (
    "context"
    "log"
    "time"

    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/jetstream"
    "github.com/ThreeDotsLabs/watermill/message"
)

func main() {
    // Configure custom stream behavior
    streamConfig := func(topic string) jetstream.StreamConfig {
        return jetstream.StreamConfig{
            Name:      topic,
            Subjects:  []string{topic},
            Retention: jetstream.InterestPolicy,
            MaxAge:    7 * 24 * time.Hour,
        }
    }

    // Configure consumer naming
    consumerConfig := func(topic, group string) jetstream.ConsumerConfig {
        return jetstream.ConsumerConfig{
            Name:      "api_" + topic,
            AckPolicy: jetstream.AckExplicitPolicy,
            AckWait:   30 * time.Second,
        }
    }

    // Create subscriber with grouped consumer
    config := jetstream.SubscriberConfig{
        URL:               "nats://localhost:4222",
        Logger:            watermill.NewStdLogger(false, false),
        AckWaitTimeout:    30 * time.Second,
        ConfigureStream:   streamConfig,
        ConfigureConsumer: consumerConfig,
        ResourceInitializer: jetstream.GroupedConsumer("api-handlers"),
    }

    sub, err := jetstream.NewSubscriber(config)
    if err != nil {
        panic(err)
    }
    defer sub.Close()

    // Subscribe to topic
    messages, err := sub.Subscribe(context.Background(), "orders.created")
    if err != nil {
        panic(err)
    }

    // Process messages
    for msg := range messages {
        log.Printf("Processing order: %s", msg.UUID)
        // Process...
        msg.Ack()
    }
}
```

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

### Initialize Subscriber with Existing Connection

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-nats-subscriber.md

Example of creating a subscriber instance using an already established NATS connection.

```go
conn, _ := nats.Connect("nats://localhost:4222")
defer conn.Close()

config := nats.SubscriberSubscriptionConfig{
    Unmarshaler:    &nats.NATSMarshaler{},
    SubscribersCount: 2,
    AckWaitTimeout:  30 * time.Second,
}
sub, err := nats.NewSubscriberWithNatsConn(conn, config, logger)
if err != nil {
    panic(err)
}
```

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

### Configure StaticDelay Subscriber

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-delay-strategies.md

Example of using StaticDelay in a NATS subscriber configuration.

```go
config := nats.SubscriberConfig{
    URL:      "nats://localhost:4222",
    NakDelay: nats.NewStaticDelay(5 * time.Second),
}

sub, err := nats.NewSubscriber(config, logger)
// NACKed messages wait 5 seconds before redelivery
```

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

### Install Watermill NATS package

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/README.md

Use the go get command to add the Watermill NATS dependency to your project.

```bash
go get github.com/ThreeDotsLabs/watermill-nats/v2
```

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

### Initialize Stream

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-subscriber.md

Ensures the required stream exists for a topic before starting a subscription.

```go
if err := subscriber.SubscribeInitialize("orders.created"); err != nil {
    log.Printf("initialization failed: %v", err)
}
```

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

### Test NATS Publisher

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/README.md

Example of a standard Go test case for verifying message publication.

```go
func TestPublish(t *testing.T) {
    pub, err := nats.NewPublisher(config, logger)
    require.NoError(t, err)
    defer pub.Close()

    msg := message.NewMessage(watermill.NewUUID(), []byte("test"))
    err = pub.Publish("test-topic", msg)
    require.NoError(t, err)
}
```

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

### Get All Subjects

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/types.md

Method signature for retrieving all subjects from a SubjectDetail instance.

```go
func (s *SubjectDetail) All() []string
```

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

### Configure MaxRetryDelay Subscriber

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-delay-strategies.md

Example of using MaxRetryDelay to limit retries in a NATS subscriber.

```go
config := nats.SubscriberConfig{
    URL:      "nats://localhost:4222",
    NakDelay: nats.NewMaxRetryDelay(10*time.Second, 5),
}

sub, err := nats.NewSubscriber(config, logger)
// NACKed messages retry up to 5 times with 10s delay
// On 6th NACK, message is terminated (TermSignal)
```

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

### Implement Graceful Shutdown in Go

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/usage-patterns.md

Ensures all in-flight messages are processed and resources are closed cleanly upon receiving interrupt signals. Requires signal notification setup and explicit calls to subscriber and publisher Close methods.

```go
package main

import (
    "context"
    "os"
    "os/signal"
    "syscall"
)

func main() {
    // ... create pub/sub ...

    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)

    messages, _ := subscriber.Subscribe(context.Background(), "events")

    go func() {
        for msg := range messages {
            if err := processMessage(msg); err != nil {
                msg.Nack()
            } else {
                msg.Ack()
            }
        }
    }()

    <-sigChan
    log.Println("shutdown signal received")

    // Drain subscriber (waits for in-flight messages)
    if err := subscriber.Close(); err != nil {
        log.Printf("subscriber close error: %v", err)
    }

    // Drain publisher (waits for pending publishes)
    if err := publisher.Close(); err != nil {
        log.Printf("publisher close error: %v", err)
    }

    log.Println("shutdown complete")
}
```

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

### Initialize Publisher with NewPublisher

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-nats-publisher.md

Creates a new publisher instance by providing a configuration object and a logger. Requires a valid NATS URL and marshaler configuration.

```go
config := nats.PublisherConfig{
    URL:       "nats://localhost:4222",
    Marshaler: &nats.NATSMarshaler{},
    JetStream: nats.JetStreamConfig{Disabled: true},
}
logger := watermill.NewStdLogger(false, false)

pub, err := nats.NewPublisher(config, logger)
if err != nil {
    panic(err)
}
defer pub.Close()
```

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

### Creating Core NATS Publishers

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Initialize a publisher using a URL configuration or an existing NATS connection.

```go
// From URL
pub, err := nats.NewPublisher(config, logger)

// From existing connection
pub, err := nats.NewPublisherWithNatsConn(conn, config, logger)

// Close
pub.Close()
```

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

### Create and Metadata Management

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Initialize new messages and manage metadata key-value pairs.

```go
msg := message.NewMessage(watermill.NewUUID(), []byte("payload"))
msg.Metadata.Set("correlation-id", "123")
msg.Metadata.Set("user-id", "456")
```

```go
correlationID := msg.Metadata.Get("correlation-id")
userID := msg.Metadata.Get("user-id")
```

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

### Initialize Basic JetStream Publisher

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Configures a publisher using a NATS URL and a standard logger.

```go
config := jetstream.PublisherConfig{
    URL: "nats://localhost:4222",
    Logger: watermill.NewStdLogger(false, false),
}

pub, err := jetstream.NewPublisher(config)
if err != nil {
    panic(err)
}
defer pub.Close()
```

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

### Unmarshal JetStream message example

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-marshalers.md

Basic usage of the DefaultUnmarshaler to convert a JetStream message into a Watermill message.

```go
unmarshaler := jetstream.DefaultUnmarshaler{}
msg, err := unmarshaler.Unmarshal(jsMsg)
if err != nil {
    log.Printf("unmarshal failed: %v", err)
}
```

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

### Configuration Inheritance Pattern

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/architecture.md

Illustrates the split between creation-time and per-operation configuration subsets.

```text
PublisherConfig → PublisherPublishConfig (subset passed to Publish)
SubscriberConfig → SubscriberSubscriptionConfig (subset passed to Subscribe)
```

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

### Configure Watermill Logging

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/README.md

Demonstrates initialization of standard loggers with varying verbosity levels and implementation of a custom logger.

```go
// Standard logger
logger := watermill.NewStdLogger(false, false)     // No debug/trace
logger := watermill.NewStdLogger(true, false)      // Debug enabled
logger := watermill.NewStdLogger(true, true)       // Debug + trace

// No logging
logger := watermill.NopLogger{}

// Custom logger
type MyLogger struct{}
func (m MyLogger) Debug(msg string, fields watermill.LogFields) { }
// ... implement Info, Error, Trace, With ...
```

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

### Initialize Standard Logger

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Configure the built-in standard logger with varying levels of verbosity.

```go
// Standard logger with debug and trace disabled
logger := watermill.NewStdLogger(false, false)

// With debug enabled
logger := watermill.NewStdLogger(true, false)

// With debug and trace
logger := watermill.NewStdLogger(true, true)

// No logging
logger := watermill.NopLogger{}
```

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

### Creating JetStream Publishers

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Initialize and close a JetStream publisher.

```go
pub, err := jetstream.NewPublisher(config)
pub.Close()
```

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

### Initialize New Subscriber

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-subscriber.md

Creates a new subscriber instance using a configuration object. If no connection is provided, it establishes a new one to the specified URL.

```go
config := jetstream.SubscriberConfig{
    URL: "nats://localhost:4222",
    Logger: watermill.NewStdLogger(false, false),
    AckWaitTimeout: 30 * time.Second,
    ResourceInitializer: jetstream.EphemeralConsumer(),
}

sub, err := jetstream.NewSubscriber(config)
if err != nil {
    panic(err)
}
defer sub.Close()
```

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

### Initialize Publisher with Existing Connection

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-nats-publisher.md

Uses an existing NATS connection to instantiate the publisher. Useful for connection pooling or shared transport configurations.

```go
conn, _ := nats.Connect("nats://localhost:4222")
defer conn.Close()

config := nats.PublisherPublishConfig{
    Marshaler: &nats.NATSMarshaler{},
}
pub, err := nats.NewPublisherWithNatsConn(conn, config, logger)
if err != nil {
    panic(err)
}
```

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

### Create Mock Publisher for Testing

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Implements a simple publisher that stores messages in a slice for verification in unit tests.

```go
type MockPublisher struct {
    Messages []*message.Message
}

func (m *MockPublisher) Publish(topic string, msgs ...*message.Message) error {
    m.Messages = append(m.Messages, msgs...)
    return nil
}

func (m *MockPublisher) Close() error { return nil }
```

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

### Initialize NewPublisher

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-publisher.md

Constructor signature for creating a new JetStream publisher instance.

```go
func NewPublisher(config PublisherConfig) (*Publisher, error)
```

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

### Configure NATS Publisher for Development

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Sets up a basic publisher with JetStream disabled for local development.

```go
config := nats.PublisherConfig{
    URL:       "nats://localhost:4222",
    Marshaler: &nats.NATSMarshaler{},
    JetStream: nats.JetStreamConfig{Disabled: true},
}
```

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

### Configure Load-Balanced Subscriber

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Sets up a subscriber with multiple instances and queue grouping.

```go
config := nats.SubscriberConfig{
    URL:              "nats://localhost:4222",
    QueueGroupPrefix: "service-name",
    SubscribersCount: 4,
    JetStream: nats.JetStreamConfig{
        Disabled:      false,
        DurablePrefix: "service-name",
    },
}
```

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

### Custom Stream Configurator Implementation

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-consumer.md

Demonstrates how to define a custom stream configuration function and apply it to a SubscriberConfig.

```go
streamConfig := func(topic string) jetstream.StreamConfig {
    return jetstream.StreamConfig{
        Name:      topic,
        Subjects:  []string{topic},
        Retention: jetstream.InterestPolicy,
        MaxAge:    24 * time.Hour,
        Storage:   jetstream.FileStorage,
    }
}

config := jetstream.SubscriberConfig{
    URL:             "nats://localhost:4222",
    ConfigureStream: streamConfig,
}

sub, err := jetstream.NewSubscriber(config)
```

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

### Configure Existing Consumer

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Initializes a subscriber that connects to a pre-existing consumer defined by a custom namer function.

```go
namer := func(topic, group string) jetstream.ConsumerConfig {
    return jetstream.ConsumerConfig{
        Name:      fmt.Sprintf("my_%s", topic),
        AckPolicy: jetstream.AckExplicitPolicy,
    }
}

config := jetstream.SubscriberConfig{
    URL:    "nats://localhost:4222",
    Logger: watermill.NewStdLogger(false, false),
    ResourceInitializer: jetstream.ExistingConsumer(namer, ""),
}

sub, err := jetstream.NewSubscriber(config)
```

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

### Implement StaticDelay

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/types.md

Fixed delay implementation for all redelivery attempts.

```go
type StaticDelay struct {
    Delay time.Duration
}
```

```go
func NewStaticDelay(delay time.Duration) StaticDelay
```

```go
delay := nats.NewStaticDelay(5 * time.Second)
config.NakDelay = delay
```

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

### Connect to an Existing Consumer

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-consumer.md

Use when connecting to pre-provisioned infrastructure where the consumer lifecycle is managed externally.

```go
// Custom namer to find pre-configured consumers
namer := func(topic, group string) jetstream.ConsumerConfig {
    return jetstream.ConsumerConfig{
        Name: fmt.Sprintf("service_%s_%s", topic, group),
    }
}

config := jetstream.SubscriberConfig{
    URL: "nats://localhost:4222",
    ResourceInitializer: jetstream.ExistingConsumer(namer, "mygroup"),
}

sub, err := jetstream.NewSubscriber(config)
// Fails if stream/consumer don't exist on broker
```

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

### Creating Core NATS Subscribers

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Initialize a subscriber using a URL configuration or an existing NATS connection.

```go
// From URL
sub, err := nats.NewSubscriber(config, logger)

// From existing connection
sub, err := nats.NewSubscriberWithNatsConn(conn, config, logger)

// Close
sub.Close()
```

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

### Initialize Publisher with Custom Stream Configuration

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Provides a custom stream configurator function and enables message ID tracking for deduplication.

```go
streamConfig := func(topic string) jetstream.StreamConfig {
    return jetstream.StreamConfig{
        Name:      topic,
        Subjects:  []string{topic},
        Retention: jetstream.InterestPolicy,
        MaxAge:    24 * time.Hour,
    }
}

config := jetstream.PublisherConfig{
    URL:             "nats://localhost:4222",
    ConfigureStream: streamConfig,
    TrackMessageID:  true,
}

pub, err := jetstream.NewPublisher(config)
```

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

### Enable Debug Logging for NATS

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/errors.md

Initialize a standard logger with debug and trace enabled to monitor NATS activity.

```go
logger := watermill.NewStdLogger(true, true) // debug=true, trace=true

pub, _ := nats.NewPublisher(config, logger)
sub, _ := nats.NewSubscriber(config, logger)

// Now all NATS activity is logged
```

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

### NewSubscriberWithNatsConn Constructor Signature

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-nats-subscriber.md

Function signature for initializing a subscriber using an existing NATS connection.

```go
func NewSubscriberWithNatsConn(conn *nats.Conn, config SubscriberSubscriptionConfig, logger watermill.LoggerAdapter) (*Subscriber, error)
```

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

### Implement Graceful Shutdown

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/README.md

Handles OS signals to trigger a clean shutdown of subscribers and publishers.

```go
// Signal handling
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
<-sigChan

// Graceful shutdown
ctx, cancel := context.WithCancel(context.Background())
cancel()                    // Stop accepting new work

if err := subscriber.Close(); err != nil {
    log.Printf("subscriber error: %v", err)
}
if err := publisher.Close(); err != nil {
    log.Printf("publisher error: %v", err)
}
```

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

### Configure Ephemeral Consumer

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Initializes a subscriber with an ephemeral consumer that is automatically cleaned up upon closing.

```go
config := jetstream.SubscriberConfig{
    URL:    "nats://localhost:4222",
    Logger: watermill.NewStdLogger(false, false),
    AckWaitTimeout: 30 * time.Second,
}

sub, err := jetstream.NewSubscriber(config)
// Automatically creates and cleans up ephemeral consumer
```

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

### Initialize NATS Subscriber

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/usage-patterns.md

Sets up a basic NATS subscriber to consume messages from a specific topic without durability.

```go
package main

import (
    "context"
    "log"
    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats"
)

func main() {
    logger := watermill.NewStdLogger(false, false)

    sub, err := nats.NewSubscriber(
        nats.SubscriberConfig{
            URL:         "nats://localhost:4222",
            Unmarshaler: &nats.NATSMarshaler{},
            JetStream:   nats.JetStreamConfig{Disabled: true},
        },
        logger,
    )
    if err != nil {
        panic(err)
    }
    defer sub.Close()

    messages, err := sub.Subscribe(context.Background(), "events")
    if err != nil {
        panic(err)
    }

    for msg := range messages {
        log.Printf("received: %s", string(msg.Payload))
        msg.Ack()
    }
}
```

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

### Create Mock Subscriber for Testing

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Implements a subscriber that returns a pre-defined channel of messages for testing message consumption flows.

```go
type MockSubscriber struct {
    Messages chan *message.Message
}

func (m *MockSubscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error) {
    return m.Messages, nil
}

func (m *MockSubscriber) Close() error { return nil }
```

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

### Configure Grouped Consumer

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Initializes a subscriber using a grouped consumer, allowing multiple instances to share state.

```go
config := jetstream.SubscriberConfig{
    URL:    "nats://localhost:4222",
    Logger: watermill.NewStdLogger(false, false),
    ResourceInitializer: jetstream.GroupedConsumer("my-consumer-group"),
    AckWaitTimeout: 30 * time.Second,
}

sub, err := jetstream.NewSubscriber(config)
// Multiple instances with same group share consumer state
```

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

### Configure Immediate Retry Delay

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Use a static delay of zero to trigger immediate retries upon a NAK.

```go
NakDelay: nats.NewStaticDelay(0)
```

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

### Configure JetStream Publisher with Auto-Provisioning

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Enables JetStream features including automatic stream provisioning and message ID tracking for deduplication.

```go
config := nats.PublisherConfig{
    URL: "nats://localhost:4222",
    Marshaler: &nats.NATSMarshaler{},
    JetStream: nats.JetStreamConfig{
        Disabled:      false,
        AutoProvision: true,
        TrackMsgID:    true,
    },
}

pub, err := nats.NewPublisher(config, logger)
```

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

### Initialize JetStream Consumers

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Defines strategies for consumer lifecycle management in JetStream.

```go
ResourceInitializer: jetstream.EphemeralConsumer()
```

```go
ResourceInitializer: jetstream.GroupedConsumer("group-name")
```

```go
ResourceInitializer: jetstream.ExistingConsumer(namer, "group")
```

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

### Subscribe to Topic

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-subscriber.md

Establishes a subscription to a specific topic and processes incoming messages. Messages must be acknowledged or negatively acknowledged within the configured timeout.

```go
messages, err := subscriber.Subscribe(context.Background(), "orders.created")
if err != nil {
    panic(err)
}

for msg := range messages {
    log.Printf("received: %s", msg.UUID)
    
    // Process message...
    
    msg.Ack()
}
```

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

### Define PublisherConfig

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/types.md

Configuration structure for initializing a core NATS publisher.

```go
type PublisherConfig struct {
    URL                 string
    NatsOptions         []nats.Option
    Marshaler           Marshaler
    SubjectCalculator   SubjectCalculator
    JetStream           JetStreamConfig
}
```

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

### Configure JSON Marshaling for NATS Publisher

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Sets up a NATS publisher using the JSONMarshaler for cross-language interoperability.

```go
config := nats.PublisherConfig{
    URL:       "nats://localhost:4222",
    Marshaler: &nats.JSONMarshaler{},
}

pub, err := nats.NewPublisher(config, logger)
```

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

### Constructor for StaticDelay

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-delay-strategies.md

Creates a new StaticDelay instance.

```go
func NewStaticDelay(delay time.Duration) StaticDelay
```

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

### Configure Core NATS Publisher

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Initializes a publisher with JetStream disabled using the Gob marshaler.

```go
config := nats.PublisherConfig{
    URL:         "nats://localhost:4222",
    Marshaler:   &nats.GobMarshaler{},
    JetStream:   nats.JetStreamConfig{Disabled: true},
}

pub, err := nats.NewPublisher(config, logger)
if err != nil {
    panic(err)
}
```

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

### NewSubscriber Constructor Signature

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-nats-subscriber.md

Function signature for initializing a subscriber with a new NATS connection.

```go
func NewSubscriber(config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)
```

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

### ExistingConsumer

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-consumer.md

Connects to a pre-existing consumer without attempting to create or modify it, suitable for pre-provisioned infrastructure.

```APIDOC
## func ExistingConsumer(consumerNamer ConsumerConfigurator, group string)

### Description
Connects to a pre-existing consumer without creating or modifying it. This requires both the stream and consumer to already exist on the broker.

### Parameters
- **consumerNamer** (ConsumerConfigurator) - Required - Function to determine consumer configuration from topic
- **group** (string) - Required - Consumer group name (can be empty)

### Returns
- **ResourceInitializer** - Function that connects to existing consumers

### Example
```go
namer := func(topic, group string) jetstream.ConsumerConfig {
    return jetstream.ConsumerConfig{
        Name: fmt.Sprintf("service_%s_%s", topic, group),
    }
}

config := jetstream.SubscriberConfig{
    URL: "nats://localhost:4222",
    ResourceInitializer: jetstream.ExistingConsumer(namer, "mygroup"),
}

sub, err := jetstream.NewSubscriber(config)
```
```

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

### Publish messages to NATS

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/README.md

Initialize a NATS publisher using PublisherConfig and publish a message to a specific topic.

```go
package main

import (
    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats"
    "github.com/ThreeDotsLabs/watermill/message"
)

func main() {
    pub, _ := nats.NewPublisher(
        nats.PublisherConfig{
            URL: "nats://localhost:4222",
        },
        watermill.NewStdLogger(false, false),
    )
    defer pub.Close()

    msg := message.NewMessage(watermill.NewUUID(), []byte("hello"))
    pub.Publish("events", msg)
}
```

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

### Configure Core NATS with Multiple Subscribers

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Sets up a subscriber with multiple concurrent handlers and disables JetStream functionality.

```go
config := nats.SubscriberConfig{
    URL:              "nats://localhost:4222",
    SubscribersCount: 4,
    CloseTimeout:     30 * time.Second,
    AckWaitTimeout:   30 * time.Second,
    Unmarshaler:      &nats.GobMarshaler{},
    JetStream:        nats.JetStreamConfig{Disabled: true},
}

sub, err := nats.NewSubscriber(config, logger)
```

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

### Create a Grouped Consumer

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-consumer.md

Use for load-balancing messages across multiple service instances with persistent state.

```go
config := jetstream.SubscriberConfig{
    URL: "nats://localhost:4222",
    ResourceInitializer: jetstream.GroupedConsumer("order-processors"),
}

sub, err := jetstream.NewSubscriber(config)
// Multiple instances can join "order-processors" group
// Messages distributed across all instances
```

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

### NewPublisherWithNatsConn

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-nats-publisher.md

Creates a new Publisher instance using an existing NATS connection.

```APIDOC
## func NewPublisherWithNatsConn(conn *nats.Conn, config PublisherPublishConfig, logger watermill.LoggerAdapter) (*Publisher, error)

### Description
Creates a new Publisher using an existing NATS connection. Useful for sharing connections or using pre-configured connection options.

### Parameters
- **conn** (*nats.Conn) - Required - Existing NATS connection
- **config** (PublisherPublishConfig) - Required - Publisher configuration subset
- **logger** (watermill.LoggerAdapter) - Optional - Logger for debug/info/error messages

### Returns
- **Publisher** (*Publisher) - Initialized publisher instance
- **error** (error) - Configuration validation error
```

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

### Creating JetStream Subscribers

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Initialize and close a JetStream subscriber.

```go
sub, err := jetstream.NewSubscriber(config)
sub.Close()
```

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

### Implement Custom Subject Calculator

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/usage-patterns.md

Define non-standard topic-to-subject mappings by providing a custom calculator function to the publisher configuration.

```go
package main

import "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats"

// Map topics to hierarchical subjects
func customSubjectCalculator(queueGroupPrefix, topic string) *nats.SubjectDetail {
    return &nats.SubjectDetail{
        Primary:    "app." + topic,  // "orders.created" → "app.orders.created"
        QueueGroup: queueGroupPrefix,
    }
}

func main() {
    config := nats.PublisherConfig{
        URL:               "nats://localhost:4222",
        Marshaler:         &nats.NATSMarshaler{},
        SubjectCalculator: customSubjectCalculator,
    }
    pub, _ := nats.NewPublisher(config, logger)
}
```

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

### Define SubscriberConfig

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/types.md

Configuration structure for initializing a core NATS subscriber.

```go
type SubscriberConfig struct {
    URL               string
    QueueGroupPrefix  string
    SubscribersCount  int
    CloseTimeout      time.Duration
    AckWaitTimeout    time.Duration
    SubscribeTimeout  time.Duration
    NatsOptions       []nats.Option
    Unmarshaler       Unmarshaler
    SubjectCalculator SubjectCalculator
    NakDelay          Delay
    JetStream         JetStreamConfig
}
```

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

### PublisherConfig

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Configuration structure for initializing a NATS publisher in Watermill.

```APIDOC
## PublisherConfig

### Description
The `PublisherConfig` struct is used to configure the NATS publisher. It supports both standard NATS and JetStream configurations.

### Configuration Options
- **URL** (string) - Required - NATS server URL (e.g., nats://localhost:4222)
- **NatsOptions** ([]nats.Option) - Optional - Custom nats.go options (for auth, TLS, timeouts, etc.)
- **Marshaler** (Marshaler) - Optional - Message format (GobMarshaler, JSONMarshaler, NATSMarshaler)
- **SubjectCalculator** (SubjectCalculator) - Optional - Function to map topic to NATS subject
- **JetStream.Disabled** (bool) - Optional - Set true for core NATS without JetStream
- **JetStream.AutoProvision** (bool) - Optional - Auto-create stream if missing
- **JetStream.TrackMsgID** (bool) - Optional - Enable deduplication via message UUID
- **JetStream.AckAsync** (bool) - Optional - Use async instead of sync ack
- **JetStream.DurablePrefix** (string) - Optional - Prefix for durable consumer names
- **JetStream.DurableCalculator** (DurableCalculator) - Optional - Custom durable name derivation
- **JetStream.ConnectOptions** ([]nats.JSOpt) - Optional - JetStream context options
- **JetStream.SubscribeOptions** ([]nats.SubOpt) - Optional - JetStream subscription options
- **JetStream.PublishOptions** ([]nats.PubOpt) - Optional - JetStream publish options
```

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

### Integrate Structured Logging with Slog

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/usage-patterns.md

Create a custom adapter to route Watermill logs through the standard library slog package.

```go
package main

import (
    "context"
    "log/slog"
    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats"
)

type SlogAdapter struct {
    logger *slog.Logger
}

func (a *SlogAdapter) Debug(msg string, fields watermill.LogFields) {
    a.logger.Debug(msg, slog.Any("fields", fields))
}

func (a *SlogAdapter) Info(msg string, fields watermill.LogFields) {
    a.logger.Info(msg, slog.Any("fields", fields))
}

// ... implement Error, Trace, With

func main() {
    logger := &SlogAdapter{logger: slog.Default()}

    sub, _ := nats.NewSubscriber(config, logger)
    // All logs go through slog
}
```

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

### SubscriberConfig

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Configuration parameters for initializing a NATS subscriber.

```APIDOC
## SubscriberConfig

### Description
Configuration struct for initializing a NATS subscriber in Watermill.

### Parameters
- **URL** (string) - Required - NATS server URL
- **QueueGroupPrefix** (string) - Optional - Queue group for load balancing
- **SubscribersCount** (int) - Optional - Number of concurrent message handlers
- **CloseTimeout** (time.Duration) - Optional - Max wait for in-flight message acks on close
- **AckWaitTimeout** (time.Duration) - Optional - Max wait for message ack before redelivery (JetStream only)
- **SubscribeTimeout** (time.Duration) - Optional - Max wait for successful subscription
- **NatsOptions** ([]nats.Option) - Optional - Custom nats.go options
- **Unmarshaler** (Unmarshaler) - Optional - Message format handler
- **SubjectCalculator** (SubjectCalculator) - Optional - Topic to subject mapping function
- **NakDelay** (Delay) - Optional - Delay strategy for NACKed messages (JetStream only)
- **JetStream.Disabled** (bool) - Optional - Set true for core NATS without JetStream
- **JetStream.AutoProvision** (bool) - Optional - Auto-create stream if missing
- **JetStream.DurablePrefix** (string) - Optional - Prefix for durable consumer names
- **JetStream.AckAsync** (bool) - Optional - Use async instead of sync ack
- **JetStream.ConnectOptions** ([]nats.JSOpt) - Optional - JetStream context options
- **JetStream.SubscribeOptions** ([]nats.SubOpt) - Optional - JetStream subscription options
```

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

### Implement Custom Logger

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Create a custom logger by implementing the watermill.LoggerAdapter interface.

```go
type CustomLogger struct{}

func (c CustomLogger) Debug(msg string, fields watermill.LogFields) { }
func (c CustomLogger) Info(msg string, fields watermill.LogFields) { }
func (c CustomLogger) Error(msg string, err error, fields watermill.LogFields) { }
func (c CustomLogger) Trace(msg string, fields watermill.LogFields) { }
func (c CustomLogger) With(fields watermill.LogFields) watermill.LoggerAdapter { }
```

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

### Implement MaxRetryDelay

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/types.md

Fixed delay with a maximum retry count before termination.

```go
type MaxRetryDelay struct {
    StaticDelay
    maxRetries uint64
}
```

```go
func NewMaxRetryDelay(delay time.Duration, retryLimit uint64) MaxRetryDelay
```

```go
delay := nats.NewMaxRetryDelay(10 * time.Second, 5)
config.NakDelay = delay
```

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

### Implement JetStream Publisher and Subscriber

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/usage-patterns.md

Utilize the JetStream beta API for modern NATS messaging, including grouped consumer initialization.

```go
package main

import (
    "context"
    "log"
    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/jetstream"
    "github.com/nats-io/nats.go"
)

func main() {
    logger := watermill.NewStdLogger(false, false)

    // Publisher
    pubConfig := jetstream.PublisherConfig{
        URL:            "nats://localhost:4222",
        Logger:         logger,
        TrackMessageID: true,
    }
    pub, _ := jetstream.NewPublisher(pubConfig)
    defer pub.Close()

    // Subscriber with grouped consumer
    subConfig := jetstream.SubscriberConfig{
        URL:    "nats://localhost:4222",
        Logger: logger,
        ResourceInitializer: jetstream.GroupedConsumer("processors"),
    }
    sub, _ := jetstream.NewSubscriber(subConfig)
    defer sub.Close()

    messages, _ := sub.Subscribe(context.Background(), "events")
    for msg := range messages {
        log.Printf("processing event: %s", msg.UUID)
        msg.Ack()
    }
}
```

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

### Define PublisherConfig struct

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/types.md

Configuration structure for initializing a JetStream publisher.

```go
type PublisherConfig struct {
    URL             string
    Conn            *nats.Conn
    Logger          watermill.LoggerAdapter
    ConfigureStream StreamConfigurator
    TrackMessageID  bool
}
```

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

### Create an Ephemeral Consumer

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-consumer.md

Use for temporary subscriptions where the consumer is automatically cleaned up upon closing.

```go
config := jetstream.SubscriberConfig{
    URL: "nats://localhost:4222",
    ResourceInitializer: jetstream.EphemeralConsumer(),
}

sub, err := jetstream.NewSubscriber(config)
// Consumer automatically created on Subscribe() and deleted on Close()
```

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

### Configure Core NATS Subscriber

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-delay-strategies.md

Demonstrates that Core NATS does not support message redelivery or NakDelay configurations.

```go
// Core NATS doesn't support message redelivery by default
config := nats.SubscriberConfig{
    URL:       "nats://localhost:4222",
    JetStream: nats.JetStreamConfig{Disabled: true},
    // NakDelay is not applicable
}

sub, err := nats.NewSubscriber(config, logger)
// Messages are delivered once; NACK has no effect
```

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

### SubscribeInitialize

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-subscriber.md

Ensures the stream for a topic exists prior to subscription.

```APIDOC
## func (s *Subscriber) SubscribeInitialize(topic string) error

### Description
Ensures the stream for a topic exists prior to subscription. It creates the JetStream stream if it does not exist using the configured stream settings.

### Parameters
- **topic** (string) - Required - Topic to initialize.

### Returns
- **error** (error) - Stream creation error.
```

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

### Implement GobMarshaler

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-marshalers.md

Uses Go's gob encoding for serialization.

```go
type GobMarshaler struct{}
```

```go
func (GobMarshaler) Marshal(topic string, msg *message.Message) (*nats.Msg, error)
```

```go
marshaler := &nats.GobMarshaler{}
msg := message.NewMessage(watermill.NewUUID(), []byte("data"))
natsMsg, err := marshaler.Marshal("orders", msg)
if err != nil {
    log.Printf("marshal failed: %v", err)
}
```

```go
func (GobMarshaler) Unmarshal(natsMsg *nats.Msg) (*message.Message, error)
```

```go
marshaler := &nats.GobMarshaler{}
msg, err := marshaler.Unmarshal(natsMsg)
if err != nil {
    log.Printf("unmarshal failed: %v", err)
}
```

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

### Initialize a NATS stream

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-nats-subscriber.md

Ensures the stream for a topic exists before subscription. This is only functional when JetStream is enabled and AutoProvision is true.

```go
if err := subscriber.SubscribeInitialize("orders.created"); err != nil {
    log.Printf("failed to initialize: %v", err)
}
```

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

### Configure NATS Marshalers

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Select the appropriate marshaler based on performance and interoperability requirements. NATSMarshaler is the default implementation.

```go
// NATSMarshaler (default)
Marshaler: &nats.NATSMarshaler{}

// GobMarshaler
Marshaler: &nats.GobMarshaler{}

// JSONMarshaler
Marshaler: &nats.JSONMarshaler{}
```

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

### Configure NATS Authentication

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Use these options to authenticate with NATS servers using either user credentials or tokens.

```go
NatsOptions: []nats.Option{
    nats.UserInfo("user", "password"),
}
```

```go
NatsOptions: []nats.Option{
    nats.Token("token-value"),
}
```

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

### Importing Watermill NATS Packages

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Required imports for core NATS, JetStream, Watermill message types, and the NATS client.

```go
// Core NATS
import "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats"

// JetStream (beta)
import "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/jetstream"

// Watermill
import (
    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill/message"
)

// NATS client
import "github.com/nats-io/nats.go"
```

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

### Configure JetStream Subscriber

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/quick-reference.md

Subscriber configurations for development and production environments.

```go
config := jetstream.SubscriberConfig{
    URL:    "nats://localhost:4222",
    Logger: watermill.NewStdLogger(true, true),
}
```

```go
config := jetstream.SubscriberConfig{
    URL:    "nats://nats-server:4222",
    Logger: watermill.NewStdLogger(false, false),
    AckWaitTimeout: 30 * time.Second,
    ResourceInitializer: jetstream.GroupedConsumer("service-name"),
}
```

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

### Implement NATSMarshaler

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-marshalers.md

Uses NATS headers for metadata and raw payload, providing the most efficient serialization.

```go
type NATSMarshaler struct{}
```

```go
func (*NATSMarshaler) Marshal(topic string, msg *message.Message) (*nats.Msg, error)
```

```go
marshaler := &nats.NATSMarshaler{}
msg := message.NewMessage(watermill.NewUUID(), []byte("data"))
msg.Metadata.Set("user-id", "123")
natsMsg, err := marshaler.Marshal("orders", msg)
// natsMsg has headers: _watermill_message_uuid and user-id
```

```go
func (*NATSMarshaler) Unmarshal(natsMsg *nats.Msg) (*message.Message, error)
```

```go
marshaler := &nats.NATSMarshaler{}
msg, err := marshaler.Unmarshal(natsMsg)
```

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

### Subscribe to NATS messages

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/README.md

Initialize a NATS subscriber and process incoming messages from a topic, ensuring to acknowledge each message.

```go
package main

import (
    "context"
    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats"
)

func main() {
    sub, _ := nats.NewSubscriber(
        nats.SubscriberConfig{
            URL: "nats://localhost:4222",
        },
        watermill.NewStdLogger(false, false),
    )
    defer sub.Close()

    messages, _ := sub.Subscribe(context.Background(), "events")
    for msg := range messages {
        println(string(msg.Payload))
        msg.Ack()
    }
}
```

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

### Publish Messages

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-publisher.md

Method signature for publishing messages to a specific topic.

```go
func (p *Publisher) Publish(topic string, messages ...*message.Message) error
```

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

### Constructor for MaxRetryDelay

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-delay-strategies.md

Creates a new MaxRetryDelay instance with a retry limit.

```go
func NewMaxRetryDelay(delay time.Duration, retryLimit uint64) MaxRetryDelay
```

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

### Configure Custom NATS Connection Options

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Applies custom NATS client options such as authentication, timeouts, and retry logic via the NatsOptions field.

```go
options := []nats.Option{
    nats.RetryOnFailedConnect(true),
    nats.Timeout(30 * time.Second),
    nats.ReconnectWait(1 * time.Second),
    nats.UserInfo("user", "password"),
}

config := nats.PublisherConfig{
    URL:         "nats://nats-server:4222",
    NatsOptions: options,
    Marshaler:   &nats.NATSMarshaler{},
}

pub, err := nats.NewPublisher(config, logger)
```

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

### Configure JetStream with Durable Consumer

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/configuration.md

Enables JetStream with auto-provisioning and custom subscription options for durable message consumption.

```go
subscribeOpts := []nc.SubOpt{
    nc.DeliverAll(),
    nc.AckExplicit(),
}

config := nats.SubscriberConfig{
    URL:              "nats://localhost:4222",
    QueueGroupPrefix: "my-service",
    SubscribersCount: 2,
    AckWaitTimeout:   30 * time.Second,
    Unmarshaler:      &nats.NATSMarshaler{},
    JetStream: nats.JetStreamConfig{
        Disabled:         false,
        AutoProvision:    true,
        DurablePrefix:    "my-service",
        SubscribeOptions: subscribeOpts,
    },
}

sub, err := nats.NewSubscriber(config, logger)
```

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

### Subscribe

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-jetstream-subscriber.md

Establishes a subscription to a specific topic and returns a channel for message consumption.

```APIDOC
## func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

### Description
Establishes a JetStream subscription to a topic and returns a channel for receiving messages. Messages must be acknowledged or nacked within the configured AckWaitTimeout.

### Parameters
- **ctx** (context.Context) - Required - Context for cancellation and timeout.
- **topic** (string) - Required - Topic to subscribe to.

### Returns
- **channel** (<-chan *message.Message) - Read-only channel of messages.
- **error** (error) - Consumer initialization error.
```

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

### Configure Immediate Redelivery in JetStream

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-delay-strategies.md

Sets the NakDelay to zero for immediate message redelivery upon a NACK.

```go
config := jetstream.SubscriberConfig{
    URL:      "nats://localhost:4222",
    NakDelay: jetstream.NewStaticDelay(0),
}

sub, err := jetstream.NewSubscriber(config)
// Messages redelivered immediately without delay
```

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

### NewPublisher

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/api-nats-publisher.md

Creates a new Publisher instance with a new NATS connection based on the provided configuration.

```APIDOC
## func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) (*Publisher, error)

### Description
Creates a new Publisher with a new NATS connection.

### Parameters
- **config** (PublisherConfig) - Required - Configuration for publisher initialization and behavior
- **logger** (watermill.LoggerAdapter) - Optional - Logger for debug/info/error messages

### Returns
- **Publisher** (*Publisher) - Initialized publisher instance
- **error** (error) - Connection or configuration error
```

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

### Define SubscriberConfig struct

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/types.md

Configuration structure for initializing a JetStream subscriber.

```go
type SubscriberConfig struct {
    URL                 string
    Conn                *nats.Conn
    Logger              watermill.LoggerAdapter
    AckWaitTimeout      time.Duration
    ResourceInitializer ResourceInitializer
    NakDelay            Delay
    ConfigureStream     StreamConfigurator
    ConfigureConsumer   ConsumerConfigurator
    ConsumeOptions      []natsJS.PullConsumeOpt
    AckAsync            bool
    Unmarshaler         Unmarshaler
}
```

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

### Consumer Initialization Errors

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/errors.md

Errors related to the initialization of Ephemeral, Grouped, and Existing consumers.

```APIDOC
## Consumer Initialization (EphemeralConsumer, GroupedConsumer, ExistingConsumer)

### Description
Errors encountered during the initialization of various consumer types.

### Error Conditions
- **Stream creation failed**: AddStream/CreateStream rejected
- **Ephemeral Consumer creation failed**: CreateOrUpdateConsumer failed
- **Grouped Consumer creation failed**: CreateOrUpdateConsumer failed
- **Existing Stream lookup failed**: Stream() call failed
- **Existing Consumer lookup failed**: Stream.Consumer() call failed
```

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

### Configure NATS Subscriber Retry Backoff

Source: https://github.com/threedotslabs/watermill-nats/blob/master/_autodocs/usage-patterns.md

Use StaticDelay for fixed-interval retries or MaxRetryDelay to limit the number of attempts with a specific delay.

```go
config := nats.SubscriberConfig{
    URL:      "nats://localhost:4222",
    NakDelay: nats.NewStaticDelay(5 * time.Second),
    // ... other config
}

sub, _ := nats.NewSubscriber(config, logger)
// Failed messages retry after 5 seconds
```

```go
config := nats.SubscriberConfig{
    URL:      "nats://localhost:4222",
    NakDelay: nats.NewMaxRetryDelay(10*time.Second, 5),
    // ... other config
}

sub, _ := nats.NewSubscriber(config, logger)
// Failed messages retry up to 5 times with 10s delay
// After 5 retries, message is terminated (dead-lettered)
```