Handlers return object and error

controller/generic_controller.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"reflect"
 	"strings"
 	"sync"
 	"time"
@@ -41,12 +42,12 @@ func init() {
 	}
 }
 
-type HandlerFunc func(key string) error
+type HandlerFunc func(key string, obj interface{}) (interface{}, error)
 
 type GenericController interface {
 	SetThreadinessOverride(count int)
 	Informer() cache.SharedIndexInformer
-	AddHandler(name string, handler HandlerFunc)
+	AddHandler(ctx context.Context, name string, handler HandlerFunc)
 	HandlerCount() int
 	Enqueue(namespace, name string)
 	Sync(ctx context.Context) error
@@ -60,15 +61,22 @@ type Backend interface {
 }
 
 type handlerDef struct {
-	name    string
-	handler HandlerFunc
+	name       string
+	generation int
+	handler    HandlerFunc
+}
+
+type generationKey struct {
+	generation int
+	key        string
 }
 
 type genericController struct {
 	sync.Mutex
 	threadinessOverride int
+	generation          int
 	informer            cache.SharedIndexInformer
-	handlers            []handlerDef
+	handlers            []*handlerDef
 	queue               workqueue.RateLimitingInterface
 	name                string
 	running             bool
@@ -116,11 +124,28 @@ func (g *genericController) Enqueue(namespace, name string) {
 	}
 }
 
-func (g *genericController) AddHandler(name string, handler HandlerFunc) {
-	g.handlers = append(g.handlers, handlerDef{
-		name:    name,
-		handler: handler,
-	})
+func (g *genericController) AddHandler(ctx context.Context, name string, handler HandlerFunc) {
+	g.Lock()
+	h := &handlerDef{
+		name:       name,
+		generation: g.generation,
+		handler:    handler,
+	}
+	g.handlers = append(g.handlers, h)
+	g.Unlock()
+
+	go func() {
+		<-ctx.Done()
+		g.Lock()
+		var handlers []*handlerDef
+		for _, handler := range g.handlers {
+			if handler != h {
+				handlers = append(handlers, h)
+			}
+		}
+		g.handlers = handlers
+		g.Unlock()
+	}()
 }
 
 func (g *genericController) Sync(ctx context.Context) error {
@@ -175,6 +200,22 @@ func (g *genericController) Start(ctx context.Context, threadiness int) error {
 		go g.run(ctx, threadiness)
 	}
 
+	if g.running {
+		for _, h := range g.handlers {
+			if h.generation != g.generation {
+				continue
+			}
+			for _, key := range g.informer.GetStore().ListKeys() {
+				g.queueObject(generationKey{
+					generation: g.generation,
+					key:        key,
+				})
+			}
+			break
+		}
+	}
+
+	g.generation++
 	g.running = true
 	return nil
 }
@@ -211,7 +252,7 @@ func (g *genericController) processNextWorkItem() bool {
 	defer g.queue.Done(key)
 
 	// do your work on the key.  This method will contains your "do stuff" logic
-	err := g.syncHandler(key.(string))
+	err := g.syncHandler(key)
 	checkErr := err
 	if handlerErr, ok := checkErr.(*handlerError); ok {
 		checkErr = handlerErr.err
@@ -265,14 +306,39 @@ func filterConflictsError(err error) error {
 	return err
 }
 
-func (g *genericController) syncHandler(s string) (err error) {
+func (g *genericController) syncHandler(key interface{}) (err error) {
 	defer utilruntime.RecoverFromPanic(&err)
 
+	generation := -1
+	var s string
+	var obj interface{}
+
+	switch v := key.(type) {
+	case string:
+		s = v
+	case generationKey:
+		generation = v.generation
+		s = v.key
+	default:
+		return nil
+	}
+
+	obj, exists, err := g.informer.GetStore().GetByKey(s)
+	if err != nil {
+		return err
+	} else if !exists {
+		obj = nil
+	}
+
 	var errs []error
 	for _, handler := range g.handlers {
+		if generation > -1 && handler.generation != generation {
+			continue
+		}
+
 		logrus.Debugf("%s calling handler %s %s", g.name, handler.name, s)
 		metrics.IncTotalHandlerExecution(g.name, handler.name)
-		if err := handler.handler(s); err != nil {
+		if newObj, err := handler.handler(s, obj); err != nil {
 			if !ignoreError(err, false) {
 				metrics.IncTotalHandlerFailure(g.name, handler.name, s)
 			}
@@ -280,6 +346,8 @@ func (g *genericController) syncHandler(s string) (err error) {
 				name: handler.name,
 				err:  err,
 			})
+		} else if newObj != nil && !reflect.ValueOf(newObj).IsNil() {
+			obj = newObj
 		}
 	}
 	err = types.NewErrors(errs...)