mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Added standalone activity Chasm tasks. Added handling of start activity and e2e implementation of standalone activity start execution with existing services. Updated protos related to standalone activities. The Chasm tasks are needed to kick off standalone activity execution via the existing services. Proto changes needed to so that the component ref can be passed and handled via service stack. - [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: Roey Berman <[email protected]> Co-authored-by: Dan Davison <[email protected]>
1411 lines
49 KiB
Go
1411 lines
49 KiB
Go
// Code generated by cmd/tools/genrpcwrappers. DO NOT EDIT.
|
|
|
|
package history
|
|
|
|
import (
|
|
"context"
|
|
|
|
"go.temporal.io/api/serviceerror"
|
|
"go.temporal.io/server/api/historyservice/v1"
|
|
"google.golang.org/grpc"
|
|
)
|
|
|
|
func (c *clientImpl) AddTasks(
|
|
ctx context.Context,
|
|
request *historyservice.AddTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.AddTasksResponse, error) {
|
|
shardID := request.GetShardId()
|
|
var response *historyservice.AddTasksResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.AddTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) CloseShard(
|
|
ctx context.Context,
|
|
request *historyservice.CloseShardRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.CloseShardResponse, error) {
|
|
shardID := request.GetShardId()
|
|
var response *historyservice.CloseShardResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.CloseShard(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) CompleteNexusOperation(
|
|
ctx context.Context,
|
|
request *historyservice.CompleteNexusOperationRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.CompleteNexusOperationResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetCompletion().GetNamespaceId(), request.GetCompletion().GetWorkflowId())
|
|
var response *historyservice.CompleteNexusOperationResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.CompleteNexusOperation(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) CompleteNexusOperationChasm(
|
|
ctx context.Context,
|
|
request *historyservice.CompleteNexusOperationChasmRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.CompleteNexusOperationChasmResponse, error) {
|
|
ref, err := c.tokenSerializer.DeserializeChasmComponentRef(request.GetCompletion().GetComponentRef())
|
|
if err != nil {
|
|
return nil, serviceerror.NewInvalidArgument("error deserializing component ref")
|
|
}
|
|
shardID := c.shardIDFromWorkflowID(ref.GetNamespaceId(), ref.GetBusinessId())
|
|
|
|
var response *historyservice.CompleteNexusOperationChasmResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.CompleteNexusOperationChasm(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) DeleteDLQTasks(
|
|
ctx context.Context,
|
|
request *historyservice.DeleteDLQTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.DeleteDLQTasksResponse, error) {
|
|
shardID := c.getRandomShard()
|
|
var response *historyservice.DeleteDLQTasksResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.DeleteDLQTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) DeleteWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.DeleteWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.DeleteWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.DeleteWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.DeleteWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) DeleteWorkflowVisibilityRecord(
|
|
ctx context.Context,
|
|
request *historyservice.DeleteWorkflowVisibilityRecordRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.DeleteWorkflowVisibilityRecordResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.DeleteWorkflowVisibilityRecordResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.DeleteWorkflowVisibilityRecord(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) DescribeMutableState(
|
|
ctx context.Context,
|
|
request *historyservice.DescribeMutableStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.DescribeMutableStateResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.DescribeMutableStateResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.DescribeMutableState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) DescribeWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.DescribeWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.DescribeWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.DescribeWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.DescribeWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ExecuteMultiOperation(
|
|
ctx context.Context,
|
|
request *historyservice.ExecuteMultiOperationRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ExecuteMultiOperationResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowId())
|
|
var response *historyservice.ExecuteMultiOperationResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ExecuteMultiOperation(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ForceDeleteWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.ForceDeleteWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ForceDeleteWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.ForceDeleteWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ForceDeleteWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GenerateLastHistoryReplicationTasks(
|
|
ctx context.Context,
|
|
request *historyservice.GenerateLastHistoryReplicationTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GenerateLastHistoryReplicationTasksResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.GenerateLastHistoryReplicationTasksResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GenerateLastHistoryReplicationTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GetDLQMessages(
|
|
ctx context.Context,
|
|
request *historyservice.GetDLQMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GetDLQMessagesResponse, error) {
|
|
shardID := request.GetShardId()
|
|
var response *historyservice.GetDLQMessagesResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GetDLQMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GetDLQReplicationMessages(
|
|
ctx context.Context,
|
|
request *historyservice.GetDLQReplicationMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GetDLQReplicationMessagesResponse, error) {
|
|
// All workflow IDs are in the same shard per request
|
|
if len(request.GetTaskInfos()) == 0 {
|
|
return nil, serviceerror.NewInvalidArgument("missing TaskInfos")
|
|
}
|
|
shardID := c.shardIDFromWorkflowID(request.GetTaskInfos()[0].NamespaceId, request.GetTaskInfos()[0].WorkflowId)
|
|
var response *historyservice.GetDLQReplicationMessagesResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GetDLQReplicationMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GetDLQTasks(
|
|
ctx context.Context,
|
|
request *historyservice.GetDLQTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GetDLQTasksResponse, error) {
|
|
shardID := c.getRandomShard()
|
|
var response *historyservice.GetDLQTasksResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GetDLQTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GetMutableState(
|
|
ctx context.Context,
|
|
request *historyservice.GetMutableStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GetMutableStateResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.GetMutableStateResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GetMutableState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GetShard(
|
|
ctx context.Context,
|
|
request *historyservice.GetShardRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GetShardResponse, error) {
|
|
shardID := request.GetShardId()
|
|
var response *historyservice.GetShardResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GetShard(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GetWorkflowExecutionHistory(
|
|
ctx context.Context,
|
|
request *historyservice.GetWorkflowExecutionHistoryRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GetWorkflowExecutionHistoryResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.GetWorkflowExecutionHistoryResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GetWorkflowExecutionHistory(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GetWorkflowExecutionHistoryReverse(
|
|
ctx context.Context,
|
|
request *historyservice.GetWorkflowExecutionHistoryReverseRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GetWorkflowExecutionHistoryReverseResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.GetWorkflowExecutionHistoryReverseResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GetWorkflowExecutionHistoryReverse(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GetWorkflowExecutionRawHistory(
|
|
ctx context.Context,
|
|
request *historyservice.GetWorkflowExecutionRawHistoryRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GetWorkflowExecutionRawHistoryResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.GetWorkflowExecutionRawHistoryResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GetWorkflowExecutionRawHistory(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) GetWorkflowExecutionRawHistoryV2(
|
|
ctx context.Context,
|
|
request *historyservice.GetWorkflowExecutionRawHistoryV2Request,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.GetWorkflowExecutionRawHistoryV2Response, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.GetWorkflowExecutionRawHistoryV2Response
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.GetWorkflowExecutionRawHistoryV2(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ImportWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.ImportWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ImportWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.ImportWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ImportWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) InvokeStateMachineMethod(
|
|
ctx context.Context,
|
|
request *historyservice.InvokeStateMachineMethodRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.InvokeStateMachineMethodResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowId())
|
|
var response *historyservice.InvokeStateMachineMethodResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.InvokeStateMachineMethod(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) IsActivityTaskValid(
|
|
ctx context.Context,
|
|
request *historyservice.IsActivityTaskValidRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.IsActivityTaskValidResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.IsActivityTaskValidResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.IsActivityTaskValid(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) IsWorkflowTaskValid(
|
|
ctx context.Context,
|
|
request *historyservice.IsWorkflowTaskValidRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.IsWorkflowTaskValidResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.IsWorkflowTaskValidResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.IsWorkflowTaskValid(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ListQueues(
|
|
ctx context.Context,
|
|
request *historyservice.ListQueuesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ListQueuesResponse, error) {
|
|
shardID := c.getRandomShard()
|
|
var response *historyservice.ListQueuesResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ListQueues(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ListTasks(
|
|
ctx context.Context,
|
|
request *historyservice.ListTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ListTasksResponse, error) {
|
|
shardID := request.GetRequest().GetShardId()
|
|
var response *historyservice.ListTasksResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ListTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) MergeDLQMessages(
|
|
ctx context.Context,
|
|
request *historyservice.MergeDLQMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.MergeDLQMessagesResponse, error) {
|
|
shardID := request.GetShardId()
|
|
var response *historyservice.MergeDLQMessagesResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.MergeDLQMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) PauseActivity(
|
|
ctx context.Context,
|
|
request *historyservice.PauseActivityRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.PauseActivityResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetFrontendRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.PauseActivityResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.PauseActivity(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) PauseWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.PauseWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.PauseWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetPauseRequest().GetWorkflowId())
|
|
var response *historyservice.PauseWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.PauseWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) PollMutableState(
|
|
ctx context.Context,
|
|
request *historyservice.PollMutableStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.PollMutableStateResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.PollMutableStateResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.PollMutableState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) PollWorkflowExecutionUpdate(
|
|
ctx context.Context,
|
|
request *historyservice.PollWorkflowExecutionUpdateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.PollWorkflowExecutionUpdateResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetUpdateRef().GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.PollWorkflowExecutionUpdateResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.PollWorkflowExecutionUpdate(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) PurgeDLQMessages(
|
|
ctx context.Context,
|
|
request *historyservice.PurgeDLQMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.PurgeDLQMessagesResponse, error) {
|
|
shardID := request.GetShardId()
|
|
var response *historyservice.PurgeDLQMessagesResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.PurgeDLQMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) QueryWorkflow(
|
|
ctx context.Context,
|
|
request *historyservice.QueryWorkflowRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.QueryWorkflowResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.QueryWorkflowResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.QueryWorkflow(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ReapplyEvents(
|
|
ctx context.Context,
|
|
request *historyservice.ReapplyEventsRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ReapplyEventsResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.ReapplyEventsResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ReapplyEvents(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RebuildMutableState(
|
|
ctx context.Context,
|
|
request *historyservice.RebuildMutableStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RebuildMutableStateResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.RebuildMutableStateResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RebuildMutableState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RecordActivityTaskHeartbeat(
|
|
ctx context.Context,
|
|
request *historyservice.RecordActivityTaskHeartbeatRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RecordActivityTaskHeartbeatResponse, error) {
|
|
taskToken, err := c.tokenSerializer.Deserialize(request.GetHeartbeatRequest().GetTaskToken())
|
|
if err != nil {
|
|
return nil, serviceerror.NewInvalidArgument("error deserializing task token")
|
|
}
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), taskToken.GetWorkflowId())
|
|
|
|
var response *historyservice.RecordActivityTaskHeartbeatResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RecordActivityTaskHeartbeat(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RecordChildExecutionCompleted(
|
|
ctx context.Context,
|
|
request *historyservice.RecordChildExecutionCompletedRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RecordChildExecutionCompletedResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetParentExecution().GetWorkflowId())
|
|
var response *historyservice.RecordChildExecutionCompletedResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RecordChildExecutionCompleted(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RecordWorkflowTaskStarted(
|
|
ctx context.Context,
|
|
request *historyservice.RecordWorkflowTaskStartedRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RecordWorkflowTaskStartedResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.RecordWorkflowTaskStartedResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RecordWorkflowTaskStarted(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RefreshWorkflowTasks(
|
|
ctx context.Context,
|
|
request *historyservice.RefreshWorkflowTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RefreshWorkflowTasksResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.RefreshWorkflowTasksResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RefreshWorkflowTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RemoveSignalMutableState(
|
|
ctx context.Context,
|
|
request *historyservice.RemoveSignalMutableStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RemoveSignalMutableStateResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.RemoveSignalMutableStateResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RemoveSignalMutableState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RemoveTask(
|
|
ctx context.Context,
|
|
request *historyservice.RemoveTaskRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RemoveTaskResponse, error) {
|
|
shardID := request.GetShardId()
|
|
var response *historyservice.RemoveTaskResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RemoveTask(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ReplicateEventsV2(
|
|
ctx context.Context,
|
|
request *historyservice.ReplicateEventsV2Request,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ReplicateEventsV2Response, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.ReplicateEventsV2Response
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ReplicateEventsV2(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ReplicateWorkflowState(
|
|
ctx context.Context,
|
|
request *historyservice.ReplicateWorkflowStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ReplicateWorkflowStateResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowState().GetExecutionInfo().GetWorkflowId())
|
|
var response *historyservice.ReplicateWorkflowStateResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ReplicateWorkflowState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RequestCancelWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.RequestCancelWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RequestCancelWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetCancelRequest().GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.RequestCancelWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RequestCancelWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ResetActivity(
|
|
ctx context.Context,
|
|
request *historyservice.ResetActivityRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ResetActivityResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetFrontendRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.ResetActivityResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ResetActivity(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ResetStickyTaskQueue(
|
|
ctx context.Context,
|
|
request *historyservice.ResetStickyTaskQueueRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ResetStickyTaskQueueResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.ResetStickyTaskQueueResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ResetStickyTaskQueue(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ResetWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.ResetWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ResetWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetResetRequest().GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.ResetWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ResetWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RespondActivityTaskCanceled(
|
|
ctx context.Context,
|
|
request *historyservice.RespondActivityTaskCanceledRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RespondActivityTaskCanceledResponse, error) {
|
|
taskToken, err := c.tokenSerializer.Deserialize(request.GetCancelRequest().GetTaskToken())
|
|
if err != nil {
|
|
return nil, serviceerror.NewInvalidArgument("error deserializing task token")
|
|
}
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), taskToken.GetWorkflowId())
|
|
|
|
var response *historyservice.RespondActivityTaskCanceledResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RespondActivityTaskCanceled(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RespondActivityTaskCompleted(
|
|
ctx context.Context,
|
|
request *historyservice.RespondActivityTaskCompletedRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RespondActivityTaskCompletedResponse, error) {
|
|
taskToken, err := c.tokenSerializer.Deserialize(request.GetCompleteRequest().GetTaskToken())
|
|
if err != nil {
|
|
return nil, serviceerror.NewInvalidArgument("error deserializing task token")
|
|
}
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), taskToken.GetWorkflowId())
|
|
|
|
var response *historyservice.RespondActivityTaskCompletedResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RespondActivityTaskCompleted(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RespondActivityTaskFailed(
|
|
ctx context.Context,
|
|
request *historyservice.RespondActivityTaskFailedRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RespondActivityTaskFailedResponse, error) {
|
|
taskToken, err := c.tokenSerializer.Deserialize(request.GetFailedRequest().GetTaskToken())
|
|
if err != nil {
|
|
return nil, serviceerror.NewInvalidArgument("error deserializing task token")
|
|
}
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), taskToken.GetWorkflowId())
|
|
|
|
var response *historyservice.RespondActivityTaskFailedResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RespondActivityTaskFailed(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RespondWorkflowTaskCompleted(
|
|
ctx context.Context,
|
|
request *historyservice.RespondWorkflowTaskCompletedRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RespondWorkflowTaskCompletedResponse, error) {
|
|
taskToken, err := c.tokenSerializer.Deserialize(request.GetCompleteRequest().GetTaskToken())
|
|
if err != nil {
|
|
return nil, serviceerror.NewInvalidArgument("error deserializing task token")
|
|
}
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), taskToken.GetWorkflowId())
|
|
|
|
var response *historyservice.RespondWorkflowTaskCompletedResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RespondWorkflowTaskCompleted(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) RespondWorkflowTaskFailed(
|
|
ctx context.Context,
|
|
request *historyservice.RespondWorkflowTaskFailedRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.RespondWorkflowTaskFailedResponse, error) {
|
|
taskToken, err := c.tokenSerializer.Deserialize(request.GetFailedRequest().GetTaskToken())
|
|
if err != nil {
|
|
return nil, serviceerror.NewInvalidArgument("error deserializing task token")
|
|
}
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), taskToken.GetWorkflowId())
|
|
|
|
var response *historyservice.RespondWorkflowTaskFailedResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.RespondWorkflowTaskFailed(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) ScheduleWorkflowTask(
|
|
ctx context.Context,
|
|
request *historyservice.ScheduleWorkflowTaskRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.ScheduleWorkflowTaskResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.ScheduleWorkflowTaskResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.ScheduleWorkflowTask(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) SignalWithStartWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.SignalWithStartWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.SignalWithStartWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetSignalWithStartRequest().GetWorkflowId())
|
|
var response *historyservice.SignalWithStartWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.SignalWithStartWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) SignalWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.SignalWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.SignalWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetSignalRequest().GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.SignalWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.SignalWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) StartWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.StartWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.StartWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetStartRequest().GetWorkflowId())
|
|
var response *historyservice.StartWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.StartWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) SyncActivity(
|
|
ctx context.Context,
|
|
request *historyservice.SyncActivityRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.SyncActivityResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowId())
|
|
var response *historyservice.SyncActivityResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.SyncActivity(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) SyncShardStatus(
|
|
ctx context.Context,
|
|
request *historyservice.SyncShardStatusRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.SyncShardStatusResponse, error) {
|
|
shardID := request.GetShardId()
|
|
var response *historyservice.SyncShardStatusResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.SyncShardStatus(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) SyncWorkflowState(
|
|
ctx context.Context,
|
|
request *historyservice.SyncWorkflowStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.SyncWorkflowStateResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetExecution().GetWorkflowId())
|
|
var response *historyservice.SyncWorkflowStateResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.SyncWorkflowState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) TerminateWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.TerminateWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.TerminateWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetTerminateRequest().GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.TerminateWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.TerminateWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) UnpauseActivity(
|
|
ctx context.Context,
|
|
request *historyservice.UnpauseActivityRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.UnpauseActivityResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetFrontendRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.UnpauseActivityResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.UnpauseActivity(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) UnpauseWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.UnpauseWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.UnpauseWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetUnpauseRequest().GetWorkflowId())
|
|
var response *historyservice.UnpauseWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.UnpauseWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) UpdateActivityOptions(
|
|
ctx context.Context,
|
|
request *historyservice.UpdateActivityOptionsRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.UpdateActivityOptionsResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetUpdateRequest().GetExecution().GetWorkflowId())
|
|
var response *historyservice.UpdateActivityOptionsResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.UpdateActivityOptions(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) UpdateWorkflowExecution(
|
|
ctx context.Context,
|
|
request *historyservice.UpdateWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.UpdateWorkflowExecutionResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetRequest().GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.UpdateWorkflowExecutionResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.UpdateWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) UpdateWorkflowExecutionOptions(
|
|
ctx context.Context,
|
|
request *historyservice.UpdateWorkflowExecutionOptionsRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.UpdateWorkflowExecutionOptionsResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetUpdateRequest().GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.UpdateWorkflowExecutionOptionsResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.UpdateWorkflowExecutionOptions(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) VerifyChildExecutionCompletionRecorded(
|
|
ctx context.Context,
|
|
request *historyservice.VerifyChildExecutionCompletionRecordedRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.VerifyChildExecutionCompletionRecordedResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetParentExecution().GetWorkflowId())
|
|
var response *historyservice.VerifyChildExecutionCompletionRecordedResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.VerifyChildExecutionCompletionRecorded(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (c *clientImpl) VerifyFirstWorkflowTaskScheduled(
|
|
ctx context.Context,
|
|
request *historyservice.VerifyFirstWorkflowTaskScheduledRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*historyservice.VerifyFirstWorkflowTaskScheduledResponse, error) {
|
|
shardID := c.shardIDFromWorkflowID(request.GetNamespaceId(), request.GetWorkflowExecution().GetWorkflowId())
|
|
var response *historyservice.VerifyFirstWorkflowTaskScheduledResponse
|
|
op := func(ctx context.Context, client historyservice.HistoryServiceClient) error {
|
|
var err error
|
|
ctx, cancel := c.createContext(ctx)
|
|
defer cancel()
|
|
response, err = client.VerifyFirstWorkflowTaskScheduled(ctx, request, opts...)
|
|
return err
|
|
}
|
|
if err := c.executeWithRedirect(ctx, shardID, op); err != nil {
|
|
return nil, err
|
|
}
|
|
return response, nil
|
|
}
|