common

Documentation

Index

• func Delete(kv nats.KeyValue, key string) error
• func ElementTable(process *model.Workflow) map[string]*model.Element
• 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 IndexProcessElements(elements []*model.Element, el map[string]*model.Element)
• 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(_ context.Context, wf nats.KeyValue, k string, v proto.Message) error
• func SetUpNats(js nats.JetStreamContext, storageType nats.StorageType) error
• func UpdateKV(wf nats.KeyValue, k string, msg proto.Message, ...) error
• func UpdateObj[T proto.Message](ctx context.Context, wf nats.KeyValue, k string, msg T, ...) error
• type NatsConn

Functions

func Delete

func Delete(kv nats.KeyValue, key string) error

func ElementTable

func ElementTable(process *model.Workflow) map[string]*model.Element

ElementTable indexes an entire process for quick ID lookups

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 IndexProcessElements

func IndexProcessElements(elements []*model.Element, el map[string]*model.Element)

IndexProcessElements is the recursive part of the index

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, traceName string, 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(_ context.Context, wf nats.KeyValue, k string, v proto.Message) error

func SetUpNats

func SetUpNats(js nats.JetStreamContext, storageType nats.StorageType) 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](ctx context.Context, 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)
	QueueSubscribe(subj string, queue string, cb nats.MsgHandler) (*nats.Subscription, error)
}