func (wh *WorkflowHandler) RespondDecisionTaskCompleted(
	ctx context.Context,
	completeRequest *types.RespondDecisionTaskCompletedRequest,
) (resp *types.RespondDecisionTaskCompletedResponse, retError error) {
	defer log.CapturePanic(wh.GetLogger(), &retError)

	scope := wh.getDefaultScope(ctx, metrics.FrontendRespondDecisionTaskCompletedScope)

	if wh.isShuttingDown() {
		return nil, errShuttingDown
	}

	if err := wh.versionChecker.ClientSupported(ctx, wh.config.EnableClientVersionCheck()); err != nil {
		return nil, wh.error(err, scope)
	}

	if completeRequest == nil {
		return nil, wh.error(errRequestNotSet, scope)
	}

	if completeRequest.TaskToken == nil {
		return nil, wh.error(errTaskTokenNotSet, scope)
	}
	taskToken, err := wh.tokenSerializer.Deserialize(completeRequest.TaskToken)
	if err != nil {
		return nil, wh.error(err, scope)
	}
	if taskToken.DomainID == "" {
		return nil, wh.error(errDomainNotSet, scope)
	}

	domainName, err := wh.GetDomainCache().GetDomainName(taskToken.DomainID)
	if err != nil {
		return nil, wh.error(err, scope)
	}

	dw := domainWrapper{
		domain: domainName,
	}
	scope, sw := wh.startRequestProfileWithDomain(
		ctx,
		metrics.FrontendRespondDecisionTaskCompletedScope,
		dw,
	)
	defer sw.Stop()

	// Count the request in the host RPS,
	// but we still accept it even if RPS is exceeded
	wh.allow(false, dw)

	tags := getDomainWfIDRunIDTags(domainName, &types.WorkflowExecution{
		WorkflowID: taskToken.WorkflowID,
		RunID:      taskToken.RunID,
	})

	if !common.ValidIDLength(
		completeRequest.GetIdentity(),
		scope,
		wh.config.MaxIDLengthWarnLimit(),
		wh.config.IdentityMaxLength(domainName),
		metrics.CadenceErrIdentityExceededWarnLimit,
		domainName,
		wh.GetLogger(),
		tag.IDTypeIdentity) {
		return nil, wh.error(errIdentityTooLong, scope, tags...)
	}

	if err := common.CheckDecisionResultLimit(
		len(completeRequest.Decisions),
		wh.config.DecisionResultCountLimit(domainName),
		scope); err != nil {
		return nil, wh.error(err, scope)
	}

	histResp, err := wh.GetHistoryClient().RespondDecisionTaskCompleted(ctx, &types.HistoryRespondDecisionTaskCompletedRequest{
		DomainUUID:      taskToken.DomainID,
		CompleteRequest: completeRequest},
	)
	if err != nil {
		return nil, wh.normalizeVersionedErrors(ctx, wh.error(err, scope, tags...))
	}

	completedResp := &types.RespondDecisionTaskCompletedResponse{}
	completedResp.ActivitiesToDispatchLocally = histResp.ActivitiesToDispatchLocally
	if completeRequest.GetReturnNewDecisionTask() && histResp != nil && histResp.StartedResponse != nil {
		taskToken := &common.TaskToken{
			DomainID:        taskToken.DomainID,
			WorkflowID:      taskToken.WorkflowID,
			RunID:           taskToken.RunID,
			ScheduleID:      histResp.StartedResponse.GetScheduledEventID(),
			ScheduleAttempt: histResp.StartedResponse.GetAttempt(),
		}
		token, _ := wh.tokenSerializer.Serialize(taskToken)
		workflowExecution := &types.WorkflowExecution{
			WorkflowID: taskToken.WorkflowID,
			RunID:      taskToken.RunID,
		}
		matchingResp := common.CreateMatchingPollForDecisionTaskResponse(histResp.StartedResponse, workflowExecution, token)

		newDecisionTask, err := wh.createPollForDecisionTaskResponse(ctx, scope, taskToken.DomainID, matchingResp, matchingResp.GetBranchToken())
		if err != nil {
			return nil, wh.error(err, scope, tags...)
		}
		completedResp.DecisionTask = newDecisionTask
	}

	return completedResp, nil
}
// RespondDecisionTaskCompleted - response to a decision task
response to a decision task
// RespondDecisionTaskCompleted - response to a decision task
func (wh *WorkflowHandler) RespondDecisionTaskCompleted(
	ctx context.Context,
	completeRequest *types.RespondDecisionTaskCompletedRequest,
) (resp *types.RespondDecisionTaskCompletedResponse, retError error) {
	defer log.CapturePanic(wh.GetLogger(), &retError)
	scope := wh.getDefaultScope(ctx, metrics.FrontendRespondDecisionTaskCompletedScope)
	if wh.isShuttingDown() {
		return nil, errShuttingDown
	}
	if err := wh.versionChecker.ClientSupported(ctx, wh.config.EnableClientVersionCheck()); err != nil {
		return nil, wh.error(err, scope)
	}
	if completeRequest == nil {
		return nil, wh.error(errRequestNotSet, scope)
	}
	if completeRequest.TaskToken == nil {
		return nil, wh.error(errTaskTokenNotSet, scope)
	}
	taskToken, err := wh.tokenSerializer.Deserialize(completeRequest.TaskToken)
	if err != nil {
		return nil, wh.error(err, scope)
	}
	if taskToken.DomainID == "" {
		return nil, wh.error(errDomainNotSet, scope)
	}
	domainName, err := wh.GetDomainCache().GetDomainName(taskToken.DomainID)
	if err != nil {
		return nil, wh.error(err, scope)
	}
	dw := domainWrapper{
		domain: domainName,
	}
	scope, sw := wh.startRequestProfileWithDomain(
		ctx,
		metrics.FrontendRespondDecisionTaskCompletedScope,
		dw,
	)
	defer sw.Stop()
	wh.allow(false, dw)
	tags := getDomainWfIDRunIDTags(domainName, &types.WorkflowExecution{
		WorkflowID: taskToken.WorkflowID,
		RunID:      taskToken.RunID,
	})
	if !common.ValidIDLength(
		completeRequest.GetIdentity(),
		scope,
		wh.config.MaxIDLengthWarnLimit(),
		wh.config.IdentityMaxLength(domainName),
		metrics.CadenceErrIdentityExceededWarnLimit,
		domainName,
		wh.GetLogger(),
		tag.IDTypeIdentity) {
		return nil, wh.error(errIdentityTooLong, scope, tags...)
	}
	if err := common.CheckDecisionResultLimit(
		len(completeRequest.Decisions),
		wh.config.DecisionResultCountLimit(domainName),
		scope); err != nil {
		return nil, wh.error(err, scope)
	}
	histResp, err := wh.GetHistoryClient().RespondDecisionTaskCompleted(ctx, &types.HistoryRespondDecisionTaskCompletedRequest{
		DomainUUID:      taskToken.DomainID,
		CompleteRequest: completeRequest},
	)
	if err != nil {
		return nil, wh.normalizeVersionedErrors(ctx, wh.error(err, scope, tags...))
	}
	completedResp := &types.RespondDecisionTaskCompletedResponse{}
	completedResp.ActivitiesToDispatchLocally = histResp.ActivitiesToDispatchLocally
	if completeRequest.GetReturnNewDecisionTask() && histResp != nil && histResp.StartedResponse != nil {
		taskToken := &common.TaskToken{
			DomainID:        taskToken.DomainID,
			WorkflowID:      taskToken.WorkflowID,
			RunID:           taskToken.RunID,
			ScheduleID:      histResp.StartedResponse.GetScheduledEventID(),
			ScheduleAttempt: histResp.StartedResponse.GetAttempt(),
		}
		token, _ := wh.tokenSerializer.Serialize(taskToken)
		workflowExecution := &types.WorkflowExecution{
			WorkflowID: taskToken.WorkflowID,
			RunID:      taskToken.RunID,
		}
		matchingResp := common.CreateMatchingPollForDecisionTaskResponse(histResp.StartedResponse, workflowExecution, token)
		newDecisionTask, err := wh.createPollForDecisionTaskResponse(ctx, scope, taskToken.DomainID, matchingResp, matchingResp.GetBranchToken())
		if err != nil {
			return nil, wh.error(err, scope, tags...)
		}
		completedResp.DecisionTask = newDecisionTask
	}
	return completedResp, nil
}
func ( wh * WorkflowHandler ) RespondDecisionTaskCompleted ( ctx context . Context , completeRequest * types . RespondDecisionTaskCompletedRequest , ) ( resp * types . RespondDecisionTaskCompletedResponse , retError error ) { defer log . CapturePanic ( wh . GetLogger ( ) , & retError ) scope := wh . getDefaultScope ( ctx , metrics . FrontendRespondDecisionTaskCompletedScope ) if wh . isShuttingDown ( ) { return nil , errShuttingDown } if err := wh . versionChecker . ClientSupported ( ctx , wh . config . EnableClientVersionCheck ( ) ) ; err != nil { return nil , wh . error ( err , scope ) } if completeRequest == nil { return nil , wh . error ( errRequestNotSet , scope ) } if completeRequest . TaskToken == nil { return nil , wh . error ( errTaskTokenNotSet , scope ) } taskToken , err := wh . tokenSerializer . Deserialize ( completeRequest . TaskToken ) if err != nil { return nil , wh . error ( err , scope ) } if taskToken . DomainID == " " { return nil , wh . error ( errDomainNotSet , scope ) } domainName , err := wh . GetDomainCache ( ) . GetDomainName ( taskToken . DomainID ) if err != nil { return nil , wh . error ( err , scope ) } dw := domainWrapper { domain : domainName , } scope , sw := wh . startRequestProfileWithDomain ( ctx , metrics . FrontendRespondDecisionTaskCompletedScope , dw , ) defer sw . Stop ( ) wh . allow ( false , dw ) tags := getDomainWfIDRunIDTags ( domainName , & types . WorkflowExecution { WorkflowID : taskToken . WorkflowID , RunID : taskToken . RunID , } ) if ! common . ValidIDLength ( completeRequest . GetIdentity ( ) , scope , wh . config . MaxIDLengthWarnLimit ( ) , wh . config . IdentityMaxLength ( domainName ) , metrics . CadenceErrIdentityExceededWarnLimit , domainName , wh . GetLogger ( ) , tag . IDTypeIdentity ) { return nil , wh . error ( errIdentityTooLong , scope , tags ... ) } if err := common . CheckDecisionResultLimit ( len ( completeRequest . Decisions ) , wh . config . DecisionResultCountLimit ( domainName ) , scope ) ; err != nil { return nil , wh . error ( err , scope ) } histResp , err := wh . GetHistoryClient ( ) . RespondDecisionTaskCompleted ( ctx , & types . HistoryRespondDecisionTaskCompletedRequest { DomainUUID : taskToken . DomainID , CompleteRequest : completeRequest } , ) if err != nil { return nil , wh . normalizeVersionedErrors ( ctx , wh . error ( err , scope , tags ... ) ) } completedResp := & types . RespondDecisionTaskCompletedResponse { } completedResp . ActivitiesToDispatchLocally = histResp . ActivitiesToDispatchLocally if completeRequest . GetReturnNewDecisionTask ( ) && histResp != nil && histResp . StartedResponse != nil { taskToken := & common . TaskToken { DomainID : taskToken . DomainID , WorkflowID : taskToken . WorkflowID , RunID : taskToken . RunID , ScheduleID : histResp . StartedResponse . GetScheduledEventID ( ) , ScheduleAttempt : histResp . StartedResponse . GetAttempt ( ) , } token , _ := wh . tokenSerializer . Serialize ( taskToken ) workflowExecution := & types . WorkflowExecution { WorkflowID : taskToken . WorkflowID , RunID : taskToken . RunID , } matchingResp := common . CreateMatchingPollForDecisionTaskResponse ( histResp . StartedResponse , workflowExecution , token ) newDecisionTask , err := wh . createPollForDecisionTaskResponse ( ctx , scope , taskToken . DomainID , matchingResp , matchingResp . GetBranchToken ( ) ) if err != nil { return nil , wh . error ( err , scope , tags ... ) } completedResp . DecisionTask = newDecisionTask } return completedResp , nil }
response to a decision task
response to a decision task
// Count the request in the host RPS, // but we still accept it even if RPS is exceeded