299
ctx chasm.Context,
300
namespace string,
301
>
) (*historyservice.RecordActivityTaskStartedResponse, error) {
activity.go
302
>
key := ctx.ExecutionKey()
303
>
lastHeartbeat, _ := a.LastHeartbeat.TryGet(ctx)
304
>
requestData := a.RequestData.Get(ctx)
305
>
attempt := a.LastAttempt.Get(ctx)
306
>
links := ctx.Links(a)
307
>
308
>
return &historyservice.RecordActivityTaskStartedResponse{
309
>
StartedTime: attempt.GetStartedTime(),
310
>
Attempt: attempt.GetCount(),
311
>
Priority: a.GetPriority(),
312
>
RetryPolicy: a.GetRetryPolicy(),
313
>
ActivityRunId: key.RunID,
314
>
WorkflowNamespace: namespace,
315
>
HeartbeatDetails: lastHeartbeat.GetDetails(),
316
>
CurrentAttemptScheduledTime: a.dispatchTimeForAttempt(attempt),
317
>
ScheduledEvent: &historypb.HistoryEvent{
318
>
EventType: enumspb.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED,
319
>
EventTime: a.GetScheduleTime(),
320
>
Attributes: &historypb.HistoryEvent_ActivityTaskScheduledEventAttributes{
321
>
ActivityTaskScheduledEventAttributes: &historypb.ActivityTaskScheduledEventAttributes{
322
>
ActivityId: key.BusinessID,
323
>
ActivityType: a.GetActivityType(),
324
>
Input: requestData.GetInput(),
325
>
Header: requestData.GetHeader(),
326
>
TaskQueue: a.GetTaskQueue(),
327
>
ScheduleToCloseTimeout: a.GetScheduleToCloseTimeout(),
328
>
ScheduleToStartTimeout: a.GetScheduleToStartTimeout(),
329
>
StartToCloseTimeout: a.GetStartToCloseTimeout(),
330
>
HeartbeatTimeout: a.GetHeartbeatTimeout(),
331
>
},
332
>
},
333
>
Links: links,
334
>
},
335
>
}, nil
336
>
}
337
338
// dispatchTimeForAttempt returns the dispatch time of the given attempt.
339
>
func (a *Activity) dispatchTimeForAttempt(attempt *activitypb.ActivityAttemptState) *timestamppb.Timestamp {
activity.go
340
>
if dispatchTime := attempt.GetDispatchTime(); dispatchTime != nil {
341
return dispatchTime
342
}
344
>
return timestamppb.New(a.firstDispatchTime())
345
>
}
346
return dispatchTimeForRetry(attempt)
347
}