dapr_client/saga/saga.go

package saga

import (
	"sync"
	"time"
)

var sagaInstance *Saga
var sagaInstanceMutex sync.Mutex
var initOnce sync.Once

type Saga struct {
	stateStoreName string
	name           string
	orchestrators  []*Orchestrator
}

func Init(stateStoreName string, sagaName string) {
	sagaInstanceMutex.Lock()
	defer sagaInstanceMutex.Unlock()

	if sagaInstance != nil {
		return
	}

	sagaInstance = new(Saga)
	sagaInstance.stateStoreName = stateStoreName
	sagaInstance.name = sagaName
	sagaInstance.orchestrators = make([]*Orchestrator, 0)
}

func Destroy() {
	sagaInstanceMutex.Lock()
	defer sagaInstanceMutex.Unlock()

	if sagaInstance == nil {
		return
	}

	for _, orchestrator := range sagaInstance.orchestrators {
		orchestrator.stop()
	}

	sagaInstance.orchestrators = nil
	sagaInstance.name = ""
	sagaInstance.stateStoreName = ""
	sagaInstance = nil
}

func GetInstance() *Saga {
	return sagaInstance
}

func (s *Saga) BuildOrchestrator(orchestratorName string, rollbackRetryPeriodSec time.Duration) *Orchestrator {
	sagaInstanceMutex.Lock()
	defer sagaInstanceMutex.Unlock()

	orchestrator := &Orchestrator{
		stateStoreName:         s.stateStoreName,
		sagaName:               s.name,
		name:                   orchestratorName,
		rollbackRetryPeriodSec: rollbackRetryPeriodSec,
	}

	sagaInstance.orchestrators = append(sagaInstance.orchestrators, orchestrator)

	return orchestrator
}

func (s *Saga) GetOrchestratorNeedRollback() ([]OrchestratorState, error) {
	return getSagaOrchestratorStates(s.stateStoreName, s.name)
}