mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
cb63a7a3d0
## What changed? Added a return flag on the NewExecution function to indicate if a new execution was actually created. Also added docs to NewExecution. Changed standalone activity on start to pass back started flag if new exec was actually created. ## Why? On NewExecution today if we use BusinessIDConflictPolicyUseExisting, we need a way to distinguish whether something is newly created or not. Based on that flag, caller code can react differently. For standalone activity, we must pass back the started flag according to whether a new exec was actually started, to matching existing workflow behavior. ## How did you test it? - [X] built - [X] run locally and tested manually - [X] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) --------- Co-authored-by: Yichao Yang <[email protected]>
208 lines
4.2 KiB
Go
208 lines
4.2 KiB
Go
package tests
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
"go.temporal.io/server/chasm"
|
|
"go.temporal.io/server/chasm/lib/tests/gen/testspb/v1"
|
|
"go.temporal.io/server/common/namespace"
|
|
)
|
|
|
|
type (
|
|
NewPayloadStoreRequest struct {
|
|
NamespaceID namespace.ID
|
|
StoreID string
|
|
IDReusePolicy chasm.BusinessIDReusePolicy
|
|
IDConflictPolicy chasm.BusinessIDConflictPolicy
|
|
}
|
|
|
|
NewPayloadStoreResponse struct {
|
|
RunID string
|
|
}
|
|
|
|
DescribePayloadStoreRequest struct {
|
|
NamespaceID namespace.ID
|
|
StoreID string
|
|
}
|
|
|
|
DescribePayloadStoreResponse struct {
|
|
State *testspb.TestPayloadStore
|
|
}
|
|
|
|
ClosePayloadStoreRequest struct {
|
|
NamespaceID namespace.ID
|
|
StoreID string
|
|
}
|
|
|
|
ClosePayloadStoreResponse struct{}
|
|
|
|
AddPayloadRequest struct {
|
|
NamespaceID namespace.ID
|
|
StoreID string
|
|
PayloadKey string
|
|
Payload *commonpb.Payload
|
|
TTL time.Duration
|
|
}
|
|
|
|
AddPayloadResponse struct {
|
|
State *testspb.TestPayloadStore
|
|
}
|
|
|
|
GetPayloadRequest struct {
|
|
NamespaceID namespace.ID
|
|
StoreID string
|
|
PayloadKey string
|
|
}
|
|
|
|
GetPayloadResponse struct {
|
|
Payload *commonpb.Payload
|
|
}
|
|
|
|
RemovePayloadRequest struct {
|
|
NamespaceID namespace.ID
|
|
StoreID string
|
|
PayloadKey string
|
|
}
|
|
|
|
RemovePayloadResponse struct {
|
|
State *testspb.TestPayloadStore
|
|
}
|
|
)
|
|
|
|
func NewPayloadStoreHandler(
|
|
ctx context.Context,
|
|
request NewPayloadStoreRequest,
|
|
) (NewPayloadStoreResponse, error) {
|
|
result, err := chasm.NewExecution(
|
|
ctx,
|
|
chasm.ExecutionKey{
|
|
NamespaceID: request.NamespaceID.String(),
|
|
BusinessID: request.StoreID,
|
|
},
|
|
func(mutableContext chasm.MutableContext, _ any) (*PayloadStore, any, error) {
|
|
store, err := NewPayloadStore(mutableContext)
|
|
return store, nil, err
|
|
},
|
|
nil,
|
|
chasm.WithBusinessIDPolicy(request.IDReusePolicy, request.IDConflictPolicy),
|
|
)
|
|
if err != nil {
|
|
return NewPayloadStoreResponse{}, err
|
|
}
|
|
return NewPayloadStoreResponse{
|
|
RunID: result.ExecutionKey.RunID,
|
|
}, nil
|
|
}
|
|
|
|
func DescribePayloadStoreHandler(
|
|
ctx context.Context,
|
|
request DescribePayloadStoreRequest,
|
|
) (DescribePayloadStoreResponse, error) {
|
|
state, err := chasm.ReadComponent(
|
|
ctx,
|
|
chasm.NewComponentRef[*PayloadStore](
|
|
chasm.ExecutionKey{
|
|
NamespaceID: request.NamespaceID.String(),
|
|
BusinessID: request.StoreID,
|
|
},
|
|
),
|
|
(*PayloadStore).Describe,
|
|
request,
|
|
)
|
|
if err != nil {
|
|
return DescribePayloadStoreResponse{}, err
|
|
}
|
|
return DescribePayloadStoreResponse{
|
|
State: state,
|
|
}, nil
|
|
}
|
|
|
|
func ClosePayloadStoreHandler(
|
|
ctx context.Context,
|
|
request ClosePayloadStoreRequest,
|
|
) (ClosePayloadStoreResponse, error) {
|
|
resp, _, err := chasm.UpdateComponent(
|
|
ctx,
|
|
chasm.NewComponentRef[*PayloadStore](
|
|
chasm.ExecutionKey{
|
|
NamespaceID: request.NamespaceID.String(),
|
|
BusinessID: request.StoreID,
|
|
},
|
|
),
|
|
(*PayloadStore).Close,
|
|
request,
|
|
)
|
|
return resp, err
|
|
}
|
|
|
|
func AddPayloadHandler(
|
|
ctx context.Context,
|
|
request AddPayloadRequest,
|
|
) (AddPayloadResponse, error) {
|
|
state, _, err := chasm.UpdateComponent(
|
|
ctx,
|
|
chasm.NewComponentRef[*PayloadStore](
|
|
chasm.ExecutionKey{
|
|
NamespaceID: request.NamespaceID.String(),
|
|
BusinessID: request.StoreID,
|
|
},
|
|
),
|
|
(*PayloadStore).AddPayload,
|
|
request,
|
|
)
|
|
if err != nil {
|
|
return AddPayloadResponse{}, err
|
|
}
|
|
return AddPayloadResponse{
|
|
State: state,
|
|
}, nil
|
|
}
|
|
|
|
func GetPayloadHandler(
|
|
ctx context.Context,
|
|
request GetPayloadRequest,
|
|
) (GetPayloadResponse, error) {
|
|
payload, err := chasm.ReadComponent(
|
|
ctx,
|
|
chasm.NewComponentRef[*PayloadStore](
|
|
chasm.ExecutionKey{
|
|
NamespaceID: request.NamespaceID.String(),
|
|
BusinessID: request.StoreID,
|
|
},
|
|
),
|
|
(*PayloadStore).GetPayload,
|
|
request.PayloadKey,
|
|
)
|
|
if err != nil {
|
|
return GetPayloadResponse{}, err
|
|
}
|
|
return GetPayloadResponse{
|
|
Payload: payload,
|
|
}, nil
|
|
}
|
|
|
|
func RemovePayloadHandler(
|
|
ctx context.Context,
|
|
request RemovePayloadRequest,
|
|
) (RemovePayloadResponse, error) {
|
|
state, _, err := chasm.UpdateComponent(
|
|
ctx,
|
|
chasm.NewComponentRef[*PayloadStore](
|
|
chasm.ExecutionKey{
|
|
NamespaceID: request.NamespaceID.String(),
|
|
BusinessID: request.StoreID,
|
|
},
|
|
),
|
|
(*PayloadStore).RemovePayload,
|
|
request.PayloadKey,
|
|
)
|
|
if err != nil {
|
|
return RemovePayloadResponse{}, err
|
|
}
|
|
return RemovePayloadResponse{
|
|
State: state,
|
|
}, nil
|
|
}
|