stream

Documentation

Overview

Package stream creates and manages ebind's JetStream streams.

Index

• Constants
• func EnsureStreams(ctx context.Context, js jetstream.JetStream, cfg Config) error
• func TargetToken(target string) string
• func TargetedTaskFilter(target string) string
• func TargetedTaskSubject(target, name string) string
• func TaskPublishSubject(name, target string) string
• func TaskSubject(name string) string
• func WaitMetaLeader(ctx context.Context, js jetstream.JetStream) error
• type Config

Constants

const (
	TaskStream     = "EBIND_TASKS"
	ResponseStream = "EBIND_RESP"
	DLQStream      = "EBIND_DLQ"

	TaskSubjectPrefix         = "TASKS."
	TargetedTaskSubjectPrefix = "TASKS_TARGET."
	ResponseSubjectPrefix     = "RESP."
	DLQSubjectPrefix          = "DLQ."
)

Functions

func EnsureStreams

func EnsureStreams(ctx context.Context, js jetstream.JetStream, cfg Config) error

EnsureStreams creates or updates all three ebind streams idempotently.

func TargetToken

func TargetToken(target string) string

func TargetedTaskFilter

func TargetedTaskFilter(target string) string

func TargetedTaskSubject

func TargetedTaskSubject(target, name string) string

func TaskPublishSubject

func TaskPublishSubject(name, target string) string

func TaskSubject

func TaskSubject(name string) string

func WaitMetaLeader

func WaitMetaLeader(ctx context.Context, js jetstream.JetStream) error

WaitMetaLeader polls until the JetStream meta-leader is elected or ctx expires. Needed on cluster startup before any stream operation.

Types

type Config

type Config struct {
	Replicas        int           // 1 for dev, 3 for HA
	TaskMaxAge      time.Duration // default 24h
	ResponseMaxAge  time.Duration // default 1h
	DLQMaxAge       time.Duration // default 7d
	DuplicateWindow time.Duration // default 5m
}