71
queue *PhysicalTaskQueueKey,
72
client matchingservice.MatchingServiceClient,
74
>
partition, ok := queue.Partition().(*tqid.NormalPartition)
75
>
if !ok {
76
return nil, serviceerror.NewInvalidArgument("physical queue of normal partition expected")
77
}
78
80
>
cfg: cfg,
81
>
client: client,
82
>
partition: partition,
83
>
queue: queue,
84
>
outstandingTasksLimit: int32(cfg.ForwarderMaxOutstandingTasks()),
85
>
outstandingPollsLimit: int32(cfg.ForwarderMaxOutstandingPolls()),
86
>
limiter: quotas.NewDefaultOutgoingRateLimiter(
87
>
func() float64 { return float64(cfg.ForwarderMaxRatePerSecond()) },
88
),
89
}
90
>
fwdr.addReqToken.Store(newForwarderReqToken(cfg.ForwarderMaxOutstandingTasks()))
forwarder.go
91
>
fwdr.pollReqToken.Store(newForwarderReqToken(cfg.ForwarderMaxOutstandingPolls()))
92
>
return fwdr, nil
93
}
94