stateful

Documentation

Index

• Constants
• func NewSingleReadWrite(configurations ...runtime.Configuration[*SingleStateReadWrite]) stateless.SingleFunction
• func TopicMissingError(topic string) error
• func WithPersistenceIdSwitchPersistenceIdFunction(topic string, f PersistenceIdFunction[[]byte, []byte]) runtime.Configuration[*PersistenceIdSwitch]
• func WithSingleReadWriteRepository(repository SingleStateRepository) runtime.Configuration[*SingleStateReadWrite]
• func WithSingleReadWriteStatefulFunction(next SingleFunction) runtime.Configuration[*SingleStateReadWrite]
• func WithSingleReadWriteTransactionPersistenceIdFunc(...) runtime.Configuration[*SingleStateReadWrite]
• func WithSingleStatefulDeduplicateNextFunction(next SingleFunction) runtime.Configuration[*SingleStatefulDeduplicate]
• func WithSingleTopicSwitchStatefulSingleFunction(topic string, f SingleFunction) runtime.Configuration[*SingleTopicSwitch]
• type OneToOneFunction
• type OneToTwoFunction
• type PersistenceIdFunction
  • func ConvertPersistenceId[IK any, IV any](source PersistenceIdFunction[IK, IV], ik format.Format[IK], ...) PersistenceIdFunction[[]byte, []byte]
  • func NewSinglePersistenceIdSwitch(configurations ...runtime.Configuration[*PersistenceIdSwitch]) PersistenceIdFunction[[]byte, []byte]
• type PersistenceIdSwitch
  • func (r *PersistenceIdSwitch) Apply(c context.Context, m message.Message[message.Bytes, message.Bytes]) (string, error)
• 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 NewSingleStatefulDeduplicate(configurations ...runtime.Configuration[*SingleStatefulDeduplicate]) SingleFunction
  • func NewSingleTopicSwitch(configurations ...runtime.Configuration[*SingleTopicSwitch]) SingleFunction
• type SingleState
  • func ConvertSingleState[V1 any, V2 any](source SingleState[V1], v1 format.Format[V1], v2 format.Format[V2]) (SingleState[V2], error)
  • func NewSingleState[S any](content S) SingleState[S]
  • func SetDefault[S any](s SingleState[S]) SingleState[S]
• type SingleStateReadWrite
  • func (r *SingleStateReadWrite) Apply(c context.Context, m message.Message[message.Bytes, message.Bytes]) ([]message.Message[message.Bytes, message.Bytes], error)
• type SingleStateRepository
• type SingleStatefulDeduplicate
  • func (r SingleStatefulDeduplicate) Apply(c context.Context, m message.Message[message.Bytes, message.Bytes], ...) ([]message.Message[message.Bytes, message.Bytes], SingleState[message.Bytes], ...)
• type SingleTopicSwitch
  • func (r *SingleTopicSwitch) Apply(c context.Context, m message.Message[message.Bytes, message.Bytes], ...) ([]message.Message[message.Bytes, message.Bytes], SingleState[message.Bytes], ...)

Constants

const (
	ContextTransaction = "ContextTransaction"
)

Functions

func NewSingleReadWrite

func NewSingleReadWrite(configurations ...runtime.Configuration[*SingleStateReadWrite]) stateless.SingleFunction

constructor

func TopicMissingError

func TopicMissingError(topic string) error

func WithPersistenceIdSwitchPersistenceIdFunction

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

configuration

func WithSingleReadWriteRepository

func WithSingleReadWriteRepository(repository SingleStateRepository) runtime.Configuration[*SingleStateReadWrite]

func WithSingleReadWriteStatefulFunction

func WithSingleReadWriteStatefulFunction(next SingleFunction) runtime.Configuration[*SingleStateReadWrite]

func WithSingleReadWriteTransactionPersistenceIdFunc

func WithSingleReadWriteTransactionPersistenceIdFunc(persistenceIdFunc func(context.Context, message.Message[message.Bytes, message.Bytes]) (string, error)) runtime.Configuration[*SingleStateReadWrite]

configuration

func WithSingleStatefulDeduplicateNextFunction

func WithSingleStatefulDeduplicateNextFunction(next SingleFunction) runtime.Configuration[*SingleStatefulDeduplicate]

configuration

func WithSingleTopicSwitchStatefulSingleFunction

func WithSingleTopicSwitchStatefulSingleFunction(topic string, f SingleFunction) runtime.Configuration[*SingleTopicSwitch]

configuration

Types

type OneToOneFunction

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

type OneToTwoFunction

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

type PersistenceIdFunction

type PersistenceIdFunction[IK any, IV any] func(context.Context, message.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 NewSinglePersistenceIdSwitch

func NewSinglePersistenceIdSwitch(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 message.Message[message.Bytes, message.Bytes]) (string, error)

type SingleFunction

type SingleFunction func(context.Context, message.Message[message.Bytes, message.Bytes], SingleState[message.Bytes]) ([]message.Message[message.Bytes, message.Bytes], SingleState[message.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 topic.Topic[IK, IV],
	outputTopic topic.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 topic.Topic[IK, IV],
	outputTopic1 topic.Topic[OK1, OV1],
	outputTopic2 topic.Topic[OK2, OV2],
) SingleFunction

func NewSingleStatefulDeduplicate

func NewSingleStatefulDeduplicate(configurations ...runtime.Configuration[*SingleStatefulDeduplicate]) SingleFunction

constructor

func NewSingleTopicSwitch

func NewSingleTopicSwitch(configurations ...runtime.Configuration[*SingleTopicSwitch]) SingleFunction

constructor

type SingleState

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

func ConvertSingleState

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

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

func NewSingleState

func NewSingleState[S any](content S) SingleState[S]

func SetDefault

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

type SingleStateReadWrite

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

implementation

func (*SingleStateReadWrite) Apply

func (r *SingleStateReadWrite) Apply(c context.Context, m message.Message[message.Bytes, message.Bytes]) ([]message.Message[message.Bytes, message.Bytes], error)

type SingleStateRepository

type SingleStateRepository interface {
	Get(ctx context.Context, persistenceId string) (SingleState[message.Bytes], error)
	Upsert(ctx context.Context, persistenceId string, dbState SingleState[message.Bytes]) error
}

type SingleStatefulDeduplicate

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

implementation

func (SingleStatefulDeduplicate) Apply

func (r SingleStatefulDeduplicate) Apply(c context.Context, m message.Message[message.Bytes, message.Bytes], inState SingleState[message.Bytes]) ([]message.Message[message.Bytes, message.Bytes], SingleState[message.Bytes], error)

type SingleTopicSwitch

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

implementation

func (*SingleTopicSwitch) Apply

func (r *SingleTopicSwitch) Apply(c context.Context, m message.Message[message.Bytes, message.Bytes], s SingleState[message.Bytes]) ([]message.Message[message.Bytes, message.Bytes], SingleState[message.Bytes], error)