common

Documentation

Index

• func CopyWorkflowState(state *model.WorkflowState) *model.WorkflowState
• func Delete(kv nats.KeyValue, key string) error
• func DeleteLarge(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, k string) error
• func DropStateParams(state *model.WorkflowState)
• func ElementTable(process *model.Workflow) map[string]*model.Element
• func EnsureBucket(js nats.JetStreamContext, storageType nats.StorageType, name string, ...) error
• func EnsureBuckets(js nats.JetStreamContext, storageType nats.StorageType, names []string) error
• func ExtendLock(kv nats.KeyValue, lockID 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 LoadLarge(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, key string, ...) ([]byte, error)
• func LoadLargeObj(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, k string, ...) error
• func LoadObj(ctx context.Context, wf nats.KeyValue, k string, v proto.Message) error
• func Lock(kv nats.KeyValue, lockID string) (bool, error)
• func Log(ctx context.Context, js nats.JetStream, trackingID string, ...) error
• func Process(ctx context.Context, js nats.JetStreamContext, traceName string, ...) error
• func RemoveWhere[T comparable](slice []T, fn func(T) bool) []T
• func Save(ctx context.Context, wf nats.KeyValue, k string, v []byte) error
• func SaveLarge(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, key string, ...) error
• func SaveLargeObj(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, k string, ...) error
• func SaveObj(ctx context.Context, wf nats.KeyValue, k string, v proto.Message) error
• func UnLock(kv nats.KeyValue, lockID string) error
• func UpdateLargeObj[T proto.Message](ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, k string, msg T, ...) (T, error)
• func UpdateObj[T proto.Message](ctx context.Context, wf nats.KeyValue, k string, msg T, ...) error
• func UpdateObjIsNew[T proto.Message](ctx context.Context, wf nats.KeyValue, k string, msg T, ...) (bool, error)
• type BackoffFn
• type BackoffProcessOption
  • func WithBackoffFn(fn BackoffFn) BackoffProcessOption
  • func (b BackoffProcessOption) Set(opts *ProcessOpts)
• type NatsConn
• type ProcessOption
• type ProcessOpts
• 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 CopyWorkflowState

func CopyWorkflowState(state *model.WorkflowState) *model.WorkflowState

CopyWorkflowState - clones a proto model.WorkflowState for modification.

func Delete

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

Delete deletes an item from a key value store.

func DeleteLarge

func DeleteLarge(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, k string) error

DeleteLarge deletes a large binary from the object store

func DropStateParams

func DropStateParams(state *model.WorkflowState)

DropStateParams removes any parameters unsafe to send across a state transition.

func ElementTable

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

ElementTable indexes an entire process for quick ID lookups

func EnsureBucket

func EnsureBucket(js nats.JetStreamContext, storageType nats.StorageType, name string, ttl time.Duration) error

EnsureBucket creates a bucket if it does not exist

func EnsureBuckets

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

EnsureBuckets ensures that a list of key value stores exist

func ExtendLock

func ExtendLock(kv nats.KeyValue, lockID string) error

ExtendLock extends the lock past its stale time.

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 LoadLarge

func LoadLarge(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, key string, opt ...nats.GetObjectOpt) ([]byte, error)

LoadLarge load a large binary from the object store

func LoadLargeObj

func LoadLargeObj(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, k string, v proto.Message, opt ...nats.GetObjectOpt) error

LoadLargeObj loads a protobuf message 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 Lock

func Lock(kv nats.KeyValue, lockID string) (bool, error)

Lock ensures a lock on a given ID, it returns true if a lock was granted.

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), opts ...ProcessOption) error

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

func RemoveWhere

func RemoveWhere[T comparable](slice []T, fn func(T) bool) []T

RemoveWhere removes an array element based upon a condition.

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 SaveLarge

func SaveLarge(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, key string, data []byte, opt ...nats.ObjectOpt) error

SaveLarge saves a large binary from the object store

func SaveLargeObj

func SaveLargeObj(ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, k string, v proto.Message, opt ...nats.ObjectOpt) error

SaveLargeObj save an protobuf message to a document 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 UnLock

func UnLock(kv nats.KeyValue, lockID string) error

UnLock closes an existing lock.

func UpdateLargeObj

func UpdateLargeObj[T proto.Message](ctx context.Context, ds nats.ObjectStore, mutex nats.KeyValue, k string, msg T, updateFn func(v T) (T, error)) (T, error)

UpdateLargeObj saves an protobuf message to a document store after using updateFN to update the message.

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.

func UpdateObjIsNew

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

UpdateObjIsNew saves an protobuf message to a key value store after using updateFN to update the message, and returns true if this is a new value.

Types

type BackoffFn

type BackoffFn func(ctx context.Context, msg *nats.Msg) error

BackoffFn represents a function that completely handles the backoff for a message including ACK/NAK

type BackoffProcessOption

type BackoffProcessOption struct {
	// contains filtered or unexported fields
}

BackoffProcessOption holds the backoff function. Don't use this directly. Use the convenience function WithBackoffFn

func WithBackoffFn

func WithBackoffFn(fn BackoffFn) BackoffProcessOption

WithBackoffFn adds a back-off function to message processing

func (BackoffProcessOption) Set

func (b BackoffProcessOption) Set(opts *ProcessOpts)

Set the backoff function in the process settings

type NatsConn

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

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

type ProcessOption

type ProcessOption interface {
	Set(opts *ProcessOpts)
}

ProcessOption represents an option function that can be passed to message processing.

type ProcessOpts

type ProcessOpts struct {
	BackoffCalc BackoffFn
}

ProcessOpts holds the settings for message processing.

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.