saga

package
v0.0.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 7, 2021 License: MIT Imports: 7 Imported by: 2

Documentation

Index

Constants

View Source 
const (
	MessageCommandSagaID   = msg.MessageCommandPrefix + "SAGA_ID"
	MessageCommandSagaName = msg.MessageCommandPrefix + "SAGA_NAME"
	MessageCommandResource = msg.MessageCommandPrefix + "RESOURCE"

	MessageReplySagaID   = msg.MessageReplyPrefix + "SAGA_ID"
	MessageReplySagaName = msg.MessageReplyPrefix + "SAGA_NAME"
)

Saga message headers

Variables

This section is empty.

Functions

func WithSagaInfo 

func WithSagaInfo(instance *Instance) msg.MessageOption

WithSagaInfo is an option to set additional Saga specific headers

Types

type Command 

type Command interface {
	SagaID() string
	SagaName() string
	Command() core.Command
	Headers() msg.Headers
}

Command is a core.Command with message header information

type CommandDispatcher 

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

CommandDispatcher is a MessageReceiver for Commands

func NewCommandDispatcher 

func NewCommandDispatcher(publisher msg.ReplyMessagePublisher, options ...CommandDispatcherOption) *CommandDispatcher

NewCommandDispatcher constructs a new CommandDispatcher

func (*CommandDispatcher) Handle 

func (d *CommandDispatcher) Handle(cmd core.Command, handler CommandHandlerFunc) *CommandDispatcher

Handle adds a new Command that will be handled by handler

func (*CommandDispatcher) ReceiveMessage 

func (d *CommandDispatcher) ReceiveMessage(ctx context.Context, message msg.Message) error

ReceiveMessage implements MessageReceiver.ReceiveMessage

type CommandDispatcherOption 

type CommandDispatcherOption func(consumer *CommandDispatcher)

CommandDispatcherOption options for CommandConsumers

func WithCommandDispatcherLogger 

func WithCommandDispatcherLogger(logger log.Logger) CommandDispatcherOption

WithCommandDispatcherLogger is an option to set the log.Logger of the CommandDispatcher

type CommandHandlerFunc 

type CommandHandlerFunc func(context.Context, Command) ([]msg.Reply, error)

type Definition 

type Definition interface {
	SagaName() string
	ReplyChannel() string
	Steps() []Step
	OnHook(hook LifecycleHook, instance *Instance)
}

Definition interface

type Instance 

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

func NewSagaInstance 

func NewSagaInstance(sagaName, sagaID string, sagaData core.SagaData, currentStep int, endState, compensating bool) *Instance

NewSagaInstance constructor for *SagaInstances

func (*Instance) Compensating 

func (i *Instance) Compensating() bool

func (*Instance) CurrentStep 

func (i *Instance) CurrentStep() int

func (*Instance) EndState 

func (i *Instance) EndState() bool

func (*Instance) SagaData 

func (i *Instance) SagaData() core.SagaData

func (*Instance) SagaID 

func (i *Instance) SagaID() string

func (*Instance) SagaName 

func (i *Instance) SagaName() string

type InstanceStore 

type InstanceStore interface {
	Find(ctx context.Context, sagaName, sagaID string) (*Instance, error)
	Save(ctx context.Context, sagaInstance *Instance) error
	Update(ctx context.Context, sagaInstance *Instance) error
}

type LifecycleHook 

type LifecycleHook int
const (
	SagaStarting LifecycleHook = iota
	SagaCompleted
	SagaCompensated
)

Definition lifecycle hooks

type LocalStep 

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

LocalStep is used to execute local saga business logic

func NewLocalStep 

func NewLocalStep(action func(context.Context, core.SagaData) error) LocalStep

NewLocalStep constructor for LocalStep

func (LocalStep) Compensation 

func (s LocalStep) Compensation(compensation func(context.Context, core.SagaData) error) LocalStep

type Orchestrator 

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

Orchestrator orchestrates local and distributed processes

func NewOrchestrator 

func NewOrchestrator(definition Definition, store InstanceStore, publisher msg.CommandMessagePublisher, options ...OrchestratorOption) *Orchestrator

NewOrchestrator constructs a new Orchestrator

func (*Orchestrator) ReceiveMessage 

func (o *Orchestrator) ReceiveMessage(ctx context.Context, message msg.Message) error

ReceiveMessage implements msg.MessageReceiver.ReceiveMessage

func (*Orchestrator) ReplyChannel 

func (o *Orchestrator) ReplyChannel() string

func (*Orchestrator) Start 

func (o *Orchestrator) Start(ctx context.Context, sagaData core.SagaData) (*Instance, error)

type OrchestratorOption 

type OrchestratorOption func(o *Orchestrator)

func WithOrchestratorLogger 

func WithOrchestratorLogger(logger log.Logger) OrchestratorOption

type RemoteStep 

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

RemoteStep is used to execute distributed saga business logic

func NewRemoteStep 

func NewRemoteStep() RemoteStep

NewRemoteStep constructor for RemoteStep

func (RemoteStep) Action 

func (s RemoteStep) Action(fn func(context.Context, core.SagaData) msg.DomainCommand, options ...RemoteStepActionOption) RemoteStep

Action adds a domain command constructor that will be called while the definition is advancing

func (RemoteStep) Compensation 

func (s RemoteStep) Compensation(fn func(context.Context, core.SagaData) msg.DomainCommand, options ...RemoteStepActionOption) RemoteStep

Compensation adds a domain command constructor that will be called while the definition is compensating

func (RemoteStep) HandleActionReply 

func (s RemoteStep) HandleActionReply(reply core.Reply, handler func(context.Context, core.SagaData, core.Reply) error) RemoteStep

HandleActionReply adds additional handling for specific replies while advancing

SuccessReply and FailureReply do not require any special handling unless desired

func (RemoteStep) HandleCompensationReply 

func (s RemoteStep) HandleCompensationReply(reply core.Reply, handler func(context.Context, core.SagaData, core.Reply) error) RemoteStep

HandleCompensationReply adds additional handling for specific replies while compensating

SuccessReply does not require any special handling unless desired

type RemoteStepActionOption 

type RemoteStepActionOption func(action *remoteStepAction)

func WithRemoteStepPredicate 

func WithRemoteStepPredicate(predicate func(context.Context, core.SagaData) bool) RemoteStepActionOption

type Step 

type Step interface {
	// contains filtered or unexported methods
}

Step interface for local, remote, ...other saga steps

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.