func (ath *activityTaskHandlerImpl) Execute(taskList string, t *s.PollForActivityTaskResponse) (result interface{}, err error) {
	traceLog(func() {
		ath.logger.Debug("Processing new activity task",
			zap.String(tagWorkflowID, t.WorkflowExecution.GetWorkflowId()),
			zap.String(tagRunID, t.WorkflowExecution.GetRunId()),
			zap.String(tagActivityType, t.ActivityType.GetName()))
	})

	rootCtx := ath.userContext
	if rootCtx == nil {
		rootCtx = context.Background()
	}
	canCtx, cancel := context.WithCancel(rootCtx)
	defer cancel()

	invoker := newServiceInvoker(t.TaskToken, ath.identity, ath.service, cancel, t.GetHeartbeatTimeoutSeconds(), ath.workerStopCh)
	defer func() {
		_, activityCompleted := result.(*s.RespondActivityTaskCompletedRequest)
		invoker.Close(!activityCompleted) // flush buffered heartbeat if activity was not successfully completed.
	}()

	workflowType := t.WorkflowType.GetName()
	activityType := t.ActivityType.GetName()
	metricsScope := getMetricsScopeForActivity(ath.metricsScope, workflowType, activityType)
	ctx := WithActivityTask(canCtx, t, taskList, invoker, ath.logger, metricsScope, ath.dataConverter, ath.workerStopCh, ath.contextPropagators, ath.tracer)

	activityImplementation := ath.getActivity(activityType)
	if activityImplementation == nil {
		// Couldn't find the activity implementation.
		supported := strings.Join(ath.getRegisteredActivityNames(), ", ")
		return nil, fmt.Errorf("unable to find activityType=%v. Supported types: [%v]", activityType, supported)
	}

	// panic handler
	defer func() {
		if p := recover(); p != nil {
			topLine := fmt.Sprintf("activity for %s [panic]:", ath.taskListName)
			st := getStackTraceRaw(topLine, 7, 0)
			ath.logger.Error("Activity panic.",
				zap.String(tagWorkflowID, t.WorkflowExecution.GetWorkflowId()),
				zap.String(tagRunID, t.WorkflowExecution.GetRunId()),
				zap.String(tagActivityType, activityType),
				zap.String("PanicError", fmt.Sprintf("%v", p)),
				zap.String("PanicStack", st))
			metricsScope.Counter(metrics.ActivityTaskPanicCounter).Inc(1)
			panicErr := newPanicError(p, st)
			result, err = convertActivityResultToRespondRequest(ath.identity, t.TaskToken, nil, panicErr, ath.dataConverter), nil
		}
	}()

	// propagate context information into the activity context from the headers
	for _, ctxProp := range ath.contextPropagators {
		var err error
		if ctx, err = ctxProp.Extract(ctx, NewHeaderReader(t.Header)); err != nil {
			return nil, fmt.Errorf("unable to propagate context %v", err)
		}
	}

	info := ctx.Value(activityEnvContextKey).(*activityEnvironment)
	ctx, dlCancelFunc := context.WithDeadline(ctx, info.deadline)
	defer dlCancelFunc()

	ctx, span := createOpenTracingActivitySpan(ctx, ath.tracer, time.Now(), activityType, t.WorkflowExecution.GetWorkflowId(), t.WorkflowExecution.GetRunId())
	defer span.Finish()
	output, err := activityImplementation.Execute(ctx, t.Input)

	dlCancelFunc()
	if <-ctx.Done(); ctx.Err() == context.DeadlineExceeded {
		return nil, ctx.Err()
	}
	if err != nil && err != ErrActivityResultPending {
		ath.logger.Error("Activity error.",
			zap.String(tagWorkflowID, t.WorkflowExecution.GetWorkflowId()),
			zap.String(tagRunID, t.WorkflowExecution.GetRunId()),
			zap.String(tagActivityType, activityType),
			zap.Error(err),
		)
	}
	return convertActivityResultToRespondRequest(ath.identity, t.TaskToken, output, err, ath.dataConverter), nil
}
// Execute executes an implementation of the activity.
Execute executes an implementation of the activity.
// Execute executes an implementation of the activity .
func (ath *activityTaskHandlerImpl) Execute(taskList string, t *s.PollForActivityTaskResponse) (result interface{}, err error) {
	traceLog(func() {
		ath.logger.Debug("Processing new activity task",
			zap.String(tagWorkflowID, t.WorkflowExecution.GetWorkflowId()),
			zap.String(tagRunID, t.WorkflowExecution.GetRunId()),
			zap.String(tagActivityType, t.ActivityType.GetName()))
	})
	rootCtx := ath.userContext
	if rootCtx == nil {
		rootCtx = context.Background()
	}
	canCtx, cancel := context.WithCancel(rootCtx)
	defer cancel()
	invoker := newServiceInvoker(t.TaskToken, ath.identity, ath.service, cancel, t.GetHeartbeatTimeoutSeconds(), ath.workerStopCh)
	defer func() {
		_, activityCompleted := result.(*s.RespondActivityTaskCompletedRequest)
		invoker.Close(!activityCompleted) 
	}()
	workflowType := t.WorkflowType.GetName()
	activityType := t.ActivityType.GetName()
	metricsScope := getMetricsScopeForActivity(ath.metricsScope, workflowType, activityType)
	ctx := WithActivityTask(canCtx, t, taskList, invoker, ath.logger, metricsScope, ath.dataConverter, ath.workerStopCh, ath.contextPropagators, ath.tracer)
	activityImplementation := ath.getActivity(activityType)
	if activityImplementation == nil {
		supported := strings.Join(ath.getRegisteredActivityNames(), ", ")
		return nil, fmt.Errorf("unable to find activityType=%v. Supported types: [%v]", activityType, supported)
	}
	defer func() {
		if p := recover(); p != nil {
			topLine := fmt.Sprintf("activity for %s [panic]:", ath.taskListName)
			st := getStackTraceRaw(topLine, 7, 0)
			ath.logger.Error("Activity panic.",
				zap.String(tagWorkflowID, t.WorkflowExecution.GetWorkflowId()),
				zap.String(tagRunID, t.WorkflowExecution.GetRunId()),
				zap.String(tagActivityType, activityType),
				zap.String("PanicError", fmt.Sprintf("%v", p)),
				zap.String("PanicStack", st))
			metricsScope.Counter(metrics.ActivityTaskPanicCounter).Inc(1)
			panicErr := newPanicError(p, st)
			result, err = convertActivityResultToRespondRequest(ath.identity, t.TaskToken, nil, panicErr, ath.dataConverter), nil
		}
	}()
	for _, ctxProp := range ath.contextPropagators {
		var err error
		if ctx, err = ctxProp.Extract(ctx, NewHeaderReader(t.Header)); err != nil {
			return nil, fmt.Errorf("unable to propagate context %v", err)
		}
	}
	info := ctx.Value(activityEnvContextKey).(*activityEnvironment)
	ctx, dlCancelFunc := context.WithDeadline(ctx, info.deadline)
	defer dlCancelFunc()
	ctx, span := createOpenTracingActivitySpan(ctx, ath.tracer, time.Now(), activityType, t.WorkflowExecution.GetWorkflowId(), t.WorkflowExecution.GetRunId())
	defer span.Finish()
	output, err := activityImplementation.Execute(ctx, t.Input)
	dlCancelFunc()
	if <-ctx.Done(); ctx.Err() == context.DeadlineExceeded {
		return nil, ctx.Err()
	}
	if err != nil && err != ErrActivityResultPending {
		ath.logger.Error("Activity error.",
			zap.String(tagWorkflowID, t.WorkflowExecution.GetWorkflowId()),
			zap.String(tagRunID, t.WorkflowExecution.GetRunId()),
			zap.String(tagActivityType, activityType),
			zap.Error(err),
		)
	}
	return convertActivityResultToRespondRequest(ath.identity, t.TaskToken, output, err, ath.dataConverter), nil
}
func ( ath * activityTaskHandlerImpl ) Execute ( taskList string , t * s . PollForActivityTaskResponse ) ( result interface { } , err error ) { traceLog ( func ( ) { ath . logger . Debug ( " " , zap . String ( tagWorkflowID , t . WorkflowExecution . GetWorkflowId ( ) ) , zap . String ( tagRunID , t . WorkflowExecution . GetRunId ( ) ) , zap . String ( tagActivityType , t . ActivityType . GetName ( ) ) ) } ) rootCtx := ath . userContext if rootCtx == nil { rootCtx = context . Background ( ) } canCtx , cancel := context . WithCancel ( rootCtx ) defer cancel ( ) invoker := newServiceInvoker ( t . TaskToken , ath . identity , ath . service , cancel , t . GetHeartbeatTimeoutSeconds ( ) , ath . workerStopCh ) defer func ( ) { _ , activityCompleted := result . ( * s . RespondActivityTaskCompletedRequest ) invoker . Close ( ! activityCompleted ) } ( ) workflowType := t . WorkflowType . GetName ( ) activityType := t . ActivityType . GetName ( ) metricsScope := getMetricsScopeForActivity ( ath . metricsScope , workflowType , activityType ) ctx := WithActivityTask ( canCtx , t , taskList , invoker , ath . logger , metricsScope , ath . dataConverter , ath . workerStopCh , ath . contextPropagators , ath . tracer ) activityImplementation := ath . getActivity ( activityType ) if activityImplementation == nil { supported := strings . Join ( ath . getRegisteredActivityNames ( ) , " " ) return nil , fmt . Errorf ( " " , activityType , supported ) } defer func ( ) { if p := recover ( ) ; p != nil { topLine := fmt . Sprintf ( " " , ath . taskListName ) st := getStackTraceRaw ( topLine , 7 , 0 ) ath . logger . Error ( " " , zap . String ( tagWorkflowID , t . WorkflowExecution . GetWorkflowId ( ) ) , zap . String ( tagRunID , t . WorkflowExecution . GetRunId ( ) ) , zap . String ( tagActivityType , activityType ) , zap . String ( " " , fmt . Sprintf ( " " , p ) ) , zap . String ( " " , st ) ) metricsScope . Counter ( metrics . ActivityTaskPanicCounter ) . Inc ( 1 ) panicErr := newPanicError ( p , st ) result , err = convertActivityResultToRespondRequest ( ath . identity , t . TaskToken , nil , panicErr , ath . dataConverter ) , nil } } ( ) for _ , ctxProp := range ath . contextPropagators { var err error if ctx , err = ctxProp . Extract ( ctx , NewHeaderReader ( t . Header ) ) ; err != nil { return nil , fmt . Errorf ( " " , err ) } } info := ctx . Value ( activityEnvContextKey ) . ( * activityEnvironment ) ctx , dlCancelFunc := context . WithDeadline ( ctx , info . deadline ) defer dlCancelFunc ( ) ctx , span := createOpenTracingActivitySpan ( ctx , ath . tracer , time . Now ( ) , activityType , t . WorkflowExecution . GetWorkflowId ( ) , t . WorkflowExecution . GetRunId ( ) ) defer span . Finish ( ) output , err := activityImplementation . Execute ( ctx , t . Input ) dlCancelFunc ( ) if <- ctx . Done ( ) ; ctx . Err ( ) == context . DeadlineExceeded { return nil , ctx . Err ( ) } if err != nil && err != ErrActivityResultPending { ath . logger . Error ( " " , zap . String ( tagWorkflowID , t . WorkflowExecution . GetWorkflowId ( ) ) , zap . String ( tagRunID , t . WorkflowExecution . GetRunId ( ) ) , zap . String ( tagActivityType , activityType ) , zap . Error ( err ) , ) } return convertActivityResultToRespondRequest ( ath . identity , t . TaskToken , output , err , ath . dataConverter ) , nil }
Execute executes an implementation of the activity.
Execute executes an implementation of the activity .
// flush buffered heartbeat if activity was not successfully completed. // Couldn't find the activity implementation. // panic handler // propagate context information into the activity context from the headers