stateful

Documentation

Index

• func TopicMissingError(topic string) error
• func WithPersistenceIdSwitchPersistenceIdFunction(topic string, f PersistenceIdFunction[[]byte, []byte]) runtime.Configuration[*PersistenceIdSwitch]
• func WithTopicSwitchFunction(topic string, f SingleFunction) runtime.Configuration[*TopicSwitch]
• type ErrorHandlerFunction
• type OneToOneFunction
• type OneToTwoFunction
• type PersistenceIdFunction
  • func ConvertPersistenceId[IK any, IV any](source PersistenceIdFunction[IK, IV], ik format.Format[IK], ...) PersistenceIdFunction[[]byte, []byte]
  • func NewPersistenceIdSwitch(configurations ...runtime.Configuration[*PersistenceIdSwitch]) PersistenceIdFunction[[]byte, []byte]
• type PersistenceIdSwitch
  • func (r *PersistenceIdSwitch) Apply(c context.Context, m flow.Message[structure.Bytes, structure.Bytes]) (string, error)
• type Repository
• type SingleFunction
  • func ConvertOneToOne[S any, IK any, IV any, OK any, OV any](source OneToOneFunction[S, IK, IV, OK, OV], s format.Format[S], ...) SingleFunction
  • func ConvertOneToTwo[S any, IK any, IV any, OK1 any, OV1 any, OK2 any, OV2 any](source OneToTwoFunction[S, IK, IV, OK1, OV1, OK2, OV2], s format.Format[S], ...) SingleFunction
  • func ConvertTopicOneToOne[S any, IK any, IV any, OK any, OV any](source OneToOneFunction[S, IK, IV, OK, OV], s format.Format[S], ...) SingleFunction
  • func ConvertTopicOneToTwo[S any, IK any, IV any, OK1 any, OV1 any, OK2 any, OV2 any](source OneToTwoFunction[S, IK, IV, OK1, OV1, OK2, OV2], s format.Format[S], ...) SingleFunction
  • func NewTopicSwitch(configurations ...runtime.Configuration[*TopicSwitch]) SingleFunction
• type State
  • func ConvertSingleState[V1 any, V2 any](source State[V1], v1 format.Format[V1], v2 format.Format[V2]) (State[V2], error)
  • func NewState[S any](id string, content S) State[S]
  • func SetDefault[S any](s State[S]) State[S]
• type TopicSwitch
  • func (r *TopicSwitch) Apply(c context.Context, m flow.Message[structure.Bytes, structure.Bytes], ...) ([]flow.Message[structure.Bytes, structure.Bytes], State[structure.Bytes], ...)

Functions

func TopicMissingError

func TopicMissingError(topic string) error

func WithPersistenceIdSwitchPersistenceIdFunction

func WithPersistenceIdSwitchPersistenceIdFunction(topic string, f PersistenceIdFunction[[]byte, []byte]) runtime.Configuration[*PersistenceIdSwitch]

configuration

func WithTopicSwitchFunction

func WithTopicSwitchFunction(topic string, f SingleFunction) runtime.Configuration[*TopicSwitch]

configuration

Types

type ErrorHandlerFunction

type ErrorHandlerFunction func(context.Context, flow.Message[structure.Bytes, structure.Bytes], State[structure.Bytes], error) ([]flow.Message[structure.Bytes, structure.Bytes], State[structure.Bytes], error)

type OneToOneFunction

type OneToOneFunction[S any, IK any, IV any, OK any, OV any] func(context.Context, flow.Message[IK, IV], State[S]) (*flow.Message[OK, OV], State[S], error)

type OneToTwoFunction

type OneToTwoFunction[S any, IK any, IV any, OK1 any, OV1 any, OK2 any, OV2 any] func(context.Context, flow.Message[IK, IV], State[S]) (*flow.Message[OK1, OV1], *flow.Message[OK2, OV2], State[S], error)

type PersistenceIdFunction

type PersistenceIdFunction[IK any, IV any] func(context.Context, flow.Message[IK, IV]) (string, error)

func ConvertPersistenceId

func ConvertPersistenceId[IK any, IV any](
	source PersistenceIdFunction[IK, IV],
	ik format.Format[IK],
	iv format.Format[IV],
) PersistenceIdFunction[[]byte, []byte]

func NewPersistenceIdSwitch

func NewPersistenceIdSwitch(configurations ...runtime.Configuration[*PersistenceIdSwitch]) PersistenceIdFunction[[]byte, []byte]

constructor

type PersistenceIdSwitch

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

implementation

func (*PersistenceIdSwitch) Apply

func (r *PersistenceIdSwitch) Apply(c context.Context, m flow.Message[structure.Bytes, structure.Bytes]) (string, error)

type Repository

type Repository interface {
	Get(ctx context.Context, persistenceId string) (State[structure.Bytes], error)
	GetAll(ctx context.Context, persistenceId []string) (map[string]State[structure.Bytes], error)
	Upsert(ctx context.Context, persistenceId string, dbState State[structure.Bytes]) error
	UpsertAll(ctx context.Context, stateMap map[string]State[structure.Bytes]) error
}

type SingleFunction

type SingleFunction func(context.Context, flow.Message[structure.Bytes, structure.Bytes], State[structure.Bytes]) ([]flow.Message[structure.Bytes, structure.Bytes], State[structure.Bytes], error)

func ConvertOneToOne

func ConvertOneToOne[S any, IK any, IV any, OK any, OV any](
	source OneToOneFunction[S, IK, IV, OK, OV],
	s format.Format[S],
	ik format.Format[IK],
	iv format.Format[IV],
	ok format.Format[OK],
	ov format.Format[OV],
) SingleFunction

func ConvertOneToTwo

func ConvertOneToTwo[S any, IK any, IV any, OK1 any, OV1 any, OK2 any, OV2 any](
	source OneToTwoFunction[S, IK, IV, OK1, OV1, OK2, OV2],
	s format.Format[S],
	ik format.Format[IK],
	iv format.Format[IV],
	ok1 format.Format[OK1],
	ov1 format.Format[OV1],
	ok2 format.Format[OK2],
	ov2 format.Format[OV2],
) SingleFunction

func ConvertTopicOneToOne

func ConvertTopicOneToOne[S any, IK any, IV any, OK any, OV any](
	source OneToOneFunction[S, IK, IV, OK, OV],
	s format.Format[S],
	inputTopic flow.Topic[IK, IV],
	outputTopic flow.Topic[OK, OV],
) SingleFunction

func ConvertTopicOneToTwo

func ConvertTopicOneToTwo[S any, IK any, IV any, OK1 any, OV1 any, OK2 any, OV2 any](
	source OneToTwoFunction[S, IK, IV, OK1, OV1, OK2, OV2],
	s format.Format[S],
	inputTopic flow.Topic[IK, IV],
	outputTopic1 flow.Topic[OK1, OV1],
	outputTopic2 flow.Topic[OK2, OV2],
) SingleFunction

func NewTopicSwitch

func NewTopicSwitch(configurations ...runtime.Configuration[*TopicSwitch]) SingleFunction

constructor

type State

type State[S any] struct {
	Id                 string
	Internal           *protobuf.State
	Results            *protobuf.Results
	Content            S
	CreatedTimestampMs int64
	UpdatedTimestampMs int64
}

func ConvertSingleState

func ConvertSingleState[V1 any, V2 any](
	source State[V1],
	v1 format.Format[V1],
	v2 format.Format[V2],
) (State[V2], error)

assuming byte compatibility, i.e. bytes <-> proto, string <-> json

func NewState

func NewState[S any](id string, content S) State[S]

func SetDefault

func SetDefault[S any](s State[S]) State[S]

type TopicSwitch

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

implementation

func (*TopicSwitch) Apply

func (r *TopicSwitch) Apply(c context.Context, m flow.Message[structure.Bytes, structure.Bytes], s State[structure.Bytes]) ([]flow.Message[structure.Bytes, structure.Bytes], State[structure.Bytes], error)