common

Documentation

Index

• func EnsureBuckets(js nats.JetStreamContext, storageType nats.StorageType, names []string) error
• func EnsureConsumer(js nats.JetStreamContext, streamName string, ...) error
• func EnsureStream(js nats.JetStreamContext, streamConfig *nats.StreamConfig) error
• func Hex(b []byte) string
• func KSuidTo128bit(k string) [16]byte
• func KSuidTo64bit(k string) [8]byte
• func Load(wf nats.KeyValue, k string) ([]byte, error)
• func LoadObj(wf nats.KeyValue, k string, v proto.Message) error
• func Process(ctx context.Context, js nats.JetStreamContext, log *zap.Logger, ...)
• func Save(wf nats.KeyValue, k string, v []byte) error
• func SaveObj(ctx context.Context, wf nats.KeyValue, k string, v proto.Message) error
• func UpdateKV(wf nats.KeyValue, k string, msg proto.Message, ...) error
• func UpdateObj[T proto.Message](wf nats.KeyValue, k string, msg T, updateFn func(v T) (T, error)) error
• type NatsConn

Functions

func EnsureBuckets

func EnsureBuckets(js nats.JetStreamContext, storageType nats.StorageType, names []string) error

func EnsureConsumer

func EnsureConsumer(js nats.JetStreamContext, streamName string, consumerConfig *nats.ConsumerConfig) error

func EnsureStream

func EnsureStream(js nats.JetStreamContext, streamConfig *nats.StreamConfig) error

func Hex

func Hex(b []byte) string

func KSuidTo128bit

func KSuidTo128bit(k string) [16]byte

func KSuidTo64bit

func KSuidTo64bit(k string) [8]byte

func Load

func Load(wf nats.KeyValue, k string) ([]byte, error)

func LoadObj

func LoadObj(wf nats.KeyValue, k string, v proto.Message) error

func Process

func Process(ctx context.Context, js nats.JetStreamContext, log *zap.Logger, closer chan struct{}, subject string, durable string, concurrency int, fn func(ctx context.Context, msg *nats.Msg) (bool, error))

func Save

func Save(wf nats.KeyValue, k string, v []byte) error

func SaveObj

func SaveObj(ctx context.Context, wf nats.KeyValue, k string, v proto.Message) error

func UpdateKV

func UpdateKV(wf nats.KeyValue, k string, msg proto.Message, updateFn func(v []byte, msg proto.Message) ([]byte, error)) error

func UpdateObj

func UpdateObj[T proto.Message](wf nats.KeyValue, k string, msg T, updateFn func(v T) (T, error)) error

Types

type NatsConn

type NatsConn interface {
	JetStream(opts ...nats.JSOpt) (nats.JetStreamContext, error)
	Subscribe(subject string, fn nats.MsgHandler) (*nats.Subscription, error)
}