106
)
107
109
>
f := &outboundQueueFactory{
110
>
outboundQueueFactoryParams: params,
111
>
hostReaderRateLimiter: queues.NewReaderPriorityRateLimiter(
112
>
NewHostRateLimiterRateFn(
113
>
params.Config.OutboundProcessorMaxPollHostRPS,
114
>
params.Config.PersistenceMaxQPS,
115
>
outboundQueuePersistenceMaxRPSRatio,
116
>
),
117
>
int64(params.Config.OutboundQueueMaxReaderCount()),
118
>
),
119
>
hostScheduler: &queues.CommonSchedulerWrapper{
120
>
Scheduler: ctasks.NewGroupByScheduler(
121
>
ctasks.GroupBySchedulerOptions[
122
>
tasks.TaskGroupNamespaceIDAndDestination,
123
>
queues.Executable,
124
>
]{
125
>
Logger: params.Logger,
126
>
KeyFn: func(e queues.Executable) tasks.TaskGroupNamespaceIDAndDestination {
127
return grouper.KeyTyped(e.GetTask())
128
},