func (wh *WorkflowHandler) PollForActivityTask(
	ctx context.Context,
	pollRequest *types.PollForActivityTaskRequest,
) (resp *types.PollForActivityTaskResponse, retError error) {
	defer log.CapturePanic(wh.GetLogger(), &retError)

	callTime := time.Now()

	scope, sw := wh.startRequestProfileWithDomain(ctx, metrics.FrontendPollForActivityTaskScope, pollRequest)
	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 pollRequest == nil {
		return nil, wh.error(errRequestNotSet, scope)
	}

	domainName := pollRequest.GetDomain()
	tags := getDomainWfIDRunIDTags(domainName, nil)

	if domainName == "" {
		return nil, wh.error(errDomainNotSet, scope, tags...)
	}

	wh.GetLogger().Debug("Received PollForActivityTask")
	if err := common.ValidateLongPollContextTimeout(
		ctx,
		"PollForActivityTask",
		wh.GetThrottledLogger(),
	); err != nil {
		return nil, wh.error(err, scope, tags...)
	}

	idLengthWarnLimit := wh.config.MaxIDLengthWarnLimit()
	if !common.ValidIDLength(
		domainName,
		scope,
		idLengthWarnLimit,
		wh.config.DomainNameMaxLength(domainName),
		metrics.CadenceErrDomainNameExceededWarnLimit,
		domainName,
		wh.GetLogger(),
		tag.IDTypeDomainName) {
		return nil, wh.error(errDomainTooLong, scope, tags...)
	}

	if err := wh.validateTaskList(pollRequest.TaskList, scope, domainName); err != nil {
		return nil, wh.error(err, scope, tags...)
	}

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

	if ok := wh.allow(false, pollRequest); !ok {
		// pollers exponentially back off up to 10s
		return nil, wh.error(createServiceBusyError(), scope, tags...)
	}

	domainID, err := wh.GetDomainCache().GetDomainID(domainName)
	if err != nil {
		return nil, wh.error(err, scope, tags...)
	}

	pollerID := uuid.New()
	op := func() error {
		resp, err = wh.GetMatchingClient().PollForActivityTask(ctx, &types.MatchingPollForActivityTaskRequest{
			DomainUUID:  domainID,
			PollerID:    pollerID,
			PollRequest: pollRequest,
		})
		return err
	}

	err = wh.throttleRetry.Do(ctx, op)
	if err != nil {
		err = wh.cancelOutstandingPoll(ctx, err, domainID, persistence.TaskListTypeActivity, pollRequest.TaskList, pollerID)
		if err != nil {
			// For all other errors log an error and return it back to client.
			ctxTimeout := "not-set"
			ctxDeadline, ok := ctx.Deadline()
			if ok {
				ctxTimeout = ctxDeadline.Sub(callTime).String()
			}
			wh.GetLogger().Error("PollForActivityTask failed.",
				tag.WorkflowTaskListName(pollRequest.GetTaskList().GetName()),
				tag.Value(ctxTimeout),
				tag.Error(err))
			return nil, wh.error(err, scope, tags...)
		}
	}
	return resp, nil
}
// PollForActivityTask - Poll for an activity task.
Poll for an activity task.
// PollForActivityTask - Poll for an activity task .
func (wh *WorkflowHandler) PollForActivityTask(
	ctx context.Context,
	pollRequest *types.PollForActivityTaskRequest,
) (resp *types.PollForActivityTaskResponse, retError error) {
	defer log.CapturePanic(wh.GetLogger(), &retError)
	callTime := time.Now()
	scope, sw := wh.startRequestProfileWithDomain(ctx, metrics.FrontendPollForActivityTaskScope, pollRequest)
	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 pollRequest == nil {
		return nil, wh.error(errRequestNotSet, scope)
	}
	domainName := pollRequest.GetDomain()
	tags := getDomainWfIDRunIDTags(domainName, nil)
	if domainName == "" {
		return nil, wh.error(errDomainNotSet, scope, tags...)
	}
	wh.GetLogger().Debug("Received PollForActivityTask")
	if err := common.ValidateLongPollContextTimeout(
		ctx,
		"PollForActivityTask",
		wh.GetThrottledLogger(),
	); err != nil {
		return nil, wh.error(err, scope, tags...)
	}
	idLengthWarnLimit := wh.config.MaxIDLengthWarnLimit()
	if !common.ValidIDLength(
		domainName,
		scope,
		idLengthWarnLimit,
		wh.config.DomainNameMaxLength(domainName),
		metrics.CadenceErrDomainNameExceededWarnLimit,
		domainName,
		wh.GetLogger(),
		tag.IDTypeDomainName) {
		return nil, wh.error(errDomainTooLong, scope, tags...)
	}
	if err := wh.validateTaskList(pollRequest.TaskList, scope, domainName); err != nil {
		return nil, wh.error(err, scope, tags...)
	}
	if !common.ValidIDLength(
		pollRequest.GetIdentity(),
		scope,
		idLengthWarnLimit,
		wh.config.IdentityMaxLength(domainName),
		metrics.CadenceErrIdentityExceededWarnLimit,
		domainName,
		wh.GetLogger(),
		tag.IDTypeIdentity) {
		return nil, wh.error(errIdentityTooLong, scope, tags...)
	}
	if ok := wh.allow(false, pollRequest); !ok {
		return nil, wh.error(createServiceBusyError(), scope, tags...)
	}
	domainID, err := wh.GetDomainCache().GetDomainID(domainName)
	if err != nil {
		return nil, wh.error(err, scope, tags...)
	}
	pollerID := uuid.New()
	op := func() error {
		resp, err = wh.GetMatchingClient().PollForActivityTask(ctx, &types.MatchingPollForActivityTaskRequest{
			DomainUUID:  domainID,
			PollerID:    pollerID,
			PollRequest: pollRequest,
		})
		return err
	}
	err = wh.throttleRetry.Do(ctx, op)
	if err != nil {
		err = wh.cancelOutstandingPoll(ctx, err, domainID, persistence.TaskListTypeActivity, pollRequest.TaskList, pollerID)
		if err != nil {
			ctxTimeout := "not-set"
			ctxDeadline, ok := ctx.Deadline()
			if ok {
				ctxTimeout = ctxDeadline.Sub(callTime).String()
			}
			wh.GetLogger().Error("PollForActivityTask failed.",
				tag.WorkflowTaskListName(pollRequest.GetTaskList().GetName()),
				tag.Value(ctxTimeout),
				tag.Error(err))
			return nil, wh.error(err, scope, tags...)
		}
	}
	return resp, nil
}
func ( wh * WorkflowHandler ) PollForActivityTask ( ctx context . Context , pollRequest * types . PollForActivityTaskRequest , ) ( resp * types . PollForActivityTaskResponse , retError error ) { defer log . CapturePanic ( wh . GetLogger ( ) , & retError ) callTime := time . Now ( ) scope , sw := wh . startRequestProfileWithDomain ( ctx , metrics . FrontendPollForActivityTaskScope , pollRequest ) 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 pollRequest == nil { return nil , wh . error ( errRequestNotSet , scope ) } domainName := pollRequest . GetDomain ( ) tags := getDomainWfIDRunIDTags ( domainName , nil ) if domainName == " " { return nil , wh . error ( errDomainNotSet , scope , tags ... ) } wh . GetLogger ( ) . Debug ( " " ) if err := common . ValidateLongPollContextTimeout ( ctx , " " , wh . GetThrottledLogger ( ) , ) ; err != nil { return nil , wh . error ( err , scope , tags ... ) } idLengthWarnLimit := wh . config . MaxIDLengthWarnLimit ( ) if ! common . ValidIDLength ( domainName , scope , idLengthWarnLimit , wh . config . DomainNameMaxLength ( domainName ) , metrics . CadenceErrDomainNameExceededWarnLimit , domainName , wh . GetLogger ( ) , tag . IDTypeDomainName ) { return nil , wh . error ( errDomainTooLong , scope , tags ... ) } if err := wh . validateTaskList ( pollRequest . TaskList , scope , domainName ) ; err != nil { return nil , wh . error ( err , scope , tags ... ) } if ! common . ValidIDLength ( pollRequest . GetIdentity ( ) , scope , idLengthWarnLimit , wh . config . IdentityMaxLength ( domainName ) , metrics . CadenceErrIdentityExceededWarnLimit , domainName , wh . GetLogger ( ) , tag . IDTypeIdentity ) { return nil , wh . error ( errIdentityTooLong , scope , tags ... ) } if ok := wh . allow ( false , pollRequest ) ; ! ok { return nil , wh . error ( createServiceBusyError ( ) , scope , tags ... ) } domainID , err := wh . GetDomainCache ( ) . GetDomainID ( domainName ) if err != nil { return nil , wh . error ( err , scope , tags ... ) } pollerID := uuid . New ( ) op := func ( ) error { resp , err = wh . GetMatchingClient ( ) . PollForActivityTask ( ctx , & types . MatchingPollForActivityTaskRequest { DomainUUID : domainID , PollerID : pollerID , PollRequest : pollRequest , } ) return err } err = wh . throttleRetry . Do ( ctx , op ) if err != nil { err = wh . cancelOutstandingPoll ( ctx , err , domainID , persistence . TaskListTypeActivity , pollRequest . TaskList , pollerID ) if err != nil { ctxTimeout := " " ctxDeadline , ok := ctx . Deadline ( ) if ok { ctxTimeout = ctxDeadline . Sub ( callTime ) . String ( ) } wh . GetLogger ( ) . Error ( " " , tag . WorkflowTaskListName ( pollRequest . GetTaskList ( ) . GetName ( ) ) , tag . Value ( ctxTimeout ) , tag . Error ( err ) ) return nil , wh . error ( err , scope , tags ... ) } } return resp , nil }
Poll for an activity task.
Poll for an activity task .
// pollers exponentially back off up to 10s // For all other errors log an error and return it back to client.