sec

Documentation

Index

• Constants
• type Orchestrator
  • func NewOrchestrator[T any](saga Saga[T], repo SagaRepository[T], publisher am.CommandPublisher) Orchestrator[T]
• type Saga
  • func NewSaga[T any](name, replyTopic string) Saga[T]
• type SagaContext
• type SagaRepository
  • func NewSagaRepository[T any](reg registry.Registry, store SagaStore) SagaRepository[T]
  • func (r SagaRepository[T]) Load(ctx context.Context, sagaName, sagaID string) (*SagaContext[T], error)
  • func (r SagaRepository[T]) Save(ctx context.Context, sagaName string, sagaCtx *SagaContext[T]) error
• type SagaStep
• type SagaStore
• type StepActionFunc
• type StepOption
  • func OnActionReply[T any](replyName string, fn StepReplyHandlerFunc[T]) StepOption[T]
  • func OnCompensationReply[T any](replyName string, fn StepReplyHandlerFunc[T]) StepOption[T]
  • func WithAction[T any](fn StepActionFunc[T]) StepOption[T]
  • func WithCompensation[T any](fn StepActionFunc[T]) StepOption[T]
• type StepReplyHandlerFunc

Constants

const (
	SagaCommandIDHdr   = am.CommandHdrPrefix + "SAGA_ID"
	SagaCommandNameHdr = am.CommandHdrPrefix + "SAGA_NAME"

	SagaReplyIDHdr   = am.ReplyHdrPrefix + "SAGA_ID"
	SagaReplyNameHdr = am.ReplyHdrPrefix + "SAGA_NAME"
)

Types

type Orchestrator

type Orchestrator[T any] interface {
	Start(ctx context.Context, id string, data T) error
	ReplyTopic() string
	HandleReply(ctx context.Context, reply ddd.Reply) error
}

func NewOrchestrator

func NewOrchestrator[T any](saga Saga[T], repo SagaRepository[T], publisher am.CommandPublisher) Orchestrator[T]

type Saga

type Saga[T any] interface {
	AddStep() SagaStep[T]
	Name() string
	ReplyTopic() string
	// contains filtered or unexported methods
}

func NewSaga

func NewSaga[T any](name, replyTopic string) Saga[T]

type SagaContext

type SagaContext[T any] struct {
	ID           string
	Data         T
	Step         int
	Done         bool
	Compensating bool
}

type SagaRepository

type SagaRepository[T any] struct {
	// contains filtered or unexported fields
}

func NewSagaRepository

func NewSagaRepository[T any](reg registry.Registry, store SagaStore) SagaRepository[T]

func (SagaRepository[T]) Load

func (r SagaRepository[T]) Load(ctx context.Context, sagaName, sagaID string) (*SagaContext[T], error)

func (SagaRepository[T]) Save

func (r SagaRepository[T]) Save(ctx context.Context, sagaName string, sagaCtx *SagaContext[T]) error

type SagaStep

type SagaStep[T any] interface {
	Action(fn StepActionFunc[T]) SagaStep[T]
	Compensation(fn StepActionFunc[T]) SagaStep[T]
	OnActionReply(replyName string, fn StepReplyHandlerFunc[T]) SagaStep[T]
	OnCompensationReply(replyName string, fn StepReplyHandlerFunc[T]) SagaStep[T]
	// contains filtered or unexported methods
}

type SagaStore

type SagaStore interface {
	Load(ctx context.Context, sagaName, sagaID string) (*SagaContext[[]byte], error)
	Save(ctx context.Context, sagaName string, sagaCtx *SagaContext[[]byte]) error
}

type StepActionFunc

type StepActionFunc[T any] func(ctx context.Context, data T) am.Command

type StepOption

type StepOption[T any] func(step *sagaStep[T])

func OnActionReply

func OnActionReply[T any](replyName string, fn StepReplyHandlerFunc[T]) StepOption[T]

func OnCompensationReply

func OnCompensationReply[T any](replyName string, fn StepReplyHandlerFunc[T]) StepOption[T]

func WithAction

func WithAction[T any](fn StepActionFunc[T]) StepOption[T]

func WithCompensation

func WithCompensation[T any](fn StepActionFunc[T]) StepOption[T]

type StepReplyHandlerFunc

type StepReplyHandlerFunc[T any] func(ctx context.Context, data T, reply ddd.Reply) error