processNextWorkItem will read a single work item off the workqueue and attempt to process it, by calling the syncHandler.
()
| 121 | // processNextWorkItem will read a single work item off the workqueue and |
| 122 | // attempt to process it, by calling the syncHandler. |
| 123 | func (c *Controller) processNextWorkItem() bool { |
| 124 | obj, shutdown := c.workqueue.Get() |
| 125 | |
| 126 | if shutdown { |
| 127 | return false |
| 128 | } |
| 129 | |
| 130 | err := func(obj interface{}) error { |
| 131 | defer c.workqueue.Done(obj) |
| 132 | var item workItem |
| 133 | var ok bool |
| 134 | if item, ok = obj.(workItem); !ok { |
| 135 | c.workqueue.Forget(obj) |
| 136 | utilruntime.HandleError(fmt.Errorf("expected workItem in workqueue but got %#v", obj)) |
| 137 | return nil |
| 138 | } |
| 139 | if err := c.syncHandler(item); err != nil { |
| 140 | c.workqueue.AddRateLimited(item) |
| 141 | return fmt.Errorf("%s/%s: error syncing %#v: %s, requeuing", item.namespace, item.name, item, err.Error()) |
| 142 | } |
| 143 | c.workqueue.Forget(obj) |
| 144 | klog.V(5).Infof("%s/%s: successfully performed %s", item.namespace, item.name, taskLabels[item.task]) |
| 145 | return nil |
| 146 | }(obj) |
| 147 | |
| 148 | if err != nil { |
| 149 | utilruntime.HandleError(err) |
| 150 | return true |
| 151 | } |
| 152 | |
| 153 | return true |
| 154 | } |
| 155 | |
| 156 | func (c *Controller) syncHandler(w workItem) error { |
| 157 | var err error |