dapr_api/saga/saga.go

package saga

import (
	"git.sxidc.com/service-supports/dapr_api/state"
	"sync"
	"time"
)

var sagaInstance *Saga
var sagaInstanceMutex sync.Mutex

type Saga struct {
	stateAPI       *state.API
	stateStoreName string
	name           string
	orchestrators  []*Orchestrator
}

func Init(stateAPI *state.API, stateStoreName string, sagaName string) {
	sagaInstanceMutex.Lock()
	defer sagaInstanceMutex.Unlock()

	if sagaInstance != nil {
		return
	}

	sagaInstance = new(Saga)
	sagaInstance.stateAPI = stateAPI
	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.stateAPI = nil
	sagaInstance = nil
}

func GetInstance() *Saga {
	return sagaInstance
}

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

	orchestrator := &Orchestrator{
		stateAPI:               s.stateAPI,
		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.stateAPI, s.stateStoreName, s.name)
}