36
)
37
39
>
flowControlStates := make(map[enumsspb.TaskPriority]*flowControlState)
40
>
highPriorityState := &flowControlState{
41
>
resume: true,
42
>
}
43
>
highPriorityState.cond = sync.NewCond(&highPriorityState.mu)
44
>
highPriorityState.rateLimiter = quotas.NewDefaultOutgoingRateLimiter(func() float64 {
45
>
return float64(config.ReplicationStreamSenderHighPriorityQPS())
46
>
})
47
49
>
resume: true,
50
>
}
51
>
lowPriorityState.cond = sync.NewCond(&lowPriorityState.mu)
52
>
lowPriorityState.rateLimiter = quotas.NewDefaultOutgoingRateLimiter(func() float64 {
53
>
return float64(config.ReplicationStreamSenderLowPriorityQPS())
54
>
})
55
>
flowControlStates[enumsspb.TASK_PRIORITY_HIGH] = highPriorityState
56
>
flowControlStates[enumsspb.TASK_PRIORITY_LOW] = lowPriorityState
57
>
return &SenderFlowControllerImpl{
58
>
flowControlStates: flowControlStates,
59
>
defaultRateLimiter: quotas.NewRateLimiter(float64(config.ReplicationStreamSenderHighPriorityQPS()), config.ReplicationStreamSenderHighPriorityQPS()),
60
>
logger: logger,
61
>
}
62
}
63