61
ctx context.Context,
62
executable queues.Executable,
64
>
task := executable.GetTask()
65
>
taskType := queues.GetOutboundTaskTypeTagValue(task, false, e.shardContext.ChasmRegistry())
66
>
namespaceTag, _ := getNamespaceTagAndReplicationStateByID(
67
>
e.shardContext.GetNamespaceRegistry(),
68
>
task.GetNamespaceID(),
69
>
executable.GetWorkflowID(),
70
>
)
71
>
respond := func(err error) queues.ExecuteResponse {
72
>
metricsTags := []metrics.Tag{
73
>
namespaceTag,
74
>
metrics.TaskTypeTag(taskType),
75
>
metrics.OperationTag(taskType),
76
>
}
77
>
return queues.ExecuteResponse{
78
>
ExecutionMetricTags: metricsTags,
79
>
ExecutedAsActive: false,
80
>
ExecutionErr: err,
81
>
}
82
>
}
83
84
>
nsRecord, err := e.shardContext.GetNamespaceRegistry().GetNamespaceByID(
outbound_queue_standby_task_executor.go
85
>
namespace.ID(task.GetNamespaceID()),
86
>
)
87
>
if err != nil {
88
return respond(err)
89
}
90
92
// namespace is not replicated to local cluster, ignore corresponding tasks
93
return respond(nil)
94
}
95
96
>
if err := validateTaskByClock(e.shardContext, task); err != nil {
outbound_queue_standby_task_executor.go
97
return respond(err)
98
}