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 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 Log(ctx context.Context, js nats.JetStream, trackingID string, ...) error
• func Process(ctx context.Context, js nats.JetStreamContext, log *zap.Logger, ...) error
• func Save(wf nats.KeyValue, k string, v []byte) error
• func SaveObj(_ context.Context, wf nats.KeyValue, k string, v proto.Message) error
• func UpdateObj[T proto.Message](ctx context.Context, wf nats.KeyValue, k string, msg T, ...) error
• type NatsConn
• type TrackingID
  • func (t TrackingID) Ancestor(gen int) string
  • func (t TrackingID) ID() string
  • func (t TrackingID) ParentID() string
  • func (t TrackingID) Pop() TrackingID
  • func (t TrackingID) Push(id string) TrackingID

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 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 Log

func Log(ctx context.Context, js nats.JetStream, trackingID string, source model.LogSource, severity messages.WorkflowLogLevel, code int32, message string, attrs map[string]string) 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)) 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 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)
}

type TrackingID

type TrackingID []string

func (TrackingID) Ancestor

func (t TrackingID) Ancestor(gen int) string

func (TrackingID) ID

func (t TrackingID) ID() string

func (TrackingID) ParentID

func (t TrackingID) ParentID() string

func (TrackingID) Pop

func (t TrackingID) Pop() TrackingID

func (TrackingID) Push

func (t TrackingID) Push(id string) TrackingID