common

func Delete ¶ added in v0.1.192

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

func ElementTable ¶ added in v0.1.134

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

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

func EnsureStream ¶ added in v0.1.78

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

func IndexProcessElements ¶ added in v0.1.134

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

IndexProcessElements is the recursive part of the index

func KSuidTo64bit ¶ added in v0.1.78

func KSuidTo64bit(k string) [8]byte

func KSuidTo128bit ¶ added in v0.1.78

func KSuidTo128bit(k string) [16]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 ¶ added in v0.1.78

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

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