func (wh *WorkflowHandler) RespondQueryTaskCompleted(
	ctx context.Context,
	completeRequest *types.RespondQueryTaskCompletedRequest,
) (retError error) {
	defer log.CapturePanic(wh.GetLogger(), &retError)

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

	if wh.isShuttingDown() {
		return errShuttingDown
	}

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

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

	if completeRequest.TaskToken == nil {
		return wh.error(errTaskTokenNotSet, scope)
	}
	queryTaskToken, err := wh.tokenSerializer.DeserializeQueryTaskToken(completeRequest.TaskToken)
	if err != nil {
		return wh.error(err, scope)
	}
	if queryTaskToken.DomainID == "" || queryTaskToken.TaskList == "" || queryTaskToken.TaskID == "" {
		return wh.error(errInvalidTaskToken, scope)
	}

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

	dw := domainWrapper{
		domain: domainName,
	}
	scope, sw := wh.startRequestProfileWithDomain(
		ctx,
		metrics.FrontendRespondQueryTaskCompletedScope,
		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)

	sizeLimitError := wh.config.BlobSizeLimitError(domainName)
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(domainName)

	if err := common.CheckEventBlobSizeLimit(
		len(completeRequest.GetQueryResult()),
		sizeLimitWarn,
		sizeLimitError,
		queryTaskToken.DomainID,
		"",
		"",
		scope,
		wh.GetThrottledLogger(),
		tag.BlobSizeViolationOperation("RespondQueryTaskCompleted"),
	); err != nil {
		completeRequest = &types.RespondQueryTaskCompletedRequest{
			TaskToken:     completeRequest.TaskToken,
			CompletedType: types.QueryTaskCompletedTypeFailed.Ptr(),
			QueryResult:   nil,
			ErrorMessage:  err.Error(),
		}
	}

	call := yarpc.CallFromContext(ctx)

	completeRequest.WorkerVersionInfo = &types.WorkerVersionInfo{
		Impl:           call.Header(common.ClientImplHeaderName),
		FeatureVersion: call.Header(common.FeatureVersionHeaderName),
	}
	matchingRequest := &types.MatchingRespondQueryTaskCompletedRequest{
		DomainUUID:       queryTaskToken.DomainID,
		TaskList:         &types.TaskList{Name: queryTaskToken.TaskList},
		TaskID:           queryTaskToken.TaskID,
		CompletedRequest: completeRequest,
	}

	err = wh.GetMatchingClient().RespondQueryTaskCompleted(ctx, matchingRequest)
	if err != nil {
		return wh.error(err, scope)
	}
	return nil
}
// RespondQueryTaskCompleted - response to a query task
response to a query task
// RespondQueryTaskCompleted - response to a query task
func (wh *WorkflowHandler) RespondQueryTaskCompleted(
	ctx context.Context,
	completeRequest *types.RespondQueryTaskCompletedRequest,
) (retError error) {
	defer log.CapturePanic(wh.GetLogger(), &retError)
	scope := wh.getDefaultScope(ctx, metrics.FrontendRespondQueryTaskCompletedScope)
	if wh.isShuttingDown() {
		return errShuttingDown
	}
	if err := wh.versionChecker.ClientSupported(ctx, wh.config.EnableClientVersionCheck()); err != nil {
		return wh.error(err, scope)
	}
	if completeRequest == nil {
		return wh.error(errRequestNotSet, scope)
	}
	if completeRequest.TaskToken == nil {
		return wh.error(errTaskTokenNotSet, scope)
	}
	queryTaskToken, err := wh.tokenSerializer.DeserializeQueryTaskToken(completeRequest.TaskToken)
	if err != nil {
		return wh.error(err, scope)
	}
	if queryTaskToken.DomainID == "" || queryTaskToken.TaskList == "" || queryTaskToken.TaskID == "" {
		return wh.error(errInvalidTaskToken, scope)
	}
	domainName, err := wh.GetDomainCache().GetDomainName(queryTaskToken.DomainID)
	if err != nil {
		return wh.error(err, scope)
	}
	dw := domainWrapper{
		domain: domainName,
	}
	scope, sw := wh.startRequestProfileWithDomain(
		ctx,
		metrics.FrontendRespondQueryTaskCompletedScope,
		dw,
	)
	defer sw.Stop()
	wh.allow(false, dw)
	sizeLimitError := wh.config.BlobSizeLimitError(domainName)
	sizeLimitWarn := wh.config.BlobSizeLimitWarn(domainName)
	if err := common.CheckEventBlobSizeLimit(
		len(completeRequest.GetQueryResult()),
		sizeLimitWarn,
		sizeLimitError,
		queryTaskToken.DomainID,
		"",
		"",
		scope,
		wh.GetThrottledLogger(),
		tag.BlobSizeViolationOperation("RespondQueryTaskCompleted"),
	); err != nil {
		completeRequest = &types.RespondQueryTaskCompletedRequest{
			TaskToken:     completeRequest.TaskToken,
			CompletedType: types.QueryTaskCompletedTypeFailed.Ptr(),
			QueryResult:   nil,
			ErrorMessage:  err.Error(),
		}
	}
	call := yarpc.CallFromContext(ctx)
	completeRequest.WorkerVersionInfo = &types.WorkerVersionInfo{
		Impl:           call.Header(common.ClientImplHeaderName),
		FeatureVersion: call.Header(common.FeatureVersionHeaderName),
	}
	matchingRequest := &types.MatchingRespondQueryTaskCompletedRequest{
		DomainUUID:       queryTaskToken.DomainID,
		TaskList:         &types.TaskList{Name: queryTaskToken.TaskList},
		TaskID:           queryTaskToken.TaskID,
		CompletedRequest: completeRequest,
	}
	err = wh.GetMatchingClient().RespondQueryTaskCompleted(ctx, matchingRequest)
	if err != nil {
		return wh.error(err, scope)
	}
	return nil
}
func ( wh * WorkflowHandler ) RespondQueryTaskCompleted ( ctx context . Context , completeRequest * types . RespondQueryTaskCompletedRequest , ) ( retError error ) { defer log . CapturePanic ( wh . GetLogger ( ) , & retError ) scope := wh . getDefaultScope ( ctx , metrics . FrontendRespondQueryTaskCompletedScope ) if wh . isShuttingDown ( ) { return errShuttingDown } if err := wh . versionChecker . ClientSupported ( ctx , wh . config . EnableClientVersionCheck ( ) ) ; err != nil { return wh . error ( err , scope ) } if completeRequest == nil { return wh . error ( errRequestNotSet , scope ) } if completeRequest . TaskToken == nil { return wh . error ( errTaskTokenNotSet , scope ) } queryTaskToken , err := wh . tokenSerializer . DeserializeQueryTaskToken ( completeRequest . TaskToken ) if err != nil { return wh . error ( err , scope ) } if queryTaskToken . DomainID == " " || queryTaskToken . TaskList == " " || queryTaskToken . TaskID == " " { return wh . error ( errInvalidTaskToken , scope ) } domainName , err := wh . GetDomainCache ( ) . GetDomainName ( queryTaskToken . DomainID ) if err != nil { return wh . error ( err , scope ) } dw := domainWrapper { domain : domainName , } scope , sw := wh . startRequestProfileWithDomain ( ctx , metrics . FrontendRespondQueryTaskCompletedScope , dw , ) defer sw . Stop ( ) wh . allow ( false , dw ) sizeLimitError := wh . config . BlobSizeLimitError ( domainName ) sizeLimitWarn := wh . config . BlobSizeLimitWarn ( domainName ) if err := common . CheckEventBlobSizeLimit ( len ( completeRequest . GetQueryResult ( ) ) , sizeLimitWarn , sizeLimitError , queryTaskToken . DomainID , " " , " " , scope , wh . GetThrottledLogger ( ) , tag . BlobSizeViolationOperation ( " " ) , ) ; err != nil { completeRequest = & types . RespondQueryTaskCompletedRequest { TaskToken : completeRequest . TaskToken , CompletedType : types . QueryTaskCompletedTypeFailed . Ptr ( ) , QueryResult : nil , ErrorMessage : err . Error ( ) , } } call := yarpc . CallFromContext ( ctx ) completeRequest . WorkerVersionInfo = & types . WorkerVersionInfo { Impl : call . Header ( common . ClientImplHeaderName ) , FeatureVersion : call . Header ( common . FeatureVersionHeaderName ) , } matchingRequest := & types . MatchingRespondQueryTaskCompletedRequest { DomainUUID : queryTaskToken . DomainID , TaskList : & types . TaskList { Name : queryTaskToken . TaskList } , TaskID : queryTaskToken . TaskID , CompletedRequest : completeRequest , } err = wh . GetMatchingClient ( ) . RespondQueryTaskCompleted ( ctx , matchingRequest ) if err != nil { return wh . error ( err , scope ) } return nil }
response to a query task
response to a query task
// Count the request in the host RPS, // but we still accept it even if RPS is exceeded