// RecordActivityTaskHeartbeatByID - Record Activity Task Heart beat.
Record Activity Task Heart beat.
// RecordActivityTaskHeartbeatByID - Record Activity Task Heart beat .
func (wh *WorkflowHandler) RecordActivityTaskHeartbeatByID(
	ctx context.Context,
	heartbeatRequest *types.RecordActivityTaskHeartbeatByIDRequest,
) (resp *types.RecordActivityTaskHeartbeatResponse, retError error) {
	defer log.CapturePanic(wh.GetLogger(), &retError)
	scope, sw := wh.startRequestProfileWithDomain(ctx, metrics.FrontendRecordActivityTaskHeartbeatByIDScope, heartbeatRequest)
	defer sw.Stop()
	if wh.isShuttingDown() {
		return nil, errShuttingDown
	}
	if err := wh.versionChecker.ClientSupported(ctx, wh.config.EnableClientVersionCheck()); err != nil {
		return nil, wh.error(err, scope)
	}
	if heartbeatRequest == nil {
		return nil, wh.error(errRequestNotSet, scope)
	}
	domainName := heartbeatRequest.GetDomain()
	tags := getDomainWfIDRunIDTags(domainName, &types.WorkflowExecution{
		WorkflowID: heartbeatRequest.GetWorkflowID(),
		RunID:      heartbeatRequest.GetRunID(),
	})
	if domainName == "" {
		return nil, wh.error(errDomainNotSet, scope, tags...)
	}
	wh.allow(false, heartbeatRequest)
	wh.GetLogger().Debug("Received RecordActivityTaskHeartbeatByID")
	domainID, err := wh.GetDomainCache().GetDomainID(domainName)
	if err != nil {
		return nil, wh.error(err, scope, tags...)
	}
	workflowID := heartbeatRequest.GetWorkflowID()
	runID := heartbeatRequest.GetRunID() 
	activityID := heartbeatRequest.GetActivityID()
	if domainID == "" {
		return nil, wh.error(errDomainNotSet, scope, tags...)
	}
	if workflowID == "" {
		return nil, wh.error(errWorkflowIDNotSet, scope, tags...)
	}
	if activityID == "" {
		return nil, wh.error(errActivityIDNotSet, scope, tags...)
	}
	taskToken := &common.TaskToken{
		DomainID:   domainID,
		RunID:      runID,
		WorkflowID: workflowID,
		ScheduleID: common.EmptyEventID,
		ActivityID: activityID,
	}
	token, err := wh.tokenSerializer.Serialize(taskToken)
	if err != nil {
		return nil, wh.error(err, scope, tags...)
	}
	sizeLimitError := wh.config.BlobSizeLimitError(domainName)
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(domainName)
	if err := common.CheckEventBlobSizeLimit(
		len(heartbeatRequest.Details),
		sizeLimitWarn,
		sizeLimitError,
		taskToken.DomainID,
		taskToken.WorkflowID,
		taskToken.RunID,
		scope,
		wh.GetThrottledLogger(),
		tag.BlobSizeViolationOperation("RecordActivityTaskHeartbeatByID"),
	); err != nil {
		failRequest := &types.RespondActivityTaskFailedRequest{
			TaskToken: token,
			Reason:    common.StringPtr(common.FailureReasonHeartbeatExceedsLimit),
			Details:   heartbeatRequest.Details[0:sizeLimitError],
			Identity:  heartbeatRequest.Identity,
		}
		err = wh.GetHistoryClient().RespondActivityTaskFailed(ctx, &types.HistoryRespondActivityTaskFailedRequest{
			DomainUUID:    taskToken.DomainID,
			FailedRequest: failRequest,
		})
		if err != nil {
			return nil, wh.normalizeVersionedErrors(ctx, wh.error(err, scope, tags...))
		}
		resp = &types.RecordActivityTaskHeartbeatResponse{CancelRequested: true}
	} else {
		req := &types.RecordActivityTaskHeartbeatRequest{
			TaskToken: token,
			Details:   heartbeatRequest.Details,
			Identity:  heartbeatRequest.Identity,
		}
		resp, err = wh.GetHistoryClient().RecordActivityTaskHeartbeat(ctx, &types.HistoryRecordActivityTaskHeartbeatRequest{
			DomainUUID:       taskToken.DomainID,
			HeartbeatRequest: req,
		})
		if err != nil {
			return nil, wh.normalizeVersionedErrors(ctx, wh.error(err, scope, tags...))
		}
	}
	return resp, nil
}
func ( wh * WorkflowHandler ) RecordActivityTaskHeartbeatByID ( ctx context . Context , heartbeatRequest * types . RecordActivityTaskHeartbeatByIDRequest , ) ( resp * types . RecordActivityTaskHeartbeatResponse , retError error ) { defer log . CapturePanic ( wh . GetLogger ( ) , & retError ) scope , sw := wh . startRequestProfileWithDomain ( ctx , metrics . FrontendRecordActivityTaskHeartbeatByIDScope , heartbeatRequest ) defer sw . Stop ( ) if wh . isShuttingDown ( ) { return nil , errShuttingDown } if err := wh . versionChecker . ClientSupported ( ctx , wh . config . EnableClientVersionCheck ( ) ) ; err != nil { return nil , wh . error ( err , scope ) } if heartbeatRequest == nil { return nil , wh . error ( errRequestNotSet , scope ) } domainName := heartbeatRequest . GetDomain ( ) tags := getDomainWfIDRunIDTags ( domainName , & types . WorkflowExecution { WorkflowID : heartbeatRequest . GetWorkflowID ( ) , RunID : heartbeatRequest . GetRunID ( ) , } ) if domainName == " " { return nil , wh . error ( errDomainNotSet , scope , tags ... ) } wh . allow ( false , heartbeatRequest ) wh . GetLogger ( ) . Debug ( " " ) domainID , err := wh . GetDomainCache ( ) . GetDomainID ( domainName ) if err != nil { return nil , wh . error ( err , scope , tags ... ) } workflowID := heartbeatRequest . GetWorkflowID ( ) runID := heartbeatRequest . GetRunID ( ) activityID := heartbeatRequest . GetActivityID ( ) if domainID == " " { return nil , wh . error ( errDomainNotSet , scope , tags ... ) } if workflowID == " " { return nil , wh . error ( errWorkflowIDNotSet , scope , tags ... ) } if activityID == " " { return nil , wh . error ( errActivityIDNotSet , scope , tags ... ) } taskToken := & common . TaskToken { DomainID : domainID , RunID : runID , WorkflowID : workflowID , ScheduleID : common . EmptyEventID , ActivityID : activityID , } token , err := wh . tokenSerializer . Serialize ( taskToken ) if err != nil { return nil , wh . error ( err , scope , tags ... ) } sizeLimitError := wh . config . BlobSizeLimitError ( domainName ) sizeLimitWarn := wh . config . BlobSizeLimitWarn ( domainName ) if err := common . CheckEventBlobSizeLimit ( len ( heartbeatRequest . Details ) , sizeLimitWarn , sizeLimitError , taskToken . DomainID , taskToken . WorkflowID , taskToken . RunID , scope , wh . GetThrottledLogger ( ) , tag . BlobSizeViolationOperation ( " " ) , ) ; err != nil { failRequest := & types . RespondActivityTaskFailedRequest { TaskToken : token , Reason : common . StringPtr ( common . FailureReasonHeartbeatExceedsLimit ) , Details : heartbeatRequest . Details [ 0 : sizeLimitError ] , Identity : heartbeatRequest . Identity , } err = wh . GetHistoryClient ( ) . RespondActivityTaskFailed ( ctx , & types . HistoryRespondActivityTaskFailedRequest { DomainUUID : taskToken . DomainID , FailedRequest : failRequest , } ) if err != nil { return nil , wh . normalizeVersionedErrors ( ctx , wh . error ( err , scope , tags ... ) ) } resp = & types . RecordActivityTaskHeartbeatResponse { CancelRequested : true } } else { req := & types . RecordActivityTaskHeartbeatRequest { TaskToken : token , Details : heartbeatRequest . Details , Identity : heartbeatRequest . Identity , } resp , err = wh . GetHistoryClient ( ) . RecordActivityTaskHeartbeat ( ctx , & types . HistoryRecordActivityTaskHeartbeatRequest { DomainUUID : taskToken . DomainID , HeartbeatRequest : req , } ) if err != nil { return nil , wh . normalizeVersionedErrors ( ctx , wh . error ( err , scope , tags ... ) ) } } return resp , nil }
Record Activity Task Heart beat.
Record Activity Task Heart beat .
// Count the request in the host RPS, // but we still accept it even if RPS is exceeded // runID is optional so can be empty // heartbeat details exceed size limit, we would fail the activity immediately with explicit error reason