This is an automated email from the ASF dual-hosted git repository.
xjlgod pushed a commit to branch feature/saga
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/feature/saga by this push:
new 269b0816 refactor-engine/core (#838)
269b0816 is described below
commit 269b081669cf3e2af5718aaddc79f74915536b12
Author: flypiggy <[email protected]>
AuthorDate: Sun Jun 15 15:55:40 2025 +0800
refactor-engine/core (#838)
* refactor-engine/core
* Decouple the core package
---
.../default_statemachine_config.go | 48 ++++-----
.../core/process_ctrl_statemachine_engine.go | 65 +++++++------
.../engine/{core => pcext}/compensation_holder.go | 5 +-
.../engine/{core => pcext}/engine_utils.go | 16 +--
.../engine/{core => pcext}/instruction.go | 11 +--
.../engine/{core => pcext}/loop_context_holder.go | 7 +-
.../engine/{core => pcext}/loop_task_utils.go | 8 +-
.../engine/{core => pcext}/parameter_utils.go | 8 +-
.../process_state.go => pcext/process_handler.go} | 15 ++-
.../engine/{core => pcext}/process_router.go | 108 +++------------------
.../state_router.go => pcext/state_router_impl.go} | 15 +--
.../repo}/repository/state_log_repository.go | 25 +++--
.../repo}/repository/state_machine_repository.go | 6 +-
.../repository/state_machine_repository_test.go | 0
.../engine/{core => repo}/statemachine_store.go | 35 +------
.../engine/{core => }/statemachine_config.go | 17 ++--
.../engine/{core => }/statemachine_engine.go | 7 +-
.../engine/{core => }/statemachine_engine_test.go | 5 +-
pkg/saga/statemachine/engine/strategy.go | 19 ++++
.../engine/{core => strategy}/status_decision.go | 26 ++---
.../utils.go => utils/process_context_utils.go} | 18 ++--
.../core => process_ctrl}/bussiness_processor.go | 2 +-
.../process_ctrl/default_process_handler.go | 7 ++
.../{engine/core => process_ctrl}/event.go | 2 +-
.../{engine/core => process_ctrl}/event_bus.go | 2 +-
.../core => process_ctrl}/event_consumer.go | 2 +-
.../core => process_ctrl}/event_publisher.go | 2 +-
.../handlers/service_task_state_handler.go | 30 +++---
.../core/event.go => process_ctrl/instruction.go} | 4 +-
.../core => process_ctrl}/process_context.go | 2 +-
.../core => process_ctrl}/process_controller.go | 2 +-
.../core => process_ctrl}/process_router.go | 106 +-------------------
pkg/saga/statemachine/store/db/statelog.go | 45 +++++----
pkg/saga/statemachine/store/db/statelog_test.go | 16 +--
pkg/saga/statemachine/store/store.go | 39 ++++++++
35 files changed, 295 insertions(+), 430 deletions(-)
diff --git a/pkg/saga/statemachine/engine/core/default_statemachine_config.go
b/pkg/saga/statemachine/engine/config/default_statemachine_config.go
similarity index 83%
rename from pkg/saga/statemachine/engine/core/default_statemachine_config.go
rename to pkg/saga/statemachine/engine/config/default_statemachine_config.go
index cd118f55..156448c7 100644
--- a/pkg/saga/statemachine/engine/core/default_statemachine_config.go
+++ b/pkg/saga/statemachine/engine/config/default_statemachine_config.go
@@ -15,12 +15,16 @@
* limitations under the License.
*/
-package core
+package config
import (
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/expr"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/invoker"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/repo"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/sequence"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
+ "github.com/seata/seata-go/pkg/saga/statemachine/store"
"sync"
)
@@ -47,14 +51,14 @@ type DefaultStateMachineConfig struct {
// Components
// Event publisher
- syncProcessCtrlEventPublisher EventPublisher
- asyncProcessCtrlEventPublisher EventPublisher
+ syncProcessCtrlEventPublisher process_ctrl.EventPublisher
+ asyncProcessCtrlEventPublisher process_ctrl.EventPublisher
// Store related components
- stateLogRepository StateLogRepository
- stateLogStore StateLogStore
- stateLangStore StateLangStore
- stateMachineRepository StateMachineRepository
+ stateLogRepository repo.StateLogRepository
+ stateLogStore store.StateLogStore
+ stateLangStore store.StateLangStore
+ stateMachineRepository repo.StateMachineRepository
// Expression related components
expressionFactoryManager expr.ExpressionFactoryManager
@@ -65,7 +69,7 @@ type DefaultStateMachineConfig struct {
scriptInvokerManager invoker.ScriptInvokerManager
// Other components
- statusDecisionStrategy StatusDecisionStrategy
+ statusDecisionStrategy engine.StatusDecisionStrategy
seqGenerator sequence.SeqGenerator
componentLock *sync.Mutex
}
@@ -94,27 +98,27 @@ func (c *DefaultStateMachineConfig)
SetDefaultTenantId(defaultTenantId string) {
c.defaultTenantId = defaultTenantId
}
-func (c *DefaultStateMachineConfig)
SetSyncProcessCtrlEventPublisher(syncProcessCtrlEventPublisher EventPublisher) {
+func (c *DefaultStateMachineConfig)
SetSyncProcessCtrlEventPublisher(syncProcessCtrlEventPublisher
process_ctrl.EventPublisher) {
c.syncProcessCtrlEventPublisher = syncProcessCtrlEventPublisher
}
-func (c *DefaultStateMachineConfig)
SetAsyncProcessCtrlEventPublisher(asyncProcessCtrlEventPublisher
EventPublisher) {
+func (c *DefaultStateMachineConfig)
SetAsyncProcessCtrlEventPublisher(asyncProcessCtrlEventPublisher
process_ctrl.EventPublisher) {
c.asyncProcessCtrlEventPublisher = asyncProcessCtrlEventPublisher
}
-func (c *DefaultStateMachineConfig) SetStateLogRepository(stateLogRepository
StateLogRepository) {
+func (c *DefaultStateMachineConfig) SetStateLogRepository(stateLogRepository
repo.StateLogRepository) {
c.stateLogRepository = stateLogRepository
}
-func (c *DefaultStateMachineConfig) SetStateLogStore(stateLogStore
StateLogStore) {
+func (c *DefaultStateMachineConfig) SetStateLogStore(stateLogStore
store.StateLogStore) {
c.stateLogStore = stateLogStore
}
-func (c *DefaultStateMachineConfig) SetStateLangStore(stateLangStore
StateLangStore) {
+func (c *DefaultStateMachineConfig) SetStateLangStore(stateLangStore
store.StateLangStore) {
c.stateLangStore = stateLangStore
}
-func (c *DefaultStateMachineConfig)
SetStateMachineRepository(stateMachineRepository StateMachineRepository) {
+func (c *DefaultStateMachineConfig)
SetStateMachineRepository(stateMachineRepository repo.StateMachineRepository) {
c.stateMachineRepository = stateMachineRepository
}
@@ -134,7 +138,7 @@ func (c *DefaultStateMachineConfig)
SetScriptInvokerManager(scriptInvokerManager
c.scriptInvokerManager = scriptInvokerManager
}
-func (c *DefaultStateMachineConfig)
SetStatusDecisionStrategy(statusDecisionStrategy StatusDecisionStrategy) {
+func (c *DefaultStateMachineConfig)
SetStatusDecisionStrategy(statusDecisionStrategy engine.StatusDecisionStrategy)
{
c.statusDecisionStrategy = statusDecisionStrategy
}
@@ -142,19 +146,19 @@ func (c *DefaultStateMachineConfig)
SetSeqGenerator(seqGenerator sequence.SeqGen
c.seqGenerator = seqGenerator
}
-func (c *DefaultStateMachineConfig) StateLogRepository() StateLogRepository {
+func (c *DefaultStateMachineConfig) StateLogRepository()
repo.StateLogRepository {
return c.stateLogRepository
}
-func (c *DefaultStateMachineConfig) StateMachineRepository()
StateMachineRepository {
+func (c *DefaultStateMachineConfig) StateMachineRepository()
repo.StateMachineRepository {
return c.stateMachineRepository
}
-func (c *DefaultStateMachineConfig) StateLogStore() StateLogStore {
+func (c *DefaultStateMachineConfig) StateLogStore() store.StateLogStore {
return c.stateLogStore
}
-func (c *DefaultStateMachineConfig) StateLangStore() StateLangStore {
+func (c *DefaultStateMachineConfig) StateLangStore() store.StateLangStore {
return c.stateLangStore
}
@@ -170,15 +174,15 @@ func (c *DefaultStateMachineConfig) SeqGenerator()
sequence.SeqGenerator {
return c.seqGenerator
}
-func (c *DefaultStateMachineConfig) StatusDecisionStrategy()
StatusDecisionStrategy {
+func (c *DefaultStateMachineConfig) StatusDecisionStrategy()
engine.StatusDecisionStrategy {
return c.statusDecisionStrategy
}
-func (c *DefaultStateMachineConfig) EventPublisher() EventPublisher {
+func (c *DefaultStateMachineConfig) EventPublisher()
process_ctrl.EventPublisher {
return c.syncProcessCtrlEventPublisher
}
-func (c *DefaultStateMachineConfig) AsyncEventPublisher() EventPublisher {
+func (c *DefaultStateMachineConfig) AsyncEventPublisher()
process_ctrl.EventPublisher {
return c.asyncProcessCtrlEventPublisher
}
diff --git
a/pkg/saga/statemachine/engine/core/process_ctrl_statemachine_engine.go
b/pkg/saga/statemachine/engine/core/process_ctrl_statemachine_engine.go
index 93f2dd48..a3e64b2a 100644
--- a/pkg/saga/statemachine/engine/core/process_ctrl_statemachine_engine.go
+++ b/pkg/saga/statemachine/engine/core/process_ctrl_statemachine_engine.go
@@ -22,7 +22,12 @@ import (
"fmt"
"github.com/pkg/errors"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/config"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/exception"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/pcext"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/utils"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl/process"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang/state"
@@ -32,12 +37,12 @@ import (
)
type ProcessCtrlStateMachineEngine struct {
- StateMachineConfig StateMachineConfig
+ StateMachineConfig engine.StateMachineConfig
}
func NewProcessCtrlStateMachineEngine() *ProcessCtrlStateMachineEngine {
return &ProcessCtrlStateMachineEngine{
- StateMachineConfig: NewDefaultStateMachineConfig(),
+ StateMachineConfig: config.NewDefaultStateMachineConfig(),
}
}
@@ -47,7 +52,7 @@ func (p ProcessCtrlStateMachineEngine) Start(ctx
context.Context, stateMachineNa
}
func (p ProcessCtrlStateMachineEngine) StartAsync(ctx context.Context,
stateMachineName string, tenantId string,
- startParams map[string]interface{}, callback CallBack)
(statelang.StateMachineInstance, error) {
+ startParams map[string]interface{}, callback engine.CallBack)
(statelang.StateMachineInstance, error) {
return p.startInternal(ctx, stateMachineName, tenantId, "",
startParams, true, callback)
}
@@ -57,7 +62,7 @@ func (p ProcessCtrlStateMachineEngine)
StartWithBusinessKey(ctx context.Context,
}
func (p ProcessCtrlStateMachineEngine) StartWithBusinessKeyAsync(ctx
context.Context, stateMachineName string,
- tenantId string, businessKey string, startParams
map[string]interface{}, callback CallBack) (statelang.StateMachineInstance,
error) {
+ tenantId string, businessKey string, startParams
map[string]interface{}, callback engine.CallBack)
(statelang.StateMachineInstance, error) {
return p.startInternal(ctx, stateMachineName, tenantId, businessKey,
startParams, true, callback)
}
@@ -66,7 +71,7 @@ func (p ProcessCtrlStateMachineEngine) Forward(ctx
context.Context, stateMachine
return p.forwardInternal(ctx, stateMachineInstId, replaceParams, false,
false, nil)
}
-func (p ProcessCtrlStateMachineEngine) ForwardAsync(ctx context.Context,
stateMachineInstId string, replaceParams map[string]interface{}, callback
CallBack) (statelang.StateMachineInstance, error) {
+func (p ProcessCtrlStateMachineEngine) ForwardAsync(ctx context.Context,
stateMachineInstId string, replaceParams map[string]interface{}, callback
engine.CallBack) (statelang.StateMachineInstance, error) {
return p.forwardInternal(ctx, stateMachineInstId, replaceParams, false,
true, callback)
}
@@ -75,7 +80,7 @@ func (p ProcessCtrlStateMachineEngine) Compensate(ctx
context.Context, stateMach
return p.compensateInternal(ctx, stateMachineInstId, replaceParams,
false, nil)
}
-func (p ProcessCtrlStateMachineEngine) CompensateAsync(ctx context.Context,
stateMachineInstId string, replaceParams map[string]interface{}, callback
CallBack) (statelang.StateMachineInstance, error) {
+func (p ProcessCtrlStateMachineEngine) CompensateAsync(ctx context.Context,
stateMachineInstId string, replaceParams map[string]interface{}, callback
engine.CallBack) (statelang.StateMachineInstance, error) {
return p.compensateInternal(ctx, stateMachineInstId, replaceParams,
true, callback)
}
@@ -83,11 +88,11 @@ func (p ProcessCtrlStateMachineEngine) SkipAndForward(ctx
context.Context, state
return p.forwardInternal(ctx, stateMachineInstId, replaceParams, true,
false, nil)
}
-func (p ProcessCtrlStateMachineEngine) SkipAndForwardAsync(ctx
context.Context, stateMachineInstId string, callback CallBack)
(statelang.StateMachineInstance, error) {
+func (p ProcessCtrlStateMachineEngine) SkipAndForwardAsync(ctx
context.Context, stateMachineInstId string, callback engine.CallBack)
(statelang.StateMachineInstance, error) {
return p.forwardInternal(ctx, stateMachineInstId, nil, true, true,
callback)
}
-func (p ProcessCtrlStateMachineEngine) GetStateMachineConfig()
StateMachineConfig {
+func (p ProcessCtrlStateMachineEngine) GetStateMachineConfig()
engine.StateMachineConfig {
return p.StateMachineConfig
}
@@ -135,7 +140,7 @@ func (p ProcessCtrlStateMachineEngine)
ReloadStateMachineInstance(ctx context.Co
}
func (p ProcessCtrlStateMachineEngine) startInternal(ctx context.Context,
stateMachineName string, tenantId string,
- businessKey string, startParams map[string]interface{}, async bool,
callback CallBack) (statelang.StateMachineInstance, error) {
+ businessKey string, startParams map[string]interface{}, async bool,
callback engine.CallBack) (statelang.StateMachineInstance, error) {
if tenantId == "" {
tenantId = p.StateMachineConfig.DefaultTenantId()
}
@@ -146,11 +151,11 @@ func (p ProcessCtrlStateMachineEngine) startInternal(ctx
context.Context, stateM
}
// Build the process_ctrl context.
- processContextBuilder := NewProcessContextBuilder().
+ processContextBuilder := utils.NewProcessContextBuilder().
WithProcessType(process.StateLang).
WithOperationName(constant.OperationNameStart).
WithAsyncCallback(callback).
- WithInstruction(NewStateInstruction(stateMachineName,
tenantId)).
+ WithInstruction(pcext.NewStateInstruction(stateMachineName,
tenantId)).
WithStateMachineInstance(stateMachineInstance).
WithStateMachineConfig(p.StateMachineConfig).
WithStateMachineEngine(p).
@@ -173,7 +178,7 @@ func (p ProcessCtrlStateMachineEngine) startInternal(ctx
context.Context, stateM
stateMachineInstance.SetID(p.StateMachineConfig.SeqGenerator().GenerateId(constant.SeqEntityStateMachineInst,
""))
}
- var eventPublisher EventPublisher
+ var eventPublisher process_ctrl.EventPublisher
if async {
eventPublisher = p.StateMachineConfig.AsyncEventPublisher()
} else {
@@ -189,7 +194,7 @@ func (p ProcessCtrlStateMachineEngine) startInternal(ctx
context.Context, stateM
}
func (p ProcessCtrlStateMachineEngine) forwardInternal(ctx context.Context,
stateMachineInstId string,
- replaceParams map[string]interface{}, skip bool, async bool, callback
CallBack) (statelang.StateMachineInstance, error) {
+ replaceParams map[string]interface{}, skip bool, async bool, callback
engine.CallBack) (statelang.StateMachineInstance, error) {
stateMachineInstance, err := p.reloadStateMachineInstance(ctx,
stateMachineInstId)
if err != nil {
return nil, err
@@ -223,7 +228,7 @@ func (p ProcessCtrlStateMachineEngine) forwardInternal(ctx
context.Context, stat
fmt.Sprintf("StateMachineInstance[id:%s] Cannot find
last forward execution stateInstance", stateMachineInstId), nil)
}
- contextBuilder := NewProcessContextBuilder().
+ contextBuilder := utils.NewProcessContextBuilder().
WithProcessType(process.StateLang).
WithOperationName(constant.OperationNameForward).
WithAsyncCallback(callback).
@@ -252,9 +257,9 @@ func (p ProcessCtrlStateMachineEngine) forwardInternal(ctx
context.Context, stat
context.SetVariable(constant.VarNameStateMachineContext,
concurrentContextVariables)
stateMachineInstance.SetContext(concurrentContextVariables)
- originStateName := GetOriginStateName(lastForwardState)
+ originStateName := pcext.GetOriginStateName(lastForwardState)
lastState := stateMachineInstance.StateMachine().State(originStateName)
- loop := GetLoopConfig(ctx, context, lastState)
+ loop := pcext.GetLoopConfig(ctx, context, lastState)
if loop != nil && lastForwardState.Status() == statelang.SU {
lastForwardState = p.findOutLastNeedForwardStateInstance(ctx,
context)
}
@@ -268,10 +273,10 @@ func (p ProcessCtrlStateMachineEngine)
forwardInternal(ctx context.Context, stat
lastForwardState.SetIgnoreStatus(true)
}
- inst := NewStateInstruction(stateMachineInstance.StateMachine().Name(),
stateMachineInstance.TenantID())
+ inst :=
pcext.NewStateInstruction(stateMachineInstance.StateMachine().Name(),
stateMachineInstance.TenantID())
if skip || lastForwardState.Status() == statelang.SU {
next := ""
- curState :=
stateMachineInstance.StateMachine().State(GetOriginStateName(lastForwardState))
+ curState :=
stateMachineInstance.StateMachine().State(pcext.GetOriginStateName(lastForwardState))
if taskState, ok := curState.(*state.AbstractTaskState); ok {
next = taskState.Next()
}
@@ -281,11 +286,11 @@ func (p ProcessCtrlStateMachineEngine)
forwardInternal(ctx context.Context, stat
}
inst.SetStateName(next)
} else {
- if lastForwardState.Status() == statelang.RU &&
!IsTimeout(lastForwardState.StartedTime(),
p.StateMachineConfig.ServiceInvokeTimeout()) {
+ if lastForwardState.Status() == statelang.RU &&
!pcext.IsTimeout(lastForwardState.StartedTime(),
p.StateMachineConfig.ServiceInvokeTimeout()) {
return nil,
exception.NewEngineExecutionException(seataErrors.OperationDenied,
fmt.Sprintf("State [%s] is running,
operation[forward] denied", lastForwardState.Name()), nil)
}
- inst.SetStateName(GetOriginStateName(lastForwardState))
+ inst.SetStateName(pcext.GetOriginStateName(lastForwardState))
}
context.SetInstruction(inst)
@@ -304,7 +309,7 @@ func (p ProcessCtrlStateMachineEngine) forwardInternal(ctx
context.Context, stat
if err != nil {
return nil, err
}
- loop = GetLoopConfig(ctx, context, curState)
+ loop = pcext.GetLoopConfig(ctx, context, curState)
if loop != nil {
inst.SetTemporaryState(state.NewLoopStartStateImpl())
}
@@ -341,7 +346,7 @@ func (p ProcessCtrlStateMachineEngine)
findOutLastForwardStateInstance(stateInst
}
}
- subInst, _ :=
p.StateMachineConfig.StateLogStore().GetStateMachineInstanceByParentId(GenerateParentId(finalState))
+ subInst, _ :=
p.StateMachineConfig.StateLogStore().GetStateMachineInstanceByParentId(pcext.GenerateParentId(finalState))
if len(subInst) > 0 {
if subInst[0].CompensationStatus() ==
statelang.SU {
continue
@@ -415,7 +420,7 @@ func (p ProcessCtrlStateMachineEngine)
createMachineInstance(stateMachineName st
}
func (p ProcessCtrlStateMachineEngine) compensateInternal(ctx context.Context,
stateMachineInstId string, replaceParams map[string]any,
- async bool, callback CallBack) (statelang.StateMachineInstance, error) {
+ async bool, callback engine.CallBack) (statelang.StateMachineInstance,
error) {
stateMachineInstance, err := p.reloadStateMachineInstance(ctx,
stateMachineInstId)
if err != nil {
return nil, err
@@ -443,7 +448,7 @@ func (p ProcessCtrlStateMachineEngine)
compensateInternal(ctx context.Context, s
}
}
- contextBuilder :=
NewProcessContextBuilder().WithProcessType(process.StateLang).
+ contextBuilder :=
utils.NewProcessContextBuilder().WithProcessType(process.StateLang).
WithOperationName(constant.OperationNameCompensate).WithAsyncCallback(callback).
WithStateMachineInstance(stateMachineInstance).
WithStateMachineConfig(p.StateMachineConfig).WithStateMachineEngine(p).WithIsAsyncExecution(async)
@@ -481,7 +486,7 @@ func (p ProcessCtrlStateMachineEngine)
compensateInternal(ctx context.Context, s
}
}
- inst := NewStateInstruction(stateMachineInstance.TenantID(),
stateMachineInstance.StateMachine().Name())
+ inst := pcext.NewStateInstruction(stateMachineInstance.TenantID(),
stateMachineInstance.StateMachine().Name())
inst.SetTemporaryState(tempCompensationTriggerState)
context.SetInstruction(inst)
@@ -559,14 +564,14 @@ func (p ProcessCtrlStateMachineEngine)
replayContextVariables(ctx context.Contex
for _, stateInstance := range stateInstanceList {
serviceOutputParams := stateInstance.OutputParams()
if serviceOutputParams != nil {
- serviceTaskStateImpl, ok :=
stateMachineInstance.StateMachine().State(GetOriginStateName(stateInstance)).(*state.ServiceTaskStateImpl)
+ serviceTaskStateImpl, ok :=
stateMachineInstance.StateMachine().State(pcext.GetOriginStateName(stateInstance)).(*state.ServiceTaskStateImpl)
if !ok {
return nil,
exception.NewEngineExecutionException(seataErrors.ObjectNotExists,
"Cannot find State by state name
["+stateInstance.Name()+"], may be this is a bug", nil)
}
if serviceTaskStateImpl.Output() != nil &&
len(serviceTaskStateImpl.Output()) != 0 {
- outputVariablesToContext, err :=
CreateOutputParams(p.StateMachineConfig,
+ outputVariablesToContext, err :=
pcext.CreateOutputParams(p.StateMachineConfig,
p.StateMachineConfig.ExpressionResolver(),
serviceTaskStateImpl.AbstractTaskState, serviceOutputParams)
if err != nil {
return nil,
exception.NewEngineExecutionException(seataErrors.ObjectNotExists,
@@ -605,7 +610,7 @@ func (p ProcessCtrlStateMachineEngine) checkStatus(ctx
context.Context, stateMac
}
if stateMachineInstance.IsRunning() &&
- !IsTimeout(stateMachineInstance.UpdatedTime(),
p.StateMachineConfig.TransOperationTimeout()) {
+ !pcext.IsTimeout(stateMachineInstance.UpdatedTime(),
p.StateMachineConfig.TransOperationTimeout()) {
return false,
exception.NewEngineExecutionException(seataErrors.OperationDenied,
"StateMachineInstance
[id:"+stateMachineInstance.ID()+"] is running, operation["+operation+
"] denied", nil)
@@ -702,14 +707,14 @@ func (p ProcessCtrlStateMachineEngine)
nullSafeCopy(srcMap map[string]any, destM
}
}
-func (p ProcessCtrlStateMachineEngine) findOutLastNeedForwardStateInstance(ctx
context.Context, processContext ProcessContext) statelang.StateInstance {
+func (p ProcessCtrlStateMachineEngine) findOutLastNeedForwardStateInstance(ctx
context.Context, processContext process_ctrl.ProcessContext)
statelang.StateInstance {
stateMachineInstance :=
processContext.GetVariable(constant.VarNameStateMachineInst).(statelang.StateMachineInstance)
lastForwardState :=
processContext.GetVariable(constant.VarNameStateInst).(statelang.StateInstance)
actList := stateMachineInstance.StateList()
for i := len(actList) - 1; i >= 0; i-- {
stateInstance := actList[i]
- if GetOriginStateName(stateInstance) ==
GetOriginStateName(lastForwardState) && stateInstance.Status() != statelang.SU {
+ if pcext.GetOriginStateName(stateInstance) ==
pcext.GetOriginStateName(lastForwardState) && stateInstance.Status() !=
statelang.SU {
return stateInstance
}
}
diff --git a/pkg/saga/statemachine/engine/core/compensation_holder.go
b/pkg/saga/statemachine/engine/pcext/compensation_holder.go
similarity index 94%
rename from pkg/saga/statemachine/engine/core/compensation_holder.go
rename to pkg/saga/statemachine/engine/pcext/compensation_holder.go
index c22fb1fe..d533460f 100644
--- a/pkg/saga/statemachine/engine/core/compensation_holder.go
+++ b/pkg/saga/statemachine/engine/pcext/compensation_holder.go
@@ -15,11 +15,12 @@
* limitations under the License.
*/
-package core
+package pcext
import (
"context"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"github.com/seata/seata-go/pkg/util/collection"
"sync"
@@ -67,7 +68,7 @@ func NewCompensationHolder() *CompensationHolder {
}
}
-func GetCurrentCompensationHolder(ctx context.Context, processContext
ProcessContext, forceCreate bool) *CompensationHolder {
+func GetCurrentCompensationHolder(ctx context.Context, processContext
process_ctrl.ProcessContext, forceCreate bool) *CompensationHolder {
compensationholder :=
processContext.GetVariable(constant.VarNameCurrentCompensationHolder).(*CompensationHolder)
lock :=
processContext.GetVariable(constant.VarNameProcessContextMutexLock).(*sync.Mutex)
lock.Lock()
diff --git a/pkg/saga/statemachine/engine/core/engine_utils.go
b/pkg/saga/statemachine/engine/pcext/engine_utils.go
similarity index 90%
rename from pkg/saga/statemachine/engine/core/engine_utils.go
rename to pkg/saga/statemachine/engine/pcext/engine_utils.go
index 8fa22d37..fc7f3566 100644
--- a/pkg/saga/statemachine/engine/core/engine_utils.go
+++ b/pkg/saga/statemachine/engine/pcext/engine_utils.go
@@ -15,12 +15,14 @@
* limitations under the License.
*/
-package core
+package pcext
import (
"context"
"errors"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang/state"
"github.com/seata/seata-go/pkg/util/log"
@@ -31,7 +33,7 @@ import (
"time"
)
-func EndStateMachine(ctx context.Context, processContext ProcessContext) error
{
+func EndStateMachine(ctx context.Context, processContext
process_ctrl.ProcessContext) error {
if processContext.HasVariable(constant.VarNameIsLoopState) {
if processContext.HasVariable(constant.LoopSemaphore) {
weighted, ok :=
processContext.GetVariable(constant.LoopSemaphore).(semaphore.Weighted)
@@ -59,7 +61,7 @@ func EndStateMachine(ctx context.Context, processContext
ProcessContext) error {
log.Debugf("Exception Occurred: %s", exp)
}
- stateMachineConfig, ok :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(StateMachineConfig)
+ stateMachineConfig, ok :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(engine.StateMachineConfig)
if err :=
stateMachineConfig.StatusDecisionStrategy().DecideOnEndState(ctx,
processContext, stateMachineInstance, exp); err != nil {
return err
@@ -91,7 +93,7 @@ func EndStateMachine(ctx context.Context, processContext
ProcessContext) error {
}
}
- callBack, ok :=
processContext.GetVariable(constant.VarNameAsyncCallback).(CallBack)
+ callBack, ok :=
processContext.GetVariable(constant.VarNameAsyncCallback).(engine.CallBack)
if ok {
if exp != nil {
callBack.OnError(ctx, processContext,
stateMachineInstance, exp)
@@ -103,7 +105,7 @@ func EndStateMachine(ctx context.Context, processContext
ProcessContext) error {
return nil
}
-func HandleException(processContext ProcessContext, abstractTaskState
*state.AbstractTaskState, err error) {
+func HandleException(processContext process_ctrl.ProcessContext,
abstractTaskState *state.AbstractTaskState, err error) {
catches := abstractTaskState.Catches()
if catches != nil && len(catches) != 0 {
for _, exceptionMatch := range catches {
@@ -127,7 +129,7 @@ func HandleException(processContext ProcessContext,
abstractTaskState *state.Abs
if reflect.TypeOf(err) == exceptionTypes[i] {
// HACK: we can not get error type in
config file during runtime, so we use exception str
if strings.Contains(err.Error(),
exceptions[i]) {
- hierarchicalProcessContext :=
processContext.(HierarchicalProcessContext)
+ hierarchicalProcessContext :=
processContext.(process_ctrl.HierarchicalProcessContext)
hierarchicalProcessContext.SetVariable(constant.VarNameCurrentExceptionRoute,
exceptionMatch.Next())
return
}
@@ -137,7 +139,7 @@ func HandleException(processContext ProcessContext,
abstractTaskState *state.Abs
}
log.Error("Task execution failed and no catches configured")
- hierarchicalProcessContext :=
processContext.(HierarchicalProcessContext)
+ hierarchicalProcessContext :=
processContext.(process_ctrl.HierarchicalProcessContext)
hierarchicalProcessContext.SetVariable(constant.VarNameIsExceptionNotCatch,
true)
}
diff --git a/pkg/saga/statemachine/engine/core/instruction.go
b/pkg/saga/statemachine/engine/pcext/instruction.go
similarity index 91%
rename from pkg/saga/statemachine/engine/core/instruction.go
rename to pkg/saga/statemachine/engine/pcext/instruction.go
index 4b8f9d3c..6e39f925 100644
--- a/pkg/saga/statemachine/engine/core/instruction.go
+++ b/pkg/saga/statemachine/engine/pcext/instruction.go
@@ -15,18 +15,17 @@
* limitations under the License.
*/
-package core
+package pcext
import (
"errors"
"fmt"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
)
-type Instruction interface {
-}
-
type StateInstruction struct {
stateName string
stateMachineName string
@@ -79,7 +78,7 @@ func (s *StateInstruction) SetTemporaryState(temporaryState
statelang.State) {
s.temporaryState = temporaryState
}
-func (s *StateInstruction) GetState(context ProcessContext) (statelang.State,
error) {
+func (s *StateInstruction) GetState(context process_ctrl.ProcessContext)
(statelang.State, error) {
if s.temporaryState != nil {
return s.temporaryState, nil
}
@@ -88,7 +87,7 @@ func (s *StateInstruction) GetState(context ProcessContext)
(statelang.State, er
return nil, errors.New("stateMachineName is required")
}
- stateMachineConfig, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(StateMachineConfig)
+ stateMachineConfig, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(engine.StateMachineConfig)
if !ok {
return nil, errors.New("stateMachineConfig is required in
context")
}
diff --git a/pkg/saga/statemachine/engine/core/loop_context_holder.go
b/pkg/saga/statemachine/engine/pcext/loop_context_holder.go
similarity index 93%
rename from pkg/saga/statemachine/engine/core/loop_context_holder.go
rename to pkg/saga/statemachine/engine/pcext/loop_context_holder.go
index 242e455b..d9708008 100644
--- a/pkg/saga/statemachine/engine/core/loop_context_holder.go
+++ b/pkg/saga/statemachine/engine/pcext/loop_context_holder.go
@@ -15,11 +15,12 @@
* limitations under the License.
*/
-package core
+package pcext
import (
"context"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"sync"
)
@@ -47,7 +48,7 @@ func NewLoopContextHolder() *LoopContextHolder {
}
}
-func GetCurrentLoopContextHolder(ctx context.Context, processContext
ProcessContext, forceCreate bool) *LoopContextHolder {
+func GetCurrentLoopContextHolder(ctx context.Context, processContext
process_ctrl.ProcessContext, forceCreate bool) *LoopContextHolder {
mutex :=
processContext.GetVariable(constant.VarNameProcessContextMutexLock).(*sync.Mutex)
mutex.Lock()
defer mutex.Unlock()
@@ -60,7 +61,7 @@ func GetCurrentLoopContextHolder(ctx context.Context,
processContext ProcessCont
return loopContextHolder
}
-func ClearCurrent(ctx context.Context, processContext ProcessContext) {
+func ClearCurrent(ctx context.Context, processContext
process_ctrl.ProcessContext) {
processContext.RemoveVariable(constant.VarNameCurrentLoopContextHolder)
}
diff --git a/pkg/saga/statemachine/engine/core/loop_task_utils.go
b/pkg/saga/statemachine/engine/pcext/loop_task_utils.go
similarity index 87%
rename from pkg/saga/statemachine/engine/core/loop_task_utils.go
rename to pkg/saga/statemachine/engine/pcext/loop_task_utils.go
index 18a3bec4..8544c3c0 100644
--- a/pkg/saga/statemachine/engine/core/loop_task_utils.go
+++ b/pkg/saga/statemachine/engine/pcext/loop_task_utils.go
@@ -15,21 +15,23 @@
* limitations under the License.
*/
-package core
+package pcext
import (
"context"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang/state"
"github.com/seata/seata-go/pkg/util/log"
)
-func GetLoopConfig(ctx context.Context, processContext ProcessContext,
currentState statelang.State) state.Loop {
+func GetLoopConfig(ctx context.Context, processContext
process_ctrl.ProcessContext, currentState statelang.State) state.Loop {
if matchLoop(currentState) {
taskState := currentState.(state.AbstractTaskState)
stateMachineInstance :=
processContext.GetVariable(constant.VarNameStateMachineInst).(statelang.StateMachineInstance)
- stateMachineConfig :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(StateMachineConfig)
+ stateMachineConfig :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(engine.StateMachineConfig)
if taskState.Loop() != nil {
loop := taskState.Loop()
diff --git a/pkg/saga/statemachine/engine/core/parameter_utils.go
b/pkg/saga/statemachine/engine/pcext/parameter_utils.go
similarity index 93%
rename from pkg/saga/statemachine/engine/core/parameter_utils.go
rename to pkg/saga/statemachine/engine/pcext/parameter_utils.go
index 56336b7d..72f99abe 100644
--- a/pkg/saga/statemachine/engine/core/parameter_utils.go
+++ b/pkg/saga/statemachine/engine/pcext/parameter_utils.go
@@ -15,19 +15,21 @@
* limitations under the License.
*/
-package core
+package pcext
import (
"fmt"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/expr"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang/state"
"strings"
"sync"
)
-func CreateInputParams(processContext ProcessContext, expressionResolver
expr.ExpressionResolver,
+func CreateInputParams(processContext process_ctrl.ProcessContext,
expressionResolver expr.ExpressionResolver,
stateInstance *statelang.StateInstanceImpl, serviceTaskState
*state.AbstractTaskState, variablesFrom any) []any {
inputAssignments := serviceTaskState.Input()
if inputAssignments == nil || len(inputAssignments) == 0 {
@@ -58,7 +60,7 @@ func CreateInputParams(processContext ProcessContext,
expressionResolver expr.Ex
return inputValues
}
-func CreateOutputParams(config StateMachineConfig, expressionResolver
expr.ExpressionResolver,
+func CreateOutputParams(config engine.StateMachineConfig, expressionResolver
expr.ExpressionResolver,
serviceTaskState *state.AbstractTaskState, variablesFrom any)
(map[string]any, error) {
outputAssignments := serviceTaskState.Output()
if outputAssignments == nil || len(outputAssignments) == 0 {
diff --git a/pkg/saga/statemachine/engine/core/process_state.go
b/pkg/saga/statemachine/engine/pcext/process_handler.go
similarity index 90%
rename from pkg/saga/statemachine/engine/core/process_state.go
rename to pkg/saga/statemachine/engine/pcext/process_handler.go
index 66dee8ff..41a07853 100644
--- a/pkg/saga/statemachine/engine/core/process_state.go
+++ b/pkg/saga/statemachine/engine/pcext/process_handler.go
@@ -15,17 +15,18 @@
* limitations under the License.
*/
-package core
+package pcext
import (
"context"
"errors"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"sync"
)
type StateHandler interface {
State() string
- ProcessHandler
+ process_ctrl.ProcessHandler
}
type InterceptAbleStateHandler interface {
@@ -34,13 +35,9 @@ type InterceptAbleStateHandler interface {
RegistryStateHandlerInterceptor(stateHandlerInterceptor
StateHandlerInterceptor)
}
-type ProcessHandler interface {
- Process(ctx context.Context, processContext ProcessContext) error
-}
-
type StateHandlerInterceptor interface {
- PreProcess(ctx context.Context, processContext ProcessContext) error
- PostProcess(ctx context.Context, processContext ProcessContext) error
+ PreProcess(ctx context.Context, processContext
process_ctrl.ProcessContext) error
+ PostProcess(ctx context.Context, processContext
process_ctrl.ProcessContext) error
Match(stateType string) bool
}
@@ -55,7 +52,7 @@ func NewStateMachineProcessHandler()
*StateMachineProcessHandler {
}
}
-func (s *StateMachineProcessHandler) Process(ctx context.Context,
processContext ProcessContext) error {
+func (s *StateMachineProcessHandler) Process(ctx context.Context,
processContext process_ctrl.ProcessContext) error {
stateInstruction, _ :=
processContext.GetInstruction().(StateInstruction)
state, err := stateInstruction.GetState(processContext)
diff --git a/pkg/saga/statemachine/engine/core/process_router.go
b/pkg/saga/statemachine/engine/pcext/process_router.go
similarity index 54%
copy from pkg/saga/statemachine/engine/core/process_router.go
copy to pkg/saga/statemachine/engine/pcext/process_router.go
index 1875cb16..2f57a1b1 100644
--- a/pkg/saga/statemachine/engine/core/process_router.go
+++ b/pkg/saga/statemachine/engine/pcext/process_router.go
@@ -15,102 +15,22 @@
* limitations under the License.
*/
-package core
+package pcext
import (
"context"
"github.com/pkg/errors"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
- "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl/process"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
- "github.com/seata/seata-go/pkg/util/log"
)
-type RouterHandler interface {
- Route(ctx context.Context, processContext ProcessContext) error
-}
-
-type ProcessRouter interface {
- Route(ctx context.Context, processContext ProcessContext) error
-}
-
-type InterceptAbleStateRouter interface {
- StateRouter
- StateRouterInterceptor() []StateRouterInterceptor
- RegistryStateRouterInterceptor(stateRouterInterceptor
StateRouterInterceptor)
-}
-
-type StateRouter interface {
- Route(ctx context.Context, processContext ProcessContext, state
statelang.State) (Instruction, error)
-}
-
-type StateRouterInterceptor interface {
- PreRoute(ctx context.Context, processContext ProcessContext, state
statelang.State) error
- PostRoute(ctx context.Context, processContext ProcessContext,
instruction Instruction, err error) error
- Match(stateType string) bool
-}
-
-type DefaultRouterHandler struct {
- eventPublisher EventPublisher
- processRouters map[string]ProcessRouter
-}
-
-func (d *DefaultRouterHandler) Route(ctx context.Context, processContext
ProcessContext) error {
- processType := d.matchProcessType(ctx, processContext)
- if processType == "" {
- log.Warnf("Process type not found, context= %s", processContext)
- return errors.New("Process type not found")
- }
-
- processRouter := d.processRouters[string(processType)]
- if processRouter == nil {
- log.Errorf("Cannot find process router by type %s, context =
%s", processType, processContext)
- return errors.New("Process router not found")
- }
-
- instruction := processRouter.Route(ctx, processContext)
- if instruction == nil {
- log.Info("route instruction is null, process end")
- } else {
- processContext.SetInstruction(instruction)
- _, err := d.eventPublisher.PushEvent(ctx, processContext)
- if err != nil {
- return err
- }
- }
-
- return nil
-}
-
-func (d *DefaultRouterHandler) matchProcessType(ctx context.Context,
processContext ProcessContext) process.ProcessType {
- processType, ok :=
processContext.GetVariable(constant.VarNameProcessType).(process.ProcessType)
- if !ok || processType == "" {
- processType = process.StateLang
- }
- return processType
-}
-
-func (d *DefaultRouterHandler) EventPublisher() EventPublisher {
- return d.eventPublisher
-}
-
-func (d *DefaultRouterHandler) SetEventPublisher(eventPublisher
EventPublisher) {
- d.eventPublisher = eventPublisher
-}
-
-func (d *DefaultRouterHandler) ProcessRouters() map[string]ProcessRouter {
- return d.processRouters
-}
-
-func (d *DefaultRouterHandler) SetProcessRouters(processRouters
map[string]ProcessRouter) {
- d.processRouters = processRouters
-}
-
type StateMachineProcessRouter struct {
- stateRouters map[string]StateRouter
+ stateRouters map[string]process_ctrl.StateRouter
}
-func (s *StateMachineProcessRouter) Route(ctx context.Context, processContext
ProcessContext) (Instruction, error) {
+func (s *StateMachineProcessRouter) Route(ctx context.Context, processContext
process_ctrl.ProcessContext) (process_ctrl.Instruction, error) {
stateInstruction, ok :=
processContext.GetInstruction().(StateInstruction)
if !ok {
return nil, errors.New("instruction is not a state instruction")
@@ -121,7 +41,7 @@ func (s *StateMachineProcessRouter) Route(ctx
context.Context, processContext Pr
state = stateInstruction.TemporaryState()
stateInstruction.SetTemporaryState(nil)
} else {
- stateMachineConfig, ok :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(StateMachineConfig)
+ stateMachineConfig, ok :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(engine.StateMachineConfig)
if !ok {
return nil, errors.New("state machine config not found")
}
@@ -138,16 +58,16 @@ func (s *StateMachineProcessRouter) Route(ctx
context.Context, processContext Pr
stateType := state.Type()
router := s.stateRouters[stateType]
- var interceptors []StateRouterInterceptor
- if interceptAbleStateRouter, ok := router.(InterceptAbleStateRouter);
ok {
+ var interceptors []process_ctrl.StateRouterInterceptor
+ if interceptAbleStateRouter, ok :=
router.(process_ctrl.InterceptAbleStateRouter); ok {
interceptors = interceptAbleStateRouter.StateRouterInterceptor()
}
- var executedInterceptors []StateRouterInterceptor
+ var executedInterceptors []process_ctrl.StateRouterInterceptor
var exception error
- instruction, exception := func() (Instruction, error) {
+ instruction, exception := func() (process_ctrl.Instruction, error) {
if interceptors == nil || len(executedInterceptors) == 0 {
- executedInterceptors = make([]StateRouterInterceptor,
0, len(interceptors))
+ executedInterceptors =
make([]process_ctrl.StateRouterInterceptor, 0, len(interceptors))
for _, interceptor := range interceptors {
executedInterceptors =
append(executedInterceptors, interceptor)
err := interceptor.PreRoute(ctx,
processContext, state)
@@ -186,7 +106,7 @@ func (s *StateMachineProcessRouter) Route(ctx
context.Context, processContext Pr
func (s *StateMachineProcessRouter) InitDefaultStateRouters() {
if s.stateRouters == nil || len(s.stateRouters) == 0 {
- s.stateRouters = make(map[string]StateRouter)
+ s.stateRouters = make(map[string]process_ctrl.StateRouter)
taskStateRouter := &TaskStateRouter{}
s.stateRouters[constant.StateTypeServiceTask] = taskStateRouter
s.stateRouters[constant.StateTypeScriptTask] = taskStateRouter
@@ -202,10 +122,10 @@ func (s *StateMachineProcessRouter)
InitDefaultStateRouters() {
}
}
-func (s *StateMachineProcessRouter) StateRouters() map[string]StateRouter {
+func (s *StateMachineProcessRouter) StateRouters()
map[string]process_ctrl.StateRouter {
return s.stateRouters
}
-func (s *StateMachineProcessRouter) SetStateRouters(stateRouters
map[string]StateRouter) {
+func (s *StateMachineProcessRouter) SetStateRouters(stateRouters
map[string]process_ctrl.StateRouter) {
s.stateRouters = stateRouters
}
diff --git a/pkg/saga/statemachine/engine/core/state_router.go
b/pkg/saga/statemachine/engine/pcext/state_router_impl.go
similarity index 91%
rename from pkg/saga/statemachine/engine/core/state_router.go
rename to pkg/saga/statemachine/engine/pcext/state_router_impl.go
index 73dd02e2..c7336b6f 100644
--- a/pkg/saga/statemachine/engine/core/state_router.go
+++ b/pkg/saga/statemachine/engine/pcext/state_router_impl.go
@@ -15,12 +15,13 @@
* limitations under the License.
*/
-package core
+package pcext
import (
"context"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/exception"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
sagaState
"github.com/seata/seata-go/pkg/saga/statemachine/statelang/state"
seataErrors "github.com/seata/seata-go/pkg/util/errors"
@@ -30,14 +31,14 @@ import (
type EndStateRouter struct {
}
-func (e EndStateRouter) Route(ctx context.Context, processContext
ProcessContext, state statelang.State) (Instruction, error) {
+func (e EndStateRouter) Route(ctx context.Context, processContext
process_ctrl.ProcessContext, state statelang.State) (process_ctrl.Instruction,
error) {
return nil, nil
}
type TaskStateRouter struct {
}
-func (t TaskStateRouter) Route(ctx context.Context, processContext
ProcessContext, state statelang.State) (Instruction, error) {
+func (t TaskStateRouter) Route(ctx context.Context, processContext
process_ctrl.ProcessContext, state statelang.State) (process_ctrl.Instruction,
error) {
stateInstruction, _ :=
processContext.GetInstruction().(StateInstruction)
if stateInstruction.End() {
log.Infof("StateInstruction is ended, Stop the StateMachine
executing. StateMachine[%s] Current State[%s]",
@@ -94,8 +95,8 @@ func (t TaskStateRouter) Route(ctx context.Context,
processContext ProcessContex
return stateInstruction, nil
}
-func (t *TaskStateRouter) compensateRoute(ctx context.Context, processContext
ProcessContext,
- compensationTriggerState statelang.State) (Instruction, error) {
+func (t *TaskStateRouter) compensateRoute(ctx context.Context, processContext
process_ctrl.ProcessContext,
+ compensationTriggerState statelang.State) (process_ctrl.Instruction,
error) {
//If there is already a compensation state that has been executed,
// it is judged whether it is wrong or unsuccessful,
// and the compensation process is interrupted.
@@ -141,11 +142,11 @@ func (t *TaskStateRouter) compensateRoute(ctx
context.Context, processContext Pr
GetCurrentCompensationHolder(ctx, processContext,
true).AddToBeCompensatedState(compensateState.Name(),
stateToBeCompensated)
- hierarchicalProcessContext :=
processContext.(HierarchicalProcessContext)
+ hierarchicalProcessContext :=
processContext.(process_ctrl.HierarchicalProcessContext)
hierarchicalProcessContext.SetVariableLocally(constant.VarNameFirstCompensationStateStarted,
true)
if _, ok :=
compensateState.(sagaState.CompensateSubStateMachineState); ok {
- hierarchicalProcessContext =
processContext.(HierarchicalProcessContext)
+ hierarchicalProcessContext =
processContext.(process_ctrl.HierarchicalProcessContext)
hierarchicalProcessContext.SetVariableLocally(
compensateState.Name()+constant.VarNameSubMachineParentId,
GenerateParentId(stateToBeCompensated))
diff --git a/pkg/saga/statemachine/store/repository/state_log_repository.go
b/pkg/saga/statemachine/engine/repo/repository/state_log_repository.go
similarity index 89%
rename from pkg/saga/statemachine/store/repository/state_log_repository.go
rename to pkg/saga/statemachine/engine/repo/repository/state_log_repository.go
index a9d970cc..8404a3d1 100644
--- a/pkg/saga/statemachine/store/repository/state_log_repository.go
+++ b/pkg/saga/statemachine/engine/repo/repository/state_log_repository.go
@@ -19,12 +19,10 @@ package repository
import (
"context"
- "database/sql"
"github.com/pkg/errors"
-
- "github.com/seata/seata-go/pkg/saga/statemachine/engine/core"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
- "github.com/seata/seata-go/pkg/saga/statemachine/store/db"
+ "github.com/seata/seata-go/pkg/saga/statemachine/store"
)
var (
@@ -32,23 +30,22 @@ var (
)
type StateLogRepositoryImpl struct {
- stateLogStore *db.StateLogStore
+ stateLogStore store.StateLogStore
}
-func NewStateLogRepositoryImpl(hsqldb *sql.DB, tablePrefix string)
*StateLogRepositoryImpl {
+func NewStateLogRepositoryImpl(stateLogStore store.StateLogStore)
*StateLogRepositoryImpl {
if stateLogRepositoryImpl == nil {
stateLogRepositoryImpl = &StateLogRepositoryImpl{
- stateLogStore: db.NewStateLogStore(hsqldb, tablePrefix),
+ stateLogStore: stateLogStore,
}
}
-
return stateLogRepositoryImpl
}
func (s *StateLogRepositoryImpl) RecordStateMachineStarted(
ctx context.Context,
machineInstance statelang.StateMachineInstance,
- processContext core.ProcessContext,
+ processContext process_ctrl.ProcessContext,
) error {
if s.stateLogStore == nil {
return errors.New("stateLogStore is not initialized")
@@ -59,7 +56,7 @@ func (s *StateLogRepositoryImpl) RecordStateMachineStarted(
func (s *StateLogRepositoryImpl) RecordStateMachineFinished(
ctx context.Context,
machineInstance statelang.StateMachineInstance,
- processContext core.ProcessContext,
+ processContext process_ctrl.ProcessContext,
) error {
if s.stateLogStore == nil {
return errors.New("stateLogStore is not initialized")
@@ -70,7 +67,7 @@ func (s *StateLogRepositoryImpl) RecordStateMachineFinished(
func (s *StateLogRepositoryImpl) RecordStateMachineRestarted(
ctx context.Context,
machineInstance statelang.StateMachineInstance,
- processContext core.ProcessContext,
+ processContext process_ctrl.ProcessContext,
) error {
if s.stateLogStore == nil {
return errors.New("stateLogStore is not initialized")
@@ -81,7 +78,7 @@ func (s *StateLogRepositoryImpl) RecordStateMachineRestarted(
func (s *StateLogRepositoryImpl) RecordStateStarted(
ctx context.Context,
stateInstance statelang.StateInstance,
- processContext core.ProcessContext,
+ processContext process_ctrl.ProcessContext,
) error {
if s.stateLogStore == nil {
return errors.New("stateLogStore is not initialized")
@@ -92,7 +89,7 @@ func (s *StateLogRepositoryImpl) RecordStateStarted(
func (s *StateLogRepositoryImpl) RecordStateFinished(
ctx context.Context,
stateInstance statelang.StateInstance,
- processContext core.ProcessContext,
+ processContext process_ctrl.ProcessContext,
) error {
if s.stateLogStore == nil {
return errors.New("stateLogStore is not initialized")
@@ -136,6 +133,6 @@ func (s *StateLogRepositoryImpl)
QueryStateInstanceListByMachineInstanceId(state
}
-func (s *StateLogRepositoryImpl) SetStateLogStore(stateLogStore
*db.StateLogStore) {
+func (s *StateLogRepositoryImpl) SetStateLogStore(stateLogStore
store.StateLogStore) {
s.stateLogStore = stateLogStore
}
diff --git a/pkg/saga/statemachine/store/repository/state_machine_repository.go
b/pkg/saga/statemachine/engine/repo/repository/state_machine_repository.go
similarity index 98%
rename from pkg/saga/statemachine/store/repository/state_machine_repository.go
rename to
pkg/saga/statemachine/engine/repo/repository/state_machine_repository.go
index 6628d8dc..5f535a31 100644
--- a/pkg/saga/statemachine/store/repository/state_machine_repository.go
+++ b/pkg/saga/statemachine/engine/repo/repository/state_machine_repository.go
@@ -23,10 +23,10 @@ import (
"time"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
- "github.com/seata/seata-go/pkg/saga/statemachine/engine/core"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/sequence"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang/parser"
+ "github.com/seata/seata-go/pkg/saga/statemachine/store"
"github.com/seata/seata-go/pkg/util/log"
)
@@ -43,7 +43,7 @@ type StateMachineRepositoryImpl struct {
stateMachineMapById map[string]statelang.StateMachine
stateMachineMapByNameAndTenant map[string]statelang.StateMachine
- stateLangStore core.StateLangStore
+ stateLangStore store.StateLangStore
seqGenerator sequence.SeqGenerator
defaultTenantId string
jsonParserName string
@@ -204,7 +204,7 @@ func (s *StateMachineRepositoryImpl)
RegistryStateMachineByReader(reader io.Read
return nil
}
-func (s *StateMachineRepositoryImpl) SetStateLangStore(stateLangStore
core.StateLangStore) {
+func (s *StateMachineRepositoryImpl) SetStateLangStore(stateLangStore
store.StateLangStore) {
s.stateLangStore = stateLangStore
}
diff --git
a/pkg/saga/statemachine/store/repository/state_machine_repository_test.go
b/pkg/saga/statemachine/engine/repo/repository/state_machine_repository_test.go
similarity index 100%
rename from
pkg/saga/statemachine/store/repository/state_machine_repository_test.go
rename to
pkg/saga/statemachine/engine/repo/repository/state_machine_repository_test.go
diff --git a/pkg/saga/statemachine/engine/core/statemachine_store.go
b/pkg/saga/statemachine/engine/repo/statemachine_store.go
similarity index 55%
rename from pkg/saga/statemachine/engine/core/statemachine_store.go
rename to pkg/saga/statemachine/engine/repo/statemachine_store.go
index 7e0a9e94..5a426b17 100644
--- a/pkg/saga/statemachine/engine/core/statemachine_store.go
+++ b/pkg/saga/statemachine/engine/repo/statemachine_store.go
@@ -15,10 +15,9 @@
* limitations under the License.
*/
-package core
+package repo
import (
- "context"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"io"
)
@@ -35,30 +34,6 @@ type StateLogRepository interface {
GetStateInstanceListByMachineInstanceId(stateMachineInstanceId string)
([]statelang.StateInstance, error)
}
-type StateLogStore interface {
- RecordStateMachineStarted(ctx context.Context, machineInstance
statelang.StateMachineInstance, context ProcessContext) error
-
- RecordStateMachineFinished(ctx context.Context, machineInstance
statelang.StateMachineInstance, context ProcessContext) error
-
- RecordStateMachineRestarted(ctx context.Context, machineInstance
statelang.StateMachineInstance, context ProcessContext) error
-
- RecordStateStarted(ctx context.Context, stateInstance
statelang.StateInstance, context ProcessContext) error
-
- RecordStateFinished(ctx context.Context, stateInstance
statelang.StateInstance, context ProcessContext) error
-
- GetStateMachineInstance(stateMachineInstanceId string)
(statelang.StateMachineInstance, error)
-
- GetStateMachineInstanceByBusinessKey(businessKey string, tenantId
string) (statelang.StateMachineInstance, error)
-
- GetStateMachineInstanceByParentId(parentId string)
([]statelang.StateMachineInstance, error)
-
- GetStateInstance(stateInstanceId string, stateMachineInstanceId string)
(statelang.StateInstance, error)
-
- GetStateInstanceListByMachineInstanceId(stateMachineInstanceId string)
([]statelang.StateInstance, error)
-
- ClearUp(context ProcessContext)
-}
-
type StateMachineRepository interface {
GetStateMachineById(stateMachineId string) (statelang.StateMachine,
error)
@@ -70,11 +45,3 @@ type StateMachineRepository interface {
RegistryStateMachineByReader(reader io.Reader) error
}
-
-type StateLangStore interface {
- GetStateMachineById(stateMachineId string) (statelang.StateMachine,
error)
-
- GetLastVersionStateMachine(stateMachineName string, tenantId string)
(statelang.StateMachine, error)
-
- StoreStateMachine(stateMachine statelang.StateMachine) error
-}
diff --git a/pkg/saga/statemachine/engine/core/statemachine_config.go
b/pkg/saga/statemachine/engine/statemachine_config.go
similarity index 76%
rename from pkg/saga/statemachine/engine/core/statemachine_config.go
rename to pkg/saga/statemachine/engine/statemachine_config.go
index e3f9da29..bd24fbc9 100644
--- a/pkg/saga/statemachine/engine/core/statemachine_config.go
+++ b/pkg/saga/statemachine/engine/statemachine_config.go
@@ -15,23 +15,26 @@
* limitations under the License.
*/
-package core
+package engine
import (
"github.com/seata/seata-go/pkg/saga/statemachine/engine/expr"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/invoker"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/repo"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/sequence"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
+ "github.com/seata/seata-go/pkg/saga/statemachine/store"
"sync"
)
type StateMachineConfig interface {
- StateLogRepository() StateLogRepository
+ StateLogRepository() repo.StateLogRepository
- StateMachineRepository() StateMachineRepository
+ StateMachineRepository() repo.StateMachineRepository
- StateLogStore() StateLogStore
+ StateLogStore() store.StateLogStore
- StateLangStore() StateLangStore
+ StateLangStore() store.StateLangStore
ExpressionFactoryManager() expr.ExpressionFactoryManager
@@ -41,9 +44,9 @@ type StateMachineConfig interface {
StatusDecisionStrategy() StatusDecisionStrategy
- EventPublisher() EventPublisher
+ EventPublisher() process_ctrl.EventPublisher
- AsyncEventPublisher() EventPublisher
+ AsyncEventPublisher() process_ctrl.EventPublisher
ServiceInvokerManager() invoker.ServiceInvokerManager
diff --git a/pkg/saga/statemachine/engine/core/statemachine_engine.go
b/pkg/saga/statemachine/engine/statemachine_engine.go
similarity index 91%
rename from pkg/saga/statemachine/engine/core/statemachine_engine.go
rename to pkg/saga/statemachine/engine/statemachine_engine.go
index 6ad36204..7b65e4c8 100644
--- a/pkg/saga/statemachine/engine/core/statemachine_engine.go
+++ b/pkg/saga/statemachine/engine/statemachine_engine.go
@@ -15,10 +15,11 @@
* limitations under the License.
*/
-package core
+package engine
import (
"context"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
)
@@ -53,6 +54,6 @@ type StateMachineEngine interface {
}
type CallBack interface {
- OnFinished(ctx context.Context, context ProcessContext,
stateMachineInstance statelang.StateMachineInstance)
- OnError(ctx context.Context, context ProcessContext,
stateMachineInstance statelang.StateMachineInstance, err error)
+ OnFinished(ctx context.Context, context process_ctrl.ProcessContext,
stateMachineInstance statelang.StateMachineInstance)
+ OnError(ctx context.Context, context process_ctrl.ProcessContext,
stateMachineInstance statelang.StateMachineInstance, err error)
}
diff --git a/pkg/saga/statemachine/engine/core/statemachine_engine_test.go
b/pkg/saga/statemachine/engine/statemachine_engine_test.go
similarity index 88%
rename from pkg/saga/statemachine/engine/core/statemachine_engine_test.go
rename to pkg/saga/statemachine/engine/statemachine_engine_test.go
index 8af2cdfc..fc049d12 100644
--- a/pkg/saga/statemachine/engine/core/statemachine_engine_test.go
+++ b/pkg/saga/statemachine/engine/statemachine_engine_test.go
@@ -15,10 +15,11 @@
* limitations under the License.
*/
-package core
+package engine
import (
"context"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/core"
"testing"
)
@@ -27,6 +28,6 @@ func TestEngine(t *testing.T) {
}
func TestSimpleStateMachine(t *testing.T) {
- engine := NewProcessCtrlStateMachineEngine()
+ engine := core.NewProcessCtrlStateMachineEngine()
engine.Start(context.Background(), "simpleStateMachine", "tenantId",
nil)
}
diff --git a/pkg/saga/statemachine/engine/strategy.go
b/pkg/saga/statemachine/engine/strategy.go
new file mode 100644
index 00000000..f400d85f
--- /dev/null
+++ b/pkg/saga/statemachine/engine/strategy.go
@@ -0,0 +1,19 @@
+package engine
+
+import (
+ "context"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
+ "github.com/seata/seata-go/pkg/saga/statemachine/statelang"
+)
+
+type StatusDecisionStrategy interface {
+ // DecideOnEndState Determine state machine execution status when
executing to EndState
+ DecideOnEndState(ctx context.Context, processContext
process_ctrl.ProcessContext,
+ stateMachineInstance statelang.StateMachineInstance, exp error)
error
+ // DecideOnTaskStateFail Determine state machine execution status when
executing TaskState error
+ DecideOnTaskStateFail(ctx context.Context, processContext
process_ctrl.ProcessContext,
+ stateMachineInstance statelang.StateMachineInstance, exp error)
error
+ // DecideMachineForwardExecutionStatus Determine the forward execution
state of the state machine
+ DecideMachineForwardExecutionStatus(ctx context.Context,
+ stateMachineInstance statelang.StateMachineInstance, exp error,
specialPolicy bool) error
+}
diff --git a/pkg/saga/statemachine/engine/core/status_decision.go
b/pkg/saga/statemachine/engine/strategy/status_decision.go
similarity index 89%
rename from pkg/saga/statemachine/engine/core/status_decision.go
rename to pkg/saga/statemachine/engine/strategy/status_decision.go
index 348ac4dd..6e453b59 100644
--- a/pkg/saga/statemachine/engine/core/status_decision.go
+++ b/pkg/saga/statemachine/engine/strategy/status_decision.go
@@ -15,29 +15,19 @@
* limitations under the License.
*/
-package core
+package strategy
import (
"context"
"errors"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/exception"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/pcext"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"github.com/seata/seata-go/pkg/util/log"
)
-type StatusDecisionStrategy interface {
- // DecideOnEndState Determine state machine execution status when
executing to EndState
- DecideOnEndState(ctx context.Context, processContext ProcessContext,
- stateMachineInstance statelang.StateMachineInstance, exp error)
error
- // DecideOnTaskStateFail Determine state machine execution status when
executing TaskState error
- DecideOnTaskStateFail(ctx context.Context, processContext
ProcessContext,
- stateMachineInstance statelang.StateMachineInstance, exp error)
error
- // DecideMachineForwardExecutionStatus Determine the forward execution
state of the state machine
- DecideMachineForwardExecutionStatus(ctx context.Context,
- stateMachineInstance statelang.StateMachineInstance, exp error,
specialPolicy bool) error
-}
-
type DefaultStatusDecisionStrategy struct {
}
@@ -45,10 +35,10 @@ func NewDefaultStatusDecisionStrategy()
*DefaultStatusDecisionStrategy {
return &DefaultStatusDecisionStrategy{}
}
-func (d DefaultStatusDecisionStrategy) DecideOnEndState(ctx context.Context,
processContext ProcessContext,
+func (d DefaultStatusDecisionStrategy) DecideOnEndState(ctx context.Context,
processContext process_ctrl.ProcessContext,
stateMachineInstance statelang.StateMachineInstance, exp error) error {
if statelang.RU == stateMachineInstance.CompensationStatus() {
- compensationHolder := GetCurrentCompensationHolder(ctx,
processContext, true)
+ compensationHolder := pcext.GetCurrentCompensationHolder(ctx,
processContext, true)
if err := decideMachineCompensateStatus(ctx,
stateMachineInstance, compensationHolder); err != nil {
return err
}
@@ -74,7 +64,7 @@ func (d DefaultStatusDecisionStrategy) DecideOnEndState(ctx
context.Context, pro
return nil
}
-func decideMachineCompensateStatus(ctx context.Context, stateMachineInstance
statelang.StateMachineInstance, compensationHolder *CompensationHolder) error {
+func decideMachineCompensateStatus(ctx context.Context, stateMachineInstance
statelang.StateMachineInstance, compensationHolder *pcext.CompensationHolder)
error {
if stateMachineInstance.Status() == "" || statelang.RU ==
stateMachineInstance.Status() {
stateMachineInstance.SetStatus(statelang.UN)
}
@@ -211,7 +201,7 @@ func setMachineStatusBasedOnException(stateMachineInstance
statelang.StateMachin
return
}
- netType := GetNetExceptionType(exp)
+ netType := pcext.GetNetExceptionType(exp)
switch netType {
case constant.ConnectException, constant.ConnectTimeoutException,
constant.NotNetException:
log.Warnf("Detected network connect issue, setting
StateMachineInstance[id:%s] status to FA", stateMachineInstance.ID())
@@ -227,7 +217,7 @@ func setMachineStatusBasedOnException(stateMachineInstance
statelang.StateMachin
}
}
-func (d DefaultStatusDecisionStrategy) DecideOnTaskStateFail(ctx
context.Context, processContext ProcessContext,
+func (d DefaultStatusDecisionStrategy) DecideOnTaskStateFail(ctx
context.Context, processContext process_ctrl.ProcessContext,
stateMachineInstance statelang.StateMachineInstance, exp error) error {
log.Debugf("Starting DecideOnTaskStateFail for
StateMachineInstance[id:%s]", stateMachineInstance.ID())
diff --git a/pkg/saga/statemachine/engine/core/utils.go
b/pkg/saga/statemachine/engine/utils/process_context_utils.go
similarity index 83%
rename from pkg/saga/statemachine/engine/core/utils.go
rename to pkg/saga/statemachine/engine/utils/process_context_utils.go
index afa1fd66..3e35c696 100644
--- a/pkg/saga/statemachine/engine/core/utils.go
+++ b/pkg/saga/statemachine/engine/utils/process_context_utils.go
@@ -15,21 +15,23 @@
* limitations under the License.
*/
-package core
+package utils
import (
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl/process"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
)
// ProcessContextBuilder process_ctrl builder
type ProcessContextBuilder struct {
- processContext ProcessContext
+ processContext process_ctrl.ProcessContext
}
func NewProcessContextBuilder() *ProcessContextBuilder {
- processContextImpl := NewProcessContextImpl()
+ processContextImpl := process_ctrl.NewProcessContextImpl()
return &ProcessContextBuilder{processContextImpl}
}
@@ -43,7 +45,7 @@ func (p *ProcessContextBuilder)
WithOperationName(operationName string) *Process
return p
}
-func (p *ProcessContextBuilder) WithAsyncCallback(callBack CallBack)
*ProcessContextBuilder {
+func (p *ProcessContextBuilder) WithAsyncCallback(callBack engine.CallBack)
*ProcessContextBuilder {
if callBack != nil {
p.processContext.SetVariable(constant.VarNameAsyncCallback,
callBack)
}
@@ -51,7 +53,7 @@ func (p *ProcessContextBuilder) WithAsyncCallback(callBack
CallBack) *ProcessCon
return p
}
-func (p *ProcessContextBuilder) WithInstruction(instruction Instruction)
*ProcessContextBuilder {
+func (p *ProcessContextBuilder) WithInstruction(instruction
process_ctrl.Instruction) *ProcessContextBuilder {
if instruction != nil {
p.processContext.SetInstruction(instruction)
}
@@ -68,7 +70,7 @@ func (p *ProcessContextBuilder)
WithStateMachineInstance(stateMachineInstance st
return p
}
-func (p *ProcessContextBuilder) WithStateMachineEngine(stateMachineEngine
StateMachineEngine) *ProcessContextBuilder {
+func (p *ProcessContextBuilder) WithStateMachineEngine(stateMachineEngine
engine.StateMachineEngine) *ProcessContextBuilder {
if stateMachineEngine != nil {
p.processContext.SetVariable(constant.VarNameStateMachineEngine,
stateMachineEngine)
}
@@ -76,7 +78,7 @@ func (p *ProcessContextBuilder)
WithStateMachineEngine(stateMachineEngine StateM
return p
}
-func (p *ProcessContextBuilder) WithStateMachineConfig(stateMachineConfig
StateMachineConfig) *ProcessContextBuilder {
+func (p *ProcessContextBuilder) WithStateMachineConfig(stateMachineConfig
engine.StateMachineConfig) *ProcessContextBuilder {
if stateMachineConfig != nil {
p.processContext.SetVariable(constant.VarNameStateMachineConfig,
stateMachineConfig)
}
@@ -106,6 +108,6 @@ func (p *ProcessContextBuilder) WithStateInstance(state
statelang.StateInstance)
return p
}
-func (p *ProcessContextBuilder) Build() ProcessContext {
+func (p *ProcessContextBuilder) Build() process_ctrl.ProcessContext {
return p.processContext
}
diff --git a/pkg/saga/statemachine/engine/core/bussiness_processor.go
b/pkg/saga/statemachine/process_ctrl/bussiness_processor.go
similarity index 99%
rename from pkg/saga/statemachine/engine/core/bussiness_processor.go
rename to pkg/saga/statemachine/process_ctrl/bussiness_processor.go
index 38846dff..5c830d21 100644
--- a/pkg/saga/statemachine/engine/core/bussiness_processor.go
+++ b/pkg/saga/statemachine/process_ctrl/bussiness_processor.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package core
+package process_ctrl
import (
"context"
diff --git a/pkg/saga/statemachine/process_ctrl/default_process_handler.go
b/pkg/saga/statemachine/process_ctrl/default_process_handler.go
new file mode 100644
index 00000000..e4b4702b
--- /dev/null
+++ b/pkg/saga/statemachine/process_ctrl/default_process_handler.go
@@ -0,0 +1,7 @@
+package process_ctrl
+
+import "context"
+
+type ProcessHandler interface {
+ Process(ctx context.Context, processContext ProcessContext) error
+}
diff --git a/pkg/saga/statemachine/engine/core/event.go
b/pkg/saga/statemachine/process_ctrl/event.go
similarity index 97%
copy from pkg/saga/statemachine/engine/core/event.go
copy to pkg/saga/statemachine/process_ctrl/event.go
index 871c0ce9..377c40cb 100644
--- a/pkg/saga/statemachine/engine/core/event.go
+++ b/pkg/saga/statemachine/process_ctrl/event.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package core
+package process_ctrl
type Event interface {
}
diff --git a/pkg/saga/statemachine/engine/core/event_bus.go
b/pkg/saga/statemachine/process_ctrl/event_bus.go
similarity index 99%
rename from pkg/saga/statemachine/engine/core/event_bus.go
rename to pkg/saga/statemachine/process_ctrl/event_bus.go
index e793419f..49c282ef 100644
--- a/pkg/saga/statemachine/engine/core/event_bus.go
+++ b/pkg/saga/statemachine/process_ctrl/event_bus.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package core
+package process_ctrl
import (
"context"
diff --git a/pkg/saga/statemachine/engine/core/event_consumer.go
b/pkg/saga/statemachine/process_ctrl/event_consumer.go
similarity index 98%
rename from pkg/saga/statemachine/engine/core/event_consumer.go
rename to pkg/saga/statemachine/process_ctrl/event_consumer.go
index 25f8762e..5058b1f3 100644
--- a/pkg/saga/statemachine/engine/core/event_consumer.go
+++ b/pkg/saga/statemachine/process_ctrl/event_consumer.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package core
+package process_ctrl
import (
"context"
diff --git a/pkg/saga/statemachine/engine/core/event_publisher.go
b/pkg/saga/statemachine/process_ctrl/event_publisher.go
similarity index 98%
rename from pkg/saga/statemachine/engine/core/event_publisher.go
rename to pkg/saga/statemachine/process_ctrl/event_publisher.go
index 447dee30..d18db8a7 100644
--- a/pkg/saga/statemachine/engine/core/event_publisher.go
+++ b/pkg/saga/statemachine/process_ctrl/event_publisher.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package core
+package process_ctrl
import "context"
diff --git
a/pkg/saga/statemachine/process_ctrl/handlers/service_task_state_handler.go
b/pkg/saga/statemachine/process_ctrl/handlers/service_task_state_handler.go
index f2a7aff2..e4298b02 100644
--- a/pkg/saga/statemachine/process_ctrl/handlers/service_task_state_handler.go
+++ b/pkg/saga/statemachine/process_ctrl/handlers/service_task_state_handler.go
@@ -21,8 +21,10 @@ import (
"context"
"errors"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
- "github.com/seata/seata-go/pkg/saga/statemachine/engine/core"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/exception"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/pcext"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang/state"
seataErrors "github.com/seata/seata-go/pkg/util/errors"
@@ -30,7 +32,7 @@ import (
)
type ServiceTaskStateHandler struct {
- interceptors []core.StateHandlerInterceptor
+ interceptors []pcext.StateHandlerInterceptor
}
func NewServiceTaskStateHandler() *ServiceTaskStateHandler {
@@ -41,8 +43,8 @@ func (s *ServiceTaskStateHandler) State() string {
return constant.StateTypeServiceTask
}
-func (s *ServiceTaskStateHandler) Process(ctx context.Context, processContext
core.ProcessContext) error {
- stateInstruction, ok :=
processContext.GetInstruction().(core.StateInstruction)
+func (s *ServiceTaskStateHandler) Process(ctx context.Context, processContext
process_ctrl.ProcessContext) error {
+ stateInstruction, ok :=
processContext.GetInstruction().(pcext.StateInstruction)
if !ok {
return errors.New("invalid state instruction from
processContext")
}
@@ -66,12 +68,12 @@ func (s *ServiceTaskStateHandler) Process(ctx
context.Context, processContext co
log.Error("<<<<<<<<<<<<<<<<<<<<<< State[%s], ServiceName[%s],
Method[%s] Execute failed.",
serviceTaskStateImpl.Name(), serviceName, methodName,
err)
- hierarchicalProcessContext, ok :=
processContext.(core.HierarchicalProcessContext)
+ hierarchicalProcessContext, ok :=
processContext.(process_ctrl.HierarchicalProcessContext)
if !ok {
return
}
hierarchicalProcessContext.SetVariable(constant.VarNameCurrentException, err)
- core.HandleException(processContext,
serviceTaskStateImpl.AbstractTaskState, err)
+ pcext.HandleException(processContext,
serviceTaskStateImpl.AbstractTaskState, err)
}
input, ok :=
processContext.GetVariable(constant.VarNameInputParams).([]any)
@@ -87,7 +89,7 @@ func (s *ServiceTaskStateHandler) Process(ctx
context.Context, processContext co
if _, ok := stateInterface.(state.CompensateSubStateMachineState); ok {
// If it is the compensation of the subState machine,
// directly call the state machine's compensate method
- stateMachineEngine, ok :=
processContext.GetVariable(constant.VarNameStateMachineEngine).(core.StateMachineEngine)
+ stateMachineEngine, ok :=
processContext.GetVariable(constant.VarNameStateMachineEngine).(engine.StateMachineEngine)
if !ok {
handleResultErr(errors.New("invalid stateMachineEngine
type from processContext"))
return nil
@@ -100,7 +102,7 @@ func (s *ServiceTaskStateHandler) Process(ctx
context.Context, processContext co
return nil
}
} else {
- stateMachineConfig, ok :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(core.StateMachineConfig)
+ stateMachineConfig, ok :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(engine.StateMachineConfig)
if !ok {
handleResultErr(errors.New("invalid stateMachineConfig
type from processContext"))
return nil
@@ -126,7 +128,7 @@ func (s *ServiceTaskStateHandler) Process(ctx
context.Context, processContext co
if result != nil {
stateInstance.SetOutputParams(result)
- hierarchicalProcessContext, ok :=
processContext.(core.HierarchicalProcessContext)
+ hierarchicalProcessContext, ok :=
processContext.(process_ctrl.HierarchicalProcessContext)
if !ok {
handleResultErr(errors.New("invalid hierarchical
process context type from processContext"))
return nil
@@ -138,17 +140,17 @@ func (s *ServiceTaskStateHandler) Process(ctx
context.Context, processContext co
return nil
}
-func (s *ServiceTaskStateHandler) StateHandlerInterceptorList()
[]core.StateHandlerInterceptor {
+func (s *ServiceTaskStateHandler) StateHandlerInterceptorList()
[]pcext.StateHandlerInterceptor {
return s.interceptors
}
-func (s *ServiceTaskStateHandler)
RegistryStateHandlerInterceptor(stateHandlerInterceptor
core.StateHandlerInterceptor) {
+func (s *ServiceTaskStateHandler)
RegistryStateHandlerInterceptor(stateHandlerInterceptor
pcext.StateHandlerInterceptor) {
s.interceptors = append(s.interceptors, stateHandlerInterceptor)
}
-func (s *ServiceTaskStateHandler) compensateSubStateMachine(ctx
context.Context, processContext core.ProcessContext,
+func (s *ServiceTaskStateHandler) compensateSubStateMachine(ctx
context.Context, processContext process_ctrl.ProcessContext,
serviceTaskState state.ServiceTaskState, input any, instance
statelang.StateInstance,
- machineEngine core.StateMachineEngine) (any, error) {
+ machineEngine engine.StateMachineEngine) (any, error) {
subStateMachineParentId, ok :=
processContext.GetVariable(serviceTaskState.Name() +
constant.VarNameSubMachineParentId).(string)
if !ok {
return nil, errors.New("invalid subStateMachineParentId type
from processContext")
@@ -159,7 +161,7 @@ func (s *ServiceTaskStateHandler)
compensateSubStateMachine(ctx context.Context,
"sub statemachine parentId is required", nil)
}
- stateMachineConfig :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(core.StateMachineConfig)
+ stateMachineConfig :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(engine.StateMachineConfig)
subInst, err :=
stateMachineConfig.StateLogStore().GetStateMachineInstanceByParentId(subStateMachineParentId)
if err != nil {
return nil, err
diff --git a/pkg/saga/statemachine/engine/core/event.go
b/pkg/saga/statemachine/process_ctrl/instruction.go
similarity index 94%
rename from pkg/saga/statemachine/engine/core/event.go
rename to pkg/saga/statemachine/process_ctrl/instruction.go
index 871c0ce9..8469f0c6 100644
--- a/pkg/saga/statemachine/engine/core/event.go
+++ b/pkg/saga/statemachine/process_ctrl/instruction.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package core
+package process_ctrl
-type Event interface {
+type Instruction interface {
}
diff --git a/pkg/saga/statemachine/engine/core/process_context.go
b/pkg/saga/statemachine/process_ctrl/process_context.go
similarity index 99%
rename from pkg/saga/statemachine/engine/core/process_context.go
rename to pkg/saga/statemachine/process_ctrl/process_context.go
index 163ea640..92b16e30 100644
--- a/pkg/saga/statemachine/engine/core/process_context.go
+++ b/pkg/saga/statemachine/process_ctrl/process_context.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package core
+package process_ctrl
import (
"sync"
diff --git a/pkg/saga/statemachine/engine/core/process_controller.go
b/pkg/saga/statemachine/process_ctrl/process_controller.go
similarity index 98%
rename from pkg/saga/statemachine/engine/core/process_controller.go
rename to pkg/saga/statemachine/process_ctrl/process_controller.go
index 4c987db5..fcb305ea 100644
--- a/pkg/saga/statemachine/engine/core/process_controller.go
+++ b/pkg/saga/statemachine/process_ctrl/process_controller.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package core
+package process_ctrl
import (
"context"
diff --git a/pkg/saga/statemachine/engine/core/process_router.go
b/pkg/saga/statemachine/process_ctrl/process_router.go
similarity index 50%
rename from pkg/saga/statemachine/engine/core/process_router.go
rename to pkg/saga/statemachine/process_ctrl/process_router.go
index 1875cb16..37b1501e 100644
--- a/pkg/saga/statemachine/engine/core/process_router.go
+++ b/pkg/saga/statemachine/process_ctrl/process_router.go
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package core
+package process_ctrl
import (
"context"
@@ -105,107 +105,3 @@ func (d *DefaultRouterHandler) ProcessRouters()
map[string]ProcessRouter {
func (d *DefaultRouterHandler) SetProcessRouters(processRouters
map[string]ProcessRouter) {
d.processRouters = processRouters
}
-
-type StateMachineProcessRouter struct {
- stateRouters map[string]StateRouter
-}
-
-func (s *StateMachineProcessRouter) Route(ctx context.Context, processContext
ProcessContext) (Instruction, error) {
- stateInstruction, ok :=
processContext.GetInstruction().(StateInstruction)
- if !ok {
- return nil, errors.New("instruction is not a state instruction")
- }
-
- var state statelang.State
- if stateInstruction.TemporaryState() != nil {
- state = stateInstruction.TemporaryState()
- stateInstruction.SetTemporaryState(nil)
- } else {
- stateMachineConfig, ok :=
processContext.GetVariable(constant.VarNameStateMachineConfig).(StateMachineConfig)
- if !ok {
- return nil, errors.New("state machine config not found")
- }
-
- stateMachine, err :=
stateMachineConfig.StateMachineRepository().GetStateMachineByNameAndTenantId(stateInstruction.StateMachineName(),
- stateInstruction.TenantId())
- if err != nil {
- return nil, err
- }
-
- state = stateMachine.States()[stateInstruction.StateName()]
- }
-
- stateType := state.Type()
- router := s.stateRouters[stateType]
-
- var interceptors []StateRouterInterceptor
- if interceptAbleStateRouter, ok := router.(InterceptAbleStateRouter);
ok {
- interceptors = interceptAbleStateRouter.StateRouterInterceptor()
- }
-
- var executedInterceptors []StateRouterInterceptor
- var exception error
- instruction, exception := func() (Instruction, error) {
- if interceptors == nil || len(executedInterceptors) == 0 {
- executedInterceptors = make([]StateRouterInterceptor,
0, len(interceptors))
- for _, interceptor := range interceptors {
- executedInterceptors =
append(executedInterceptors, interceptor)
- err := interceptor.PreRoute(ctx,
processContext, state)
- if err != nil {
- return nil, err
- }
- }
- }
-
- instruction, err := router.Route(ctx, processContext, state)
- if err != nil {
- return nil, err
- }
- return instruction, nil
- }()
-
- if interceptors == nil || len(executedInterceptors) == 0 {
- for i := len(executedInterceptors) - 1; i >= 0; i-- {
- err := executedInterceptors[i].PostRoute(ctx,
processContext, instruction, exception)
- if err != nil {
- return nil, err
- }
- }
-
- // if 'Succeed' or 'Fail' State did not configured, we must end
the state machine
- if instruction == nil && !stateInstruction.End() {
- err := EndStateMachine(ctx, processContext)
- if err != nil {
- return nil, err
- }
- }
- }
-
- return instruction, nil
-}
-
-func (s *StateMachineProcessRouter) InitDefaultStateRouters() {
- if s.stateRouters == nil || len(s.stateRouters) == 0 {
- s.stateRouters = make(map[string]StateRouter)
- taskStateRouter := &TaskStateRouter{}
- s.stateRouters[constant.StateTypeServiceTask] = taskStateRouter
- s.stateRouters[constant.StateTypeScriptTask] = taskStateRouter
- s.stateRouters[constant.StateTypeChoice] = taskStateRouter
- s.stateRouters[constant.StateTypeCompensationTrigger] =
taskStateRouter
- s.stateRouters[constant.StateTypeSubStateMachine] =
taskStateRouter
- s.stateRouters[constant.StateTypeCompensateSubMachine] =
taskStateRouter
- s.stateRouters[constant.StateTypeLoopStart] = taskStateRouter
-
- endStateRouter := &EndStateRouter{}
- s.stateRouters[constant.StateTypeSucceed] = endStateRouter
- s.stateRouters[constant.StateTypeFail] = endStateRouter
- }
-}
-
-func (s *StateMachineProcessRouter) StateRouters() map[string]StateRouter {
- return s.stateRouters
-}
-
-func (s *StateMachineProcessRouter) SetStateRouters(stateRouters
map[string]StateRouter) {
- s.stateRouters = stateRouters
-}
diff --git a/pkg/saga/statemachine/store/db/statelog.go
b/pkg/saga/statemachine/store/db/statelog.go
index a9235dbb..ccf70798 100644
--- a/pkg/saga/statemachine/store/db/statelog.go
+++ b/pkg/saga/statemachine/store/db/statelog.go
@@ -21,6 +21,10 @@ import (
"context"
"database/sql"
"fmt"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/config"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/pcext"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"regexp"
"strconv"
"strings"
@@ -32,7 +36,6 @@ import (
"github.com/seata/seata-go/pkg/protocol/message"
"github.com/seata/seata-go/pkg/rm"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
- "github.com/seata/seata-go/pkg/saga/statemachine/engine/core"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/sequence"
"github.com/seata/seata-go/pkg/saga/statemachine/engine/serializer"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
@@ -113,7 +116,7 @@ func NewStateLogStore(db *sql.DB, tablePrefix string)
*StateLogStore {
}
func (s *StateLogStore) RecordStateMachineStarted(ctx context.Context,
machineInstance statelang.StateMachineInstance,
- context core.ProcessContext) error {
+ context process_ctrl.ProcessContext) error {
if machineInstance == nil {
return nil
}
@@ -161,8 +164,8 @@ func (s *StateLogStore) RecordStateMachineStarted(ctx
context.Context, machineIn
return nil
}
-func (s *StateLogStore) beginTransaction(ctx context.Context, machineInstance
statelang.StateMachineInstance, context core.ProcessContext) error {
- cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(core.StateMachineConfig)
+func (s *StateLogStore) beginTransaction(ctx context.Context, machineInstance
statelang.StateMachineInstance, context process_ctrl.ProcessContext) error {
+ cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(engine.StateMachineConfig)
if !ok {
return errors.New("begin transaction fail, stateMachineConfig
is required in context")
}
@@ -188,7 +191,7 @@ func (s *StateLogStore) beginTransaction(ctx
context.Context, machineInstance st
}
func (s *StateLogStore) RecordStateMachineFinished(ctx context.Context,
machineInstance statelang.StateMachineInstance,
- context core.ProcessContext) error {
+ context process_ctrl.ProcessContext) error {
if machineInstance == nil {
return nil
}
@@ -230,12 +233,12 @@ func (s *StateLogStore) RecordStateMachineFinished(ctx
context.Context, machineI
}
// check if timeout or else report transaction finished
- cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(core.StateMachineConfig)
+ cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(engine.StateMachineConfig)
if !ok {
return errors.New("stateMachineConfig is required in context")
}
- if core.IsTimeout(machineInstance.UpdatedTime(),
cfg.TransOperationTimeout()) {
+ if pcext.IsTimeout(machineInstance.UpdatedTime(),
cfg.TransOperationTimeout()) {
log.Warnf("StateMachineInstance[%s] is execution timeout, skip
report transaction finished to server.", machineInstance.ID())
} else if machineInstance.ParentID() == "" {
//if parentId is not null, machineInstance is a
SubStateMachine, do not report global transaction.
@@ -247,7 +250,7 @@ func (s *StateLogStore) RecordStateMachineFinished(ctx
context.Context, machineI
return nil
}
-func (s *StateLogStore) reportTransactionFinished(ctx context.Context,
machineInstance statelang.StateMachineInstance, context core.ProcessContext)
error {
+func (s *StateLogStore) reportTransactionFinished(ctx context.Context,
machineInstance statelang.StateMachineInstance, context
process_ctrl.ProcessContext) error {
var err error
defer func() {
s.ClearUp(context)
@@ -286,7 +289,7 @@ func (s *StateLogStore) reportTransactionFinished(ctx
context.Context, machineIn
return nil
}
-func (s *StateLogStore) getGlobalTransaction(machineInstance
statelang.StateMachineInstance, context core.ProcessContext)
(*tm.GlobalTransaction, error) {
+func (s *StateLogStore) getGlobalTransaction(machineInstance
statelang.StateMachineInstance, context process_ctrl.ProcessContext)
(*tm.GlobalTransaction, error) {
globalTransaction, ok :=
context.GetVariable(constant.VarNameGlobalTx).(*tm.GlobalTransaction)
if ok {
return globalTransaction, nil
@@ -310,7 +313,7 @@ func (s *StateLogStore)
getGlobalTransaction(machineInstance statelang.StateMach
}
func (s *StateLogStore) RecordStateMachineRestarted(ctx context.Context,
machineInstance statelang.StateMachineInstance,
- context core.ProcessContext) error {
+ context process_ctrl.ProcessContext) error {
if machineInstance == nil {
return nil
}
@@ -329,7 +332,7 @@ func (s *StateLogStore) RecordStateMachineRestarted(ctx
context.Context, machine
}
func (s *StateLogStore) RecordStateStarted(ctx context.Context, stateInstance
statelang.StateInstance,
- context core.ProcessContext) error {
+ context process_ctrl.ProcessContext) error {
if stateInstance == nil {
return nil
}
@@ -381,13 +384,13 @@ func (s *StateLogStore) RecordStateStarted(ctx
context.Context, stateInstance st
return nil
}
-func (s *StateLogStore) isUpdateMode(stateInstance statelang.StateInstance,
context core.ProcessContext) (bool, error) {
- cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(core.DefaultStateMachineConfig)
+func (s *StateLogStore) isUpdateMode(stateInstance statelang.StateInstance,
context process_ctrl.ProcessContext) (bool, error) {
+ cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(config.DefaultStateMachineConfig)
if !ok {
return false, errors.New("stateMachineConfig is required in
context")
}
- instruction, ok := context.GetInstruction().(*core.StateInstruction)
+ instruction, ok := context.GetInstruction().(*pcext.StateInstruction)
if !ok {
return false, errors.New("stateInstruction is required in
processContext")
}
@@ -458,8 +461,8 @@ func (s *StateLogStore)
generateCompensateStateInstanceId(stateInstance statelan
return fmt.Sprintf("%s-%d", originalCompensateStateInstId, maxIndex)
}
-func (s *StateLogStore) branchRegister(stateInstance statelang.StateInstance,
context core.ProcessContext) error {
- cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(core.DefaultStateMachineConfig)
+func (s *StateLogStore) branchRegister(stateInstance statelang.StateInstance,
context process_ctrl.ProcessContext) error {
+ cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(config.DefaultStateMachineConfig)
if !ok {
return errors.New("stateMachineConfig is required in context")
}
@@ -518,7 +521,7 @@ func (s *StateLogStore) getIdIndex(stateInstanceId string,
separator string) int
}
func (s *StateLogStore) RecordStateFinished(ctx context.Context, stateInstance
statelang.StateInstance,
- context core.ProcessContext) error {
+ context process_ctrl.ProcessContext) error {
if stateInstance == nil {
return nil
}
@@ -541,7 +544,7 @@ func (s *StateLogStore) RecordStateFinished(ctx
context.Context, stateInstance s
}
// A switch to skip branch report on branch success, in order to
optimize performance
- cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(core.DefaultStateMachineConfig)
+ cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(config.DefaultStateMachineConfig)
if !(ok && !cfg.IsRmReportSuccessEnable() && statelang.SU ==
stateInstance.Status()) {
err = s.branchReport(stateInstance, context)
return err
@@ -551,8 +554,8 @@ func (s *StateLogStore) RecordStateFinished(ctx
context.Context, stateInstance s
}
-func (s *StateLogStore) branchReport(stateInstance statelang.StateInstance,
context core.ProcessContext) error {
- cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(core.DefaultStateMachineConfig)
+func (s *StateLogStore) branchReport(stateInstance statelang.StateInstance,
context process_ctrl.ProcessContext) error {
+ cfg, ok :=
context.GetVariable(constant.VarNameStateMachineConfig).(config.DefaultStateMachineConfig)
if ok && !cfg.IsSagaBranchRegisterEnable() {
log.Debugf("sagaBranchRegisterEnable = false, skip branch
report. state[%s]", stateInstance.Name())
return nil
@@ -843,7 +846,7 @@ func (s *StateLogStore) SetSeqGenerator(seqGenerator
sequence.SeqGenerator) {
s.seqGenerator = seqGenerator
}
-func (s *StateLogStore) ClearUp(context core.ProcessContext) {
+func (s *StateLogStore) ClearUp(context process_ctrl.ProcessContext) {
context.RemoveVariable(constant2.XidKey)
context.RemoveVariable(constant2.BranchTypeKey)
}
diff --git a/pkg/saga/statemachine/store/db/statelog_test.go
b/pkg/saga/statemachine/store/db/statelog_test.go
index 1445f2e9..08e9e445 100644
--- a/pkg/saga/statemachine/store/db/statelog_test.go
+++ b/pkg/saga/statemachine/store/db/statelog_test.go
@@ -22,7 +22,11 @@ import (
"fmt"
"github.com/pkg/errors"
"github.com/seata/seata-go/pkg/saga/statemachine/constant"
- "github.com/seata/seata-go/pkg/saga/statemachine/engine/core"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/config"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/pcext"
+ "github.com/seata/seata-go/pkg/saga/statemachine/engine/utils"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
"github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl/process"
"github.com/seata/seata-go/pkg/saga/statemachine/statelang"
"github.com/stretchr/testify/assert"
@@ -30,11 +34,11 @@ import (
"time"
)
-func mockProcessContext(stateMachineName string, stateMachineInstance
statelang.StateMachineInstance) core.ProcessContext {
- ctx := core.NewProcessContextBuilder().
+func mockProcessContext(stateMachineName string, stateMachineInstance
statelang.StateMachineInstance) process_ctrl.ProcessContext {
+ ctx := utils.NewProcessContextBuilder().
WithProcessType(process.StateLang).
WithOperationName(constant.OperationNameStart).
- WithInstruction(core.NewStateInstruction(stateMachineName,
"000001")).
+ WithInstruction(pcext.NewStateInstruction(stateMachineName,
"000001")).
WithStateMachineInstance(stateMachineInstance).
Build()
return ctx
@@ -57,8 +61,8 @@ func mockMachineInstance(stateMachineName string)
statelang.StateMachineInstance
return inst
}
-func mockStateMachineConfig(context core.ProcessContext)
core.StateMachineConfig {
- cfg := core.NewDefaultStateMachineConfig()
+func mockStateMachineConfig(context process_ctrl.ProcessContext)
engine.StateMachineConfig {
+ cfg := config.NewDefaultStateMachineConfig()
context.SetVariable(constant.VarNameStateMachineConfig, cfg)
return cfg
}
diff --git a/pkg/saga/statemachine/store/store.go
b/pkg/saga/statemachine/store/store.go
new file mode 100644
index 00000000..33ee8e54
--- /dev/null
+++ b/pkg/saga/statemachine/store/store.go
@@ -0,0 +1,39 @@
+package store
+
+import (
+ "context"
+ "github.com/seata/seata-go/pkg/saga/statemachine/process_ctrl"
+ "github.com/seata/seata-go/pkg/saga/statemachine/statelang"
+)
+
+type StateLogStore interface {
+ RecordStateMachineStarted(ctx context.Context, machineInstance
statelang.StateMachineInstance, context process_ctrl.ProcessContext) error
+
+ RecordStateMachineFinished(ctx context.Context, machineInstance
statelang.StateMachineInstance, context process_ctrl.ProcessContext) error
+
+ RecordStateMachineRestarted(ctx context.Context, machineInstance
statelang.StateMachineInstance, context process_ctrl.ProcessContext) error
+
+ RecordStateStarted(ctx context.Context, stateInstance
statelang.StateInstance, context process_ctrl.ProcessContext) error
+
+ RecordStateFinished(ctx context.Context, stateInstance
statelang.StateInstance, context process_ctrl.ProcessContext) error
+
+ GetStateMachineInstance(stateMachineInstanceId string)
(statelang.StateMachineInstance, error)
+
+ GetStateMachineInstanceByBusinessKey(businessKey string, tenantId
string) (statelang.StateMachineInstance, error)
+
+ GetStateMachineInstanceByParentId(parentId string)
([]statelang.StateMachineInstance, error)
+
+ GetStateInstance(stateInstanceId string, stateMachineInstanceId string)
(statelang.StateInstance, error)
+
+ GetStateInstanceListByMachineInstanceId(stateMachineInstanceId string)
([]statelang.StateInstance, error)
+
+ ClearUp(context process_ctrl.ProcessContext)
+}
+
+type StateLangStore interface {
+ GetStateMachineById(stateMachineId string) (statelang.StateMachine,
error)
+
+ GetLastVersionStateMachine(stateMachineName string, tenantId string)
(statelang.StateMachine, error)
+
+ StoreStateMachine(stateMachine statelang.StateMachine) error
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
