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(ctx context.Context, wf nats.KeyValue, k string) ([]byte, error)
• func LoadObj(ctx context.Context, 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, traceName string, ...) error
• func Save(ctx context.Context, wf nats.KeyValue, k string, v []byte) error
• func SaveObj(ctx 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

Delete deletes an item from a key value store.

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

EnsureBuckets ensures that a list of key value stores exist

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

KSuidTo128bit returns a KSuid as bytes.

func KSuidTo64bit

func KSuidTo64bit(k string) [8]byte

KSuidTo64bit takes the most variable 64 bits of a KSuid and returns them as bytes.

func Load

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

Load loads a value from a key value store

func LoadObj

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

LoadObj loads a protobuf message from a key value store

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

Log is the generic metod to output to SHAR telemetry.

func Process

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

Process processes messages from a nats consumer and executes a function against each one.

func Save

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

Save saves a value to a key value store

func SaveObj

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

SaveObj save an protobuf message to a key value store

func UpdateObj

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

UpdateObj saves an protobuf message to a key value store after using updateFN to update the message.

Types

type NatsConn

type NatsConn interface {
	JetStream(opts ...nats.JSOpt) (nats.JetStreamContext, error)
	QueueSubscribe(subj string, queue string, cb nats.MsgHandler) (*nats.Subscription, error)
}

NatsConn is the trimmad down NATS Connection interface that only emcompasses the methods used by SHAR

type TrackingID

type TrackingID []string

TrackingID is an ID stack that maintains the callstack

func (TrackingID) Ancestor

func (t TrackingID) Ancestor(gen int) string

Ancestor provides the ID of the caller back <gen> generations.

func (TrackingID) ID

func (t TrackingID) ID() string

ID provides the current ID

func (TrackingID) ParentID

func (t TrackingID) ParentID() string

ParentID provides the ID of the caller.

func (TrackingID) Pop

func (t TrackingID) Pop() TrackingID

Pop removes the current ID from the callstack.

func (TrackingID) Push

func (t TrackingID) Push(id string) TrackingID

Push adds a new ID to the callstack.