Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@ services:
| `tls_key` | string | `""` | Path to TLS private key |
| `tls_ca_cert` | string | `""` | Path to TLS CA certificate |
| `tls_insecure` | bool | `false` | Skip TLS certificate verification |
| `domain` | string | `""` | JetStream domain; scopes API calls to `$JS.<domain>.API`. Needed when NATS is reached across a leafnode boundary (optional) |

## Built-in Functions

Expand Down
7 changes: 7 additions & 0 deletions pkg/natsstore/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,13 @@ type Config struct {
TLSCACert string `json:"tls_ca_cert,omitempty"`
TLSInsecure bool `json:"tls_insecure,omitempty"`

// Domain is an optional JetStream domain. When set, JetStream API calls are
// scoped to $JS.<domain>.API instead of the default $JS.API. Required when
// the target cluster is reached across a leafnode boundary; harmless for
// direct connections since the server serves its own domain prefix locally.
// Empty preserves the default (domain-less) prefix.
Domain string `json:"domain,omitempty"`

// Cache settings
TTL Duration `json:"ttl"`
RefreshInterval Duration `json:"refresh_interval"`
Expand Down
2 changes: 2 additions & 0 deletions pkg/natsstore/config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -229,6 +229,7 @@ func TestConfig_JSONMarshaling(t *testing.T) {
ReconnectWait: Duration(1 * time.Second),
MaxBucketsWatchers: 20,
RootTenant: "test-bucket",
Domain: "cloud",
}

// Marshal to JSON
Expand All @@ -248,4 +249,5 @@ func TestConfig_JSONMarshaling(t *testing.T) {
assert.Equal(t, config.MaxBucketsWatchers, unmarshaled.MaxBucketsWatchers)
assert.Equal(t, config.Bucket, unmarshaled.Bucket)
assert.Equal(t, config.RootTenant, unmarshaled.RootTenant)
assert.Equal(t, config.Domain, unmarshaled.Domain)
}
17 changes: 12 additions & 5 deletions pkg/natsstore/nats_client.go
Original file line number Diff line number Diff line change
Expand Up @@ -101,15 +101,22 @@ func (nc *NATSClient) connect() error {
return fmt.Errorf("failed to connect to NATS server: %w", err)
}

nc.conn = conn
nc.setConnected(true)

// Create JetStream context
js, err := conn.JetStream()
// Create JetStream context, scoped to the configured domain when set
var jsOpts []nats.JSOpt
if nc.config.Domain != "" {
jsOpts = append(jsOpts, nats.Domain(nc.config.Domain))
}
js, err := conn.JetStream(jsOpts...)
if err != nil {
// Close the connection so a JetStream init failure doesn't leak it;
// only mark the client connected once setup fully succeeds.
conn.Close()
return fmt.Errorf("failed to create JetStream context: %w", err)
}

nc.conn = conn
nc.js = js
nc.setConnected(true)

nc.logger.Info("Connected to NATS at %s", nc.config.ServerURL)
return nil
Expand Down
Loading