cf_vpq

package module

v0.0.7 Latest Latest Go to latest Published: Aug 16, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

README ¶

caerus-framework-vpq

Caerus Framework Valkey Priority Queue Component. A weighted priority queue backed by valkey-go, sharing the connection of the caerus-framework-valkey component. The weight of an item is the number of times it was added since it last left the queue; consumers always pop the highest-weighted item first.

Not a general job queue (no DLQ/cron/dashboard out of the box). For retries, DLQ, scheduling, or UIs use River/asynq/NATS (or similar), not this component.

Wiring

Chassis valkey is usually declared in main. Product queues (interest heat, orders, …) are typically app-owned: construct them in the app’s New, wire WithHandler there, and expose them via cf.Subcomponents so the framework registers them (see caerus-framework-demoapp). Sibling fw.AddComponent(queue) in main still works for simple binaries:

fw := cf.New()

logs := cf_logs.New(cf_logs.WithWriter(os.Stdout))
fw.AddComponent(logs) // "logs" is a required dependency

valkey := cf_valkey.New(cf_valkey.WithAddress("127.0.0.1:6379"))
queue := cf_vpq.New(
	cf_vpq.WithQueueName("orders"),
	cf_vpq.WithHandler(func(item *cf_vpq.BGetObject) error {
		return processOrder(item.ObjectID, item.ObjectValue)
	}),
)
fw.AddComponent(valkey)
fw.AddComponent(queue) // GetDependencies() -> [valkey logs]

if err := fw.Run(context.Background()); err != nil {
	log.Fatal(err)
}

The queue is a cf.Runnable: when a handler is set, Run consumes items until the framework shuts down. A failed handler requeues the item (weight +1). With a handler, Run also recovers abandoned in-flight items every 30s by default (WithRecoverInterval(0) to disable).

Usage

Manual producer/consumer:

queue := cf.MustGet[*cf_vpq.PriorityQueue](fw)

queue.Add(ctx, "order-1", `{"amount": 42}`) // returns (true, nil); weight 1
queue.Add(ctx, "order-1", `{"amount": 42}`) // returns (false, nil); weight 2, payload kept

n, _ := queue.Count(ctx) // 1 (distinct ids)

item, err := queue.BlockingBGet(ctx) // pops highest weight, blocks up to 1s
if item == nil { /* timeout, nothing to pop */ }
if err := process(item); err != nil {
	queue.Requeue(ctx, item.ObjectID) // back to the queue, weight +1
} else {
	queue.Ack(ctx, item.ObjectID) // drop deadlock tracking + payload
}

Add returns whether the payload was newly stored: false means the id is already queued and the existing payload is kept (this is the "add more weight" semantic — it does not overwrite).

Options

Option	Description
`WithConfig(PQConfig)`	static queue config snapshot; non-zero fields override option-set defaults
`WithConfigSource(name, path, …)`	bind a configuration source for Init + `OnConfigReload` tunables; the module registers the `Source[PQConfig]` itself (declares `configuration` dep)
`WithQueueName(name)`	queue name (required); part of the pub/sub channel and key namespace
`WithKeyPrefix(prefix)`	key namespace prefix (default `""` → `squeue:<queue>:<id>`, `zqueue:<queue>`, `pqdeadlocks:<queue>`; the pub/sub channel is `prefix + queue`)
`WithBlockDuration(d)`	blocking pop wait (default `1s`)
`WithPublishWatermarkDelay(d)`	min interval between pub/sub notifications on Add (default `0` = off; consumers then poll)
`WithCacheTimeout(d)`	max queue residence time (default `0` = unlimited); purges zqueue+payload together (no payload-only EXPIRE)
`WithPollInterval(d)`	auto-consumer fallback poll interval (default `1s`)
`WithHandler(Handler)`	auto-consumer callback used by `Run`; enables default 30s recover ticker and default Health thresholds
`WithWorkers(n)`	concurrent auto-consumer goroutines (default `1`); handlers must be concurrency-safe when `n > 1`
`WithRecoverInterval(d)`	how often `Run` calls `RecoverDeadlocked` (default `30s` with handler; `0` = off)
`WithRecoverMaxAge(d)`	minimum in-flight age before recovery (default `5m`; `0` = recover all in-flight)
`WithMaxDepth(n)`	fail `Health` when queued ids exceed `n`; with handler default `10000`; explicit `0` = off
`WithMaxInFlight(n)`	fail `Health` when unacked pops exceed `n`; with handler default `max(64, workers*16)`; explicit `0` = off
`WithName(name)`	custom component name for multiple instances (default `"vpq"`)
`WithLogger(*slog.Logger)`	explicit logger override; defaults to the framework `logs` component's logger (re-delivered on `logs` `Reconfigure`), falling back to `slog.Default()`

Multiple instances

Use WithName to run multiple VPQ queues in the same process (e.g., email and billing):

email := cf_vpq.New(
    cf_vpq.WithName("email-queue"),
    cf_vpq.WithQueueName("email"),
    cf_vpq.WithHandler(func(item *cf_vpq.BGetObject) error {
        return sendEmail(item.ObjectID, item.ObjectValue)
    }),
)
billing := cf_vpq.New(
    cf_vpq.WithName("billing-queue"),
    cf_vpq.WithQueueName("billing"),
    cf_vpq.WithHandler(func(item *cf_vpq.BGetObject) error {
        return processInvoice(item.ObjectID, item.ObjectValue)
    }),
)

fw.AddComponent(email)
fw.AddComponent(billing)

// Retrieve by name
emailQueue := cf.MustGetByName[*cf_vpq.PriorityQueue](fw, "email-queue")
billingQueue := cf.MustGetByName[*cf_vpq.PriorityQueue](fw, "billing-queue")

When multiple instances exist, cf.Get[*cf_vpq.PriorityQueue](fw) returns false to prevent ambiguous lookups. Always use GetByName for named instances.

Configuration

Same approach as caerus-framework-valkey: load PQConfig through the configuration component and pass it via WithConfig. Durations are in seconds:

# config.yaml
queue_name: orders
key_prefix: prod:
block_duration_sec: 5
publish_watermark_delay_sec: 1
poll_interval_sec: 1

Data model

Every multi-key mutating path is an atomic Lua script (EVALSHA with EVAL fallback).

Payload: SET NX on squeue:<queue>:<id> (kept on duplicate Add; never Redis-EXPIRE'd — see CacheTimeout below).
Priority: ZINCRBY on zqueue:<queue> per Add.
Residence index: optional zexpiry:<queue> (member → expire-at) when CacheTimeout > 0. Purge deletes zqueue member + payload together.
Wake list: pqwake:<queue> (LPUSH/LTRIM on Add/Requeue/Recover; BRPOP in BlockingBGet) so waiting is non-destructive.
Claim: one Lua script does ZPOPMAX + expiry clear + payload GET + ZADD pqdeadlocks — no pop→track crash window.
Deadlock tracking: pqdeadlocks:<queue> stores member → pop timestamp. Cleared by Ack or Requeue (back to zqueue, weight +1, payload kept).

CacheTimeout (queue residence)

When set, items may sit in the queue for at most that duration. Expiry is enforced by PurgeExpired (also from recover ticks / before claim), which removes the zset member and payload in one script. This replaces the old payload-EXPIRE model that could leave ghost queue members.

Corrupt / orphan members

A zqueue member with a missing payload (manual corruption) is dropped on claim and logged — not requeued (avoids an infinite loop). Normal CacheTimeout expiry never takes this path.

Crashed consumers

A consumer that dies between claim and Ack/Requeue leaves its item in the deadlock set (already tracked atomically at claim). With a handler, Run recovers those automatically every 30s (items older than WithRecoverMaxAge, default 5m). Pure producers leave recovery off unless you set WithRecoverInterval or call RecoverDeadlocked yourself.

Ops checklist:

Prefer WithHandler consumers (recover-by-default) or an explicit recoverer.
Set CacheTimeout only when queued items should expire as a whole.
Multiple queues: WithName + distinct WithQueueName; look up via GetByName.
Handlers should be idempotent (at-least-once after recover).

Configuration reload

The module is self-sufficient: WithConfigSource(name, path) registers its own Source[PQConfig] with the configuration component (via cf.ConfigSourceRegistrar, run by the framework during argv absorption). The default EnvPrefix is the uppercase source name ("vpq" → "VPQ_"); override with WithSourceEnvPrefix. main only points the instance at where config lives:

queue := cf_vpq.New(
	cf_vpq.WithConfigSource("vpq", "vpq.yaml"),
	cf_vpq.WithHandler(handler),
)

For low-level control, register the source manually instead:

_ = cf_configuration.AddSource(conf, cf_configuration.Source[cf_vpq.PQConfig]{
	Name:   "vpq",
	Path:   "vpq.yaml",
	Format: cf_configuration.FormatYAML,
	Owner:  queue.Name(),
})
queue := cf_vpq.New(
	cf_vpq.WithConfigSource("vpq", ""), // bind by name only
	cf_vpq.WithHandler(handler),
)

OnConfigReload re-applies tunables (poll/block/cache/recover/health thresholds). Queue name and key prefix stay fixed after Init. Valkey credential rotation is handled by the valkey component’s own WithConfigSource.

Observability

PriorityQueue implements cf.HealthProvider: Health(ctx) pings the backing valkey server and enforces WithMaxDepth / WithMaxInFlight when non-zero so /readyz can fail on backlog. With WithHandler, those thresholds default on (see options table); producers without a handler leave them off unless set. WithWorkers(n) runs n concurrent consumers (claim is atomic; make handlers safe for concurrent use). Before Init or after Shutdown Health is unhealthy.

cf.MetricsProvider samples (lazy pickup when uninitialized). All samples carry queue and component (= Name()) labels; counters are always emitted while initialized (zero until first fire):

Metric	Meaning
`vpq_info`	queue present (`queue`, `component` labels)
`vpq_depth`	distinct ids in the zset
`vpq_in_flight`	popped but unacked
`vpq_recoveries_total`	cumulative recoveries

Tests

Unit tests cover the contract without a server. Integration tests (queue mechanics, auto-consumer, requeue-on-failure, ghost handling, deadlock recovery) are gated on VALKEY_ADDR:

VALKEY_ADDR=127.0.0.1:6379 go test -race ./...

License

Apache License 2.0 — see LICENSE.

Documentation ¶

Index ¶

Constants
Variables
type BGetObject
type Handler
type InFlightItem
type Option
type PQConfig
type PriorityQueue
- func New(opts ...Option) *PriorityQueue
type SourceOption
- func WithSourceEnvPrefix(prefix string) SourceOption
- func WithSourceFormat(f cf_configuration.Format) SourceOption

Constants ¶

View Source

const (
	// ComponentName is the framework component name for the valkey priority
	// queue component.
	ComponentName = "vpq"

	// ComponentStage is the stage the queue initializes in. It is not a
	// built-in bootstrap stage; AddComponent registers it automatically the
	// first time a component declares it.
	ComponentStage = cf.Stage("data")
)

Variables ¶

View Source

var ErrClosed = errors.New("cf_vpq: queue is not initialized or is shut down")

ErrClosed is returned by queue operations after Shutdown or before Init.

Functions ¶

This section is empty.

Types ¶

type BGetObject ¶

type BGetObject struct {
	ObjectID    string
	ObjectScore float64
	ObjectValue string
}

BGetObject is an item popped from the queue. ObjectScore is the item weight (the number of times it was added since it last left the queue).

type Handler ¶

type Handler func(*BGetObject) error

Handler processes one item in the auto-consumer loop. Returning an error requeues the item (weight +1) and drops its deadlock tracking.

type InFlightItem ¶

type InFlightItem struct {
	ObjectID string
	PoppedAt time.Time
}

InFlightItem is an item that was popped (removed from the queue) but not yet acked or requeued.

type Option ¶

type Option func(*options)

Option configures the queue at construction time.

func WithBlockDuration ¶

func WithBlockDuration(d time.Duration) Option

WithBlockDuration sets how long a blocking pop waits for an item (default 1s).

func WithCacheTimeout ¶

func WithCacheTimeout(d time.Duration) Option

WithCacheTimeout sets how long an item may remain queued before it is discarded (default 0 = no residence limit). Expiry removes the zqueue member and payload together; the payload key itself is never Redis-EXPIRE'd, so partial "ghost" expiry cannot occur. In-flight items are unaffected.

func WithConfig ¶

func WithConfig(cfg PQConfig) Option

WithConfig sets a static queue configuration snapshot. Non-zero fields of cfg override the values set by the convenience options. Prefer WithConfigSource when using caerus-framework-configuration with hot-reload.

func WithConfigSource ¶

func WithConfigSource(name, path string, opts ...SourceOption) Option

WithConfigSource binds this component to a named configuration source and registers that source with the configuration component (via the framework's ConfigSourceRegistrar pass during argv absorption). The module owns the Source: the config type, default EnvPrefix and its Owner (Name(), so named instances reload correctly). main only points the instance at where the config lives.

cf_vpq.New(cf_vpq.WithConfigSource("vpq", "config/vpq.json"), ...)

A path of "" registers an env-only (fileless) source when the EnvPrefix is non-empty. The path CLI override stays --<source-name> (ParseFlags). Declares a dependency on "configuration". Queue identity (name/prefix) is applied at Init; reload updates tunables only (durations, recover, health thresholds).

func WithHandler ¶

func WithHandler(h Handler) Option

WithHandler sets the auto-consumer callback used by Run. When set, deadlock recovery runs every 30s by default (see WithRecoverInterval), and Health gets default MaxDepth / MaxInFlight thresholds (see WithMaxDepth / WithMaxInFlight). Handlers must be safe for concurrent use when WithWorkers(n) has n > 1.

func WithKeyPrefix ¶

func WithKeyPrefix(prefix string) Option

WithKeyPrefix sets a key namespace prefix (default ""). With the default the keys are "squeue:<queue>:<id>", "zqueue:<queue>" and "pqdeadlocks:<queue>".

func WithLogger ¶

func WithLogger(logger *slog.Logger) Option

WithLogger overrides the logger used for component diagnostics. By default the component logs through the framework logs component (declared in GetDependencies); WithLogger is an explicit override for tests and embedded use and wins over the framework logger. slog.Default() remains the fallback only when neither is available.

func WithMaxDepth ¶

func WithMaxDepth(n int64) Option

WithMaxDepth fails Health when the queued item count exceeds n. With a handler, the default is 10000 when this option is omitted; pass 0 to disable the check explicitly.

func WithMaxInFlight ¶

func WithMaxInFlight(n int64) Option

WithMaxInFlight fails Health when popped-but-unacked items exceed n. With a handler, the default is max(64, workers*16) when this option is omitted; pass 0 to disable the check explicitly.

func WithName ¶

func WithName(name string) Option

WithName sets a custom component name, allowing multiple VPQ instances in the same process. The default name is "vpq" (ComponentName). Use this when you need multiple queues (e.g., email and billing) in one binary. Retrieve named instances with GetByName[*PriorityQueue](fw, "email").

func WithPollInterval ¶

func WithPollInterval(d time.Duration) Option

WithPollInterval sets the fallback consumer poll interval (default 1s).

func WithPublishWatermarkDelay ¶

func WithPublishWatermarkDelay(d time.Duration) Option

WithPublishWatermarkDelay sets the minimum interval between pub/sub notifications on Add (default 0 = no pub/sub; consumers then rely on the poll interval).

func WithQueueName ¶

func WithQueueName(name string) Option

WithQueueName sets the queue name. It is the pub/sub channel and the key namespace segment; it is required.

func WithRecoverInterval ¶

func WithRecoverInterval(d time.Duration) Option

WithRecoverInterval sets how often Run calls RecoverDeadlocked. With a handler, the default is 30s. Pass 0 to disable. Without a handler, recovery stays off unless this sets a positive interval (dedicated recoverer).

func WithRecoverMaxAge ¶

func WithRecoverMaxAge(d time.Duration) Option

WithRecoverMaxAge sets the minimum age of in-flight items before RecoverDeadlocked requeues them (default 5m). Pass 0 to recover all currently in-flight items on each tick.

func WithWorkers ¶

func WithWorkers(n int) Option

WithWorkers sets how many concurrent auto-consumer goroutines Run/Consume use (default 1). Values below 1 are treated as 1. Claim is atomic; each worker runs the handler independently — handlers must be concurrency-safe when n > 1.

type PQConfig ¶

type PQConfig struct {
	QueueName             string `json:"queue_name" yaml:"queue_name" env:"QUEUE_NAME"`
	KeyPrefix             string `json:"key_prefix,omitempty" yaml:"key_prefix,omitempty" env:"KEY_PREFIX"`
	BlockDuration         int    `json:"block_duration_sec,omitempty" yaml:"block_duration_sec,omitempty" env:"BLOCK_DURATION_SEC"`
	PublishWatermarkDelay int    `json:"publish_watermark_delay_sec,omitempty" yaml:"publish_watermark_delay_sec,omitempty" env:"PUBLISH_WATERMARK_DELAY_SEC"`
	CacheTimeout          int    `json:"cache_timeout_sec,omitempty" yaml:"cache_timeout_sec,omitempty" env:"CACHE_TIMEOUT_SEC"`
	PollInterval          int    `json:"poll_interval_sec,omitempty" yaml:"poll_interval_sec,omitempty" env:"POLL_INTERVAL_SEC"`
	RecoverInterval       int    `json:"recover_interval_sec,omitempty" yaml:"recover_interval_sec,omitempty" env:"RECOVER_INTERVAL_SEC"`
	RecoverMaxAge         int    `json:"recover_max_age_sec,omitempty" yaml:"recover_max_age_sec,omitempty" env:"RECOVER_MAX_AGE_SEC"`
	MaxDepth              int    `json:"max_depth,omitempty" yaml:"max_depth,omitempty" env:"MAX_DEPTH"`
	MaxInFlight           int    `json:"max_in_flight,omitempty" yaml:"max_in_flight,omitempty" env:"MAX_IN_FLIGHT"`
	Workers               int    `json:"workers,omitempty" yaml:"workers,omitempty" env:"WORKERS"`
}

PQConfig is the file/env-drivable queue configuration. Load it through the configuration component via WithConfigSource (preferred) or WithConfig. Durations are in seconds; zero means "unset" (keep option defaults).

type PriorityQueue ¶

type PriorityQueue struct {
	// contains filtered or unexported fields
}

PriorityQueue is the caerus-framework-vpq component: a weighted priority queue backed by valkey. Add increments the item weight; consumers pop the highest-weighted item. It depends on the valkey component for its client.

func New ¶

func New(opts ...Option) *PriorityQueue

New creates a priority queue component. It requires a valkey component and a queue name at Init.

func (*PriorityQueue) Ack ¶

func (q *PriorityQueue) Ack(ctx context.Context, id string) error

Ack completes a popped item: it removes the deadlock tracking, deletes the payload and clears any lingering queue/expiry member. Idempotent; safe to call for already-acked ids.

func (*PriorityQueue) Add ¶

func (q *PriorityQueue) Add(ctx context.Context, id, value string) (bool, error)

Add enqueues an item and returns whether its payload was newly stored. A false result means the id is already queued: the existing payload is kept and the item weight is incremented (this is the "add more weight" semantic; it does not overwrite). When a watermark delay is configured, a pub/sub notification is throttled to that interval; notification failures are logged but do not fail the add.

func (*PriorityQueue) BlockingBGet ¶

func (q *PriorityQueue) BlockingBGet(ctx context.Context) (*BGetObject, error)

BlockingBGet pops and returns the highest-weighted item, waiting up to the block duration. It returns (nil, nil) when the wait expires with no item. Pop and deadlock tracking are a single atomic Lua claim (no crash window). Waiting uses a wake list (BRPOP), not BZPOPMAX, so the wait cannot orphan an item.

func (*PriorityQueue) Consume ¶

func (q *PriorityQueue) Consume(ctx context.Context, handler Handler) error

Consume runs the auto-consumer until ctx is canceled: a pub/sub wakeup triggers an immediate pass, otherwise the queue is polled at the poll interval. Uses WithWorkers concurrent loops sharing one subscription. A failed handler is logged and the item requeued. Handlers must be safe for concurrent use when workers > 1.

func (*PriorityQueue) Count ¶

func (q *PriorityQueue) Count(ctx context.Context) (int64, error)

Count returns the number of queued items (distinct ids).

func (*PriorityQueue) Deadlocked ¶

func (q *PriorityQueue) Deadlocked(ctx context.Context) ([]InFlightItem, error)

Deadlocked returns the items currently in flight (popped but not acked or requeued), with the time each was popped. A consumer that crashed between a successful pop and its Ack/Requeue leaves its item listed here until RecoverDeadlocked returns it to the queue.

func (*PriorityQueue) GetDependencies ¶

func (q *PriorityQueue) GetDependencies() []string

GetDependencies implements cf.Dependencies. Depends on configuration when WithConfigSource is set.

func (*PriorityQueue) GetInitOrderStage ¶

func (q *PriorityQueue) GetInitOrderStage() cf.Stage

GetInitOrderStage implements cf.CaerusComponent.

func (*PriorityQueue) Health ¶

func (q *PriorityQueue) Health(ctx context.Context) error

Health implements cf.HealthProvider. It pings the backing valkey server and, when thresholds are non-zero, fails if queue depth or in-flight count exceed them (WithMaxDepth / WithMaxInFlight; defaults apply when WithHandler is set). A nil client is unhealthy.

func (*PriorityQueue) InFlightCount ¶

func (q *PriorityQueue) InFlightCount(ctx context.Context) (int64, error)

InFlightCount returns the number of popped but unacked items (deadlock set).

func (*PriorityQueue) Init ¶

func (q *PriorityQueue) Init(ctx context.Context, fw *cf.CaerusFramework) error

Init implements cf.CaerusComponent. It resolves the valkey client and validates the queue name, failing fast before the framework starts runners.

func (*PriorityQueue) IntCount ¶

func (q *PriorityQueue) IntCount(ctx context.Context) int64

IntCount returns the number of queued items, or 0 on error.

func (*PriorityQueue) Metrics ¶

func (q *PriorityQueue) Metrics() []cf_observability.Metric

Metrics implements cf_observability.MetricsProvider. While initialized it reports info, depth, in-flight, and cumulative recoveries; before Init or after Shutdown it returns nil (observability lazy pickup).

func (*PriorityQueue) Name ¶

func (q *PriorityQueue) Name() string

Name implements cf.CaerusComponent. Name implements cf.CaerusComponent. Returns the custom name set via WithName, or the default ComponentName ("vpq") if no custom name was set.

func (*PriorityQueue) OnConfigReload ¶

func (q *PriorityQueue) OnConfigReload(source string, cfg any)

OnConfigReload implements cf.ConfigReloader. It re-reads queue tunables from the bound configuration source. Queue name and key prefix are not changed. Credential rotation for the shared client is handled by the valkey component.

func (*PriorityQueue) PurgeExpired ¶

func (q *PriorityQueue) PurgeExpired(ctx context.Context) (int64, error)

PurgeExpired removes queued items whose CacheTimeout residence has elapsed, deleting the zqueue member and payload together. It is a no-op when CacheTimeout is 0. Safe to call concurrently; also run from recover ticks and before each claim.

func (*PriorityQueue) RecoverDeadlocked ¶

func (q *PriorityQueue) RecoverDeadlocked(ctx context.Context, maxAge time.Duration) (int64, error)

RecoverDeadlocked requeues every in-flight item that has been unacked for longer than maxAge, restoring it to the queue with weight +1. It returns the number of items recovered. Run invokes this on WithRecoverInterval (default 30s when a handler is set); it may also be called manually.

func (*PriorityQueue) RegisterConfigSources ¶

func (q *PriorityQueue) RegisterConfigSources(conf any) error

RegisterConfigSources implements cf.ConfigSourceRegistrar. The framework calls it during argv absorption; it registers this component's configuration source (name, path, env prefix, format and Owner) with the configuration component. No-op when no source is bound.

func (*PriorityQueue) Requeue ¶

func (q *PriorityQueue) Requeue(ctx context.Context, id string) error

Requeue returns a popped item to the queue with weight +1 and drops its deadlock tracking. Call it after a manual consumer failed to process the item; the payload is preserved. CacheTimeout residence is refreshed.

func (*PriorityQueue) Run ¶

func (q *PriorityQueue) Run(ctx context.Context) error

Run implements cf.Runnable. It runs the auto-consumer (when WithHandler is set) and/or the deadlock-recovery ticker until ctx is canceled.

func (*PriorityQueue) Shutdown ¶

func (q *PriorityQueue) Shutdown(ctx context.Context) error

Shutdown implements cf.CaerusComponent. It stops serving; the shared valkey client is owned and closed by the valkey component.

type SourceOption ¶

type SourceOption func(*sourceOptions)

SourceOption configures the self-registered configuration source created by WithConfigSource.

func WithSourceEnvPrefix ¶

func WithSourceEnvPrefix(prefix string) SourceOption

WithSourceEnvPrefix sets the environment overlay prefix for the source (default: the uppercase source name with "-" replaced by "_", plus "_"). An empty prefix disables env overlay.

func WithSourceFormat ¶

func WithSourceFormat(f cf_configuration.Format) SourceOption

WithSourceFormat forces the file format instead of inferring it from the path extension (".yaml"/".yml" → YAML; anything else JSON).

Source Files ¶

View all Source files

vpq.go

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL