nats

package

v0.12.1 Latest Latest Go to latest Published: Nov 17, 2023 License: Apache-2.0 Imports: 15 Imported by: 0

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/rancher/opni

Links

Open Source Insights

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

Constants ¶

View Source

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"
)

Variables ¶

This section is empty.

Functions ¶

func AcquireNATSConnection ¶ added in v0.6.1

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 ¶ added in v0.9.0

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 ¶ added in v0.8.0

func NatsObjectNameFromURL(url string) string

func NewDurableReplayConsumer ¶ added in v0.8.0

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

func NewPersistentStream ¶ added in v0.6.1

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

Types ¶

type NatsAcquireOption ¶ added in v0.8.2

type NatsAcquireOption func(*natsAcquireOptions)

func WithCreateStreams ¶ added in v0.6.1

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

func WithLogger ¶ added in v0.6.1

func WithLogger(lg *slog.Logger) NatsAcquireOption

func WithNatsOptions ¶ added in v0.6.1

func WithNatsOptions(opts []nats.Option) NatsAcquireOption

func WithRetrier ¶ added in v0.6.1

func WithRetrier(retrier backoffv2.Policy) NatsAcquireOption

Source Files ¶

View all Source files

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