mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-24 04:06:03 +00:00
Merge pull request #87393 from MikeSpreitzer/remove-unused-loop
remove unused layer of loop structure in processorListener::run
This commit is contained in:
commit
94ea2be3cc
@ -88,7 +88,6 @@ go_library(
|
||||
"//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
|
||||
"//staging/src/k8s.io/client-go/rest:go_default_library",
|
||||
"//staging/src/k8s.io/client-go/tools/pager:go_default_library",
|
||||
"//staging/src/k8s.io/client-go/util/retry:go_default_library",
|
||||
"//vendor/k8s.io/klog:go_default_library",
|
||||
"//vendor/k8s.io/utils/buffer:go_default_library",
|
||||
"//vendor/k8s.io/utils/trace:go_default_library",
|
||||
|
@ -25,7 +25,6 @@ import (
|
||||
"k8s.io/apimachinery/pkg/util/clock"
|
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/client-go/util/retry"
|
||||
"k8s.io/utils/buffer"
|
||||
|
||||
"k8s.io/klog"
|
||||
@ -700,34 +699,26 @@ func (p *processorListener) pop() {
|
||||
|
||||
func (p *processorListener) run() {
|
||||
// this call blocks until the channel is closed. When a panic happens during the notification
|
||||
// we will catch it, **the offending item will be skipped!**, and after a short delay (one minute)
|
||||
// we will catch it, **the offending item will be skipped!**, and after a short delay (one second)
|
||||
// the next notification will be attempted. This is usually better than the alternative of never
|
||||
// delivering again.
|
||||
stopCh := make(chan struct{})
|
||||
wait.Until(func() {
|
||||
// this gives us a few quick retries before a long pause and then a few more quick retries
|
||||
err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
|
||||
for next := range p.nextCh {
|
||||
switch notification := next.(type) {
|
||||
case updateNotification:
|
||||
p.handler.OnUpdate(notification.oldObj, notification.newObj)
|
||||
case addNotification:
|
||||
p.handler.OnAdd(notification.newObj)
|
||||
case deleteNotification:
|
||||
p.handler.OnDelete(notification.oldObj)
|
||||
default:
|
||||
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %T", next))
|
||||
}
|
||||
for next := range p.nextCh {
|
||||
switch notification := next.(type) {
|
||||
case updateNotification:
|
||||
p.handler.OnUpdate(notification.oldObj, notification.newObj)
|
||||
case addNotification:
|
||||
p.handler.OnAdd(notification.newObj)
|
||||
case deleteNotification:
|
||||
p.handler.OnDelete(notification.oldObj)
|
||||
default:
|
||||
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %T", next))
|
||||
}
|
||||
// the only way to get here is if the p.nextCh is empty and closed
|
||||
return true, nil
|
||||
})
|
||||
|
||||
// the only way to get here is if the p.nextCh is empty and closed
|
||||
if err == nil {
|
||||
close(stopCh)
|
||||
}
|
||||
}, 1*time.Minute, stopCh)
|
||||
// the only way to get here is if the p.nextCh is empty and closed
|
||||
close(stopCh)
|
||||
}, 1*time.Second, stopCh)
|
||||
}
|
||||
|
||||
// shouldResync deterimines if the listener needs a resync. If the listener's resyncPeriod is 0,
|
||||
|
Loading…
Reference in New Issue
Block a user