nats

Documentation

Index

• Constants
• func AcquireNATSConnection(ctx context.Context, cfg *v1beta1.JetStreamStorageSpec, ...) (*nats.Conn, error)
• func BuildK8sServiceUrl(name string, namespace string, port ...int32) string
• func ExternalNatsObjects(ctx context.Context, k8sClient client.Client, ...) (envVars []corev1.EnvVar, volumeMounts []corev1.VolumeMount, ...)
• func NatsObjectNameFromURL(url string) string
• func NewDurableReplayConsumer(mgr nats.JetStreamContext, streamName string, ...) error
• func NewPersistentStream(mgr nats.JetStreamContext, streamConfig *nats.StreamConfig) error
• type NatsAcquireOption
  • func WithCreateStreams(streamNames ...*nats.StreamConfig) NatsAcquireOption
  • func WithLogger(lg *slog.Logger) NatsAcquireOption
  • func WithNatsOptions(opts []nats.Option) NatsAcquireOption
  • func WithRetrier(retrier backoffv2.Policy) NatsAcquireOption

Constants

const (
	DefaultImage               = "bitnami/nats:2.3.2-debian-10-r0"
	DefaultConfigReloaderImage = "connecteverything/nats-server-config-reloader:0.4.5-v1alpha2"
	DefaultPidFilePath         = "/opt/nats/pid"
	DefaultPidFileName         = "nats-server.pid"
	DefaultConfigPath          = "/opt/nats/conf"
	ConfigFileName             = "nats-config.conf"
	DefaultClientPort          = 4222
	DefaultClusterPort         = 6222
	DefaultHTTPPort            = 8222
	NkeyDir                    = "/etc/nkey"
	NkeySeedFilename           = "seed"
)

Functions

func AcquireNATSConnection

func AcquireNATSConnection(
	ctx context.Context,
	cfg *v1beta1.JetStreamStorageSpec,
	opts ...NatsAcquireOption,
) (*nats.Conn, error)

!!blocking

default retrier policy :

retrier := backoffv2.Exponential(

backoffv2.WithMaxRetries(0),
backoffv2.WithMinInterval(5*time.Second),
backoffv2.WithMaxInterval(1*time.Minute),
backoffv2.WithMultiplier(1.1),

)

default nats client options:

SeedKey : os.Getenv("NKEY_SEED_FILENAME")

nats.MaxReconnects(-1), nats.CustomReconnectDelay(

func(i int) time.Duration {
	if i == 1 {
		retryBackoff.Reset()
	}
	return retryBackoff.NextBackOff()
},

), nats.DisconnectErrHandler(

func(nc *nats.Conn, err error) {
	lg.Error(err)
},

),

func BuildK8sServiceUrl

func BuildK8sServiceUrl(name string, namespace string, port ...int32) string

func ExternalNatsObjects

func ExternalNatsObjects(
	ctx context.Context,
	k8sClient client.Client,
	namespacedNats types.NamespacedName,
) (
	envVars []corev1.EnvVar,
	volumeMounts []corev1.VolumeMount,
	volumes []corev1.Volume,
)

func NatsObjectNameFromURL

func NatsObjectNameFromURL(url string) string

func NewDurableReplayConsumer

func NewDurableReplayConsumer(mgr nats.JetStreamContext, streamName string, consumerConfig *nats.ConsumerConfig) error

func NewPersistentStream

func NewPersistentStream(mgr nats.JetStreamContext, streamConfig *nats.StreamConfig) error

Types

type NatsAcquireOption

type NatsAcquireOption func(*natsAcquireOptions)

func WithCreateStreams

func WithCreateStreams(streamNames ...*nats.StreamConfig) NatsAcquireOption

func WithLogger

func WithLogger(lg *slog.Logger) NatsAcquireOption

func WithNatsOptions

func WithNatsOptions(opts []nats.Option) NatsAcquireOption

func WithRetrier

func WithRetrier(retrier backoffv2.Policy) NatsAcquireOption