@@ -351,14 +351,15 @@ func (c *MasterConfig) RunDeploymentConfigController() {

 // RunDeploymentTriggerController starts the deployment trigger controller process.

 func (c *MasterConfig) RunDeploymentTriggerController() {

+	dcInfomer := c.Informers.DeploymentConfigs().Informer()

+	streamInformer := c.Informers.ImageStreams().Informer()

 	osclient, kclient := c.DeploymentTriggerControllerClients()

-	factory := triggercontroller.DeploymentTriggerControllerFactory{

-		Client:     osclient,

-		KubeClient: kclient,

-		Codec:      c.EtcdHelper.Codec(),

-	}

-	controller := factory.Create()

-	controller.Run()

+

+	controller := triggercontroller.NewDeploymentTriggerController(dcInfomer, streamInformer, osclient, kclient, c.EtcdHelper.Codec())

+	// TODO: Make the stop channel actually work.

+	stopCh := make(chan struct{})

+	// TODO: Make the number of workers configurable.

+	go controller.Run(5, stopCh)

 }

 // RunDeploymentImageChangeTriggerController starts the image change trigger controller process.

@@ -21,10 +21,6 @@ type ClusterPolicyInformer interface {

 	Lister() client.SyncedClusterPoliciesListerInterface

 }

-func (f *sharedInformerFactory) ClusterPolicies() ClusterPolicyInformer {

-	return &clusterPolicyInformer{sharedInformerFactory: f}

-}

-

 type clusterPolicyInformer struct {

 	*sharedInformerFactory

 }

@@ -79,10 +75,6 @@ type ClusterPolicyBindingInformer interface {

 	Lister() client.SyncedClusterPolicyBindingsListerInterface

 }

-func (f *sharedInformerFactory) ClusterPolicyBindings() ClusterPolicyBindingInformer {

-	return &clusterPolicyBindingInformer{sharedInformerFactory: f}

-}

-

 type clusterPolicyBindingInformer struct {

 	*sharedInformerFactory

 }

@@ -137,10 +129,6 @@ type PolicyInformer interface {

 	Lister() client.SyncedPoliciesListerNamespacer

 }

-func (f *sharedInformerFactory) Policies() PolicyInformer {

-	return &policyInformer{sharedInformerFactory: f}

-}

-

 type policyInformer struct {

 	*sharedInformerFactory

 }

@@ -195,10 +183,6 @@ type PolicyBindingInformer interface {

 	Lister() client.SyncedPolicyBindingsListerNamespacer

 }

-func (f *sharedInformerFactory) PolicyBindings() PolicyBindingInformer {

-	return &policyBindingInformer{sharedInformerFactory: f}

-}

-

 type policyBindingInformer struct {

 	*sharedInformerFactory

 }

new file mode 100644

@@ -0,0 +1,62 @@

+package controller

+

+import (

+	"reflect"

+

+	kapi "k8s.io/kubernetes/pkg/api"

+	"k8s.io/kubernetes/pkg/client/cache"

+	"k8s.io/kubernetes/pkg/controller/framework"

+	"k8s.io/kubernetes/pkg/runtime"

+	"k8s.io/kubernetes/pkg/watch"

+

+	oscache "github.com/openshift/origin/pkg/client/cache"

+)

+

+type ImageStreamInformer interface {

+	Informer() framework.SharedIndexInformer

+	Indexer() cache.Indexer

+	Lister() *oscache.StoreToImageStreamLister

+}

+

+type imageStreamInformer struct {

+	*sharedInformerFactory

+}

+

+func (f *imageStreamInformer) Informer() framework.SharedIndexInformer {

+	f.lock.Lock()

+	defer f.lock.Unlock()

+

+	informerObj := &imageapi.ImageStream{}

+	informerType := reflect.TypeOf(informerObj)

+	informer, exists := f.informers[informerType]

+	if exists {

+		return informer

+	}

+

+	informer = framework.NewSharedIndexInformer(

+		&cache.ListWatch{

+			ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

+				return f.originClient.ImageStreams(kapi.NamespaceAll).List(options)

+			},

+			WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

+				return f.originClient.ImageStreams(kapi.NamespaceAll).Watch(options)

+			},

+		},

+		informerObj,

+		f.defaultResync,

+		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},

+	)

+	f.informers[informerType] = informer

+

+	return informer

+}

+

+func (f *imageStreamInformer) Indexer() cache.Indexer {

+	informer := f.Informer()

+	return informer.GetIndexer()

+}

+

+func (f *imageStreamInformer) Lister() *oscache.StoreToImageStreamLister {

+	informer := f.Informer()

+	return &oscache.StoreToImageStreamLister{Indexer: informer.GetIndexer()}

+}

@@ -16,10 +16,6 @@ type PodInformer interface {

 	Lister() *cache.StoreToPodLister

 }

-func (f *sharedInformerFactory) Pods() PodInformer {

-	return &podInformer{sharedInformerFactory: f}

-}

-

 type podInformer struct {

 	*sharedInformerFactory

 }

@@ -75,10 +71,6 @@ type ReplicationControllerInformer interface {

 	Lister() *cache.StoreToReplicationControllerLister

 }

-func (f *sharedInformerFactory) ReplicationControllers() ReplicationControllerInformer {

-	return &replicationControllerInformer{sharedInformerFactory: f}

-}

-

 type replicationControllerInformer struct {

 	*sharedInformerFactory

 }

@@ -28,6 +28,7 @@ type InformerFactory interface {

 	PolicyBindings() PolicyBindingInformer

 	DeploymentConfigs() DeploymentConfigInformer

+	ImageStreams() ImageStreamInformer

 }

 // ListerWatcherOverrides allows a caller to specify special behavior for particular ListerWatchers

@@ -73,12 +74,40 @@ func (f *sharedInformerFactory) Start(stopCh <-chan struct{}) {

 	}

 }

-func (f *sharedInformerFactory) DeploymentConfigs() DeploymentConfigInformer {

-	return &deploymentConfigInformer{sharedInformerFactory: f}

-}

-

 func (f *sharedInformerFactory) StartCore(stopCh <-chan struct{}) {

 	for _, informer := range f.coreInformers {

 		go informer.Run(stopCh)

 	}

 }

+

+func (f *sharedInformerFactory) Pods() PodInformer {

+	return &podInformer{sharedInformerFactory: f}

+}

+

+func (f *sharedInformerFactory) ReplicationControllers() ReplicationControllerInformer {

+	return &replicationControllerInformer{sharedInformerFactory: f}

+}

+

+func (f *sharedInformerFactory) ClusterPolicies() ClusterPolicyInformer {

+	return &clusterPolicyInformer{sharedInformerFactory: f}

+}

+

+func (f *sharedInformerFactory) ClusterPolicyBindings() ClusterPolicyBindingInformer {

+	return &clusterPolicyBindingInformer{sharedInformerFactory: f}

+}

+

+func (f *sharedInformerFactory) Policies() PolicyInformer {

+	return &policyInformer{sharedInformerFactory: f}

+}

+

+func (f *sharedInformerFactory) PolicyBindings() PolicyBindingInformer {

+	return &policyBindingInformer{sharedInformerFactory: f}

+}

+

+func (f *sharedInformerFactory) DeploymentConfigs() DeploymentConfigInformer {

+	return &deploymentConfigInformer{sharedInformerFactory: f}

+}

+

+func (f *sharedInformerFactory) ImageStreams() ImageStreamInformer {

+	return &imageStreamInformer{sharedInformerFactory: f}

+}

@@ -1,7 +1,6 @@

 package deploymentconfig

 import (

-	"fmt"

 	"time"

 	"github.com/golang/glog"

@@ -18,11 +17,13 @@ import (

 	osclient "github.com/openshift/origin/pkg/client"

 	deployapi "github.com/openshift/origin/pkg/deploy/api"

+	deployutil "github.com/openshift/origin/pkg/deploy/util"

 )

 const (

-	// We must avoid creating new replication controllers until the {replication controller,pods} store

-	// has synced. If it hasn't synced, to avoid a hot loop, we'll wait this long between checks.

+	// We must avoid creating new replication controllers until the deployment config and replication

+	// controller stores have synced. If it hasn't synced, to avoid a hot loop, we'll wait this long

+	// between checks.

 	StoreSyncedPollPeriod = 100 * time.Millisecond

 )

@@ -69,7 +70,7 @@ func (c *DeploymentConfigController) Run(workers int, stopCh <-chan struct{}) {

 	// Wait for the rc and dc stores to sync before starting any work in this controller.

 	ready := make(chan struct{})

-	go c.waitForSyncedStores(ready)

+	go c.waitForSyncedStores(ready, stopCh)

 	select {

 	case <-ready:

 	case <-stopCh:

@@ -85,12 +86,16 @@ func (c *DeploymentConfigController) Run(workers int, stopCh <-chan struct{}) {

 	c.queue.ShutDown()

 }

-func (c *DeploymentConfigController) waitForSyncedStores(ready chan struct{}) {

+func (c *DeploymentConfigController) waitForSyncedStores(ready chan<- struct{}, stopCh <-chan struct{}) {

 	defer utilruntime.HandleCrash()

 	for !c.dcStoreSynced() || !c.rcStoreSynced() || !c.podStoreSynced() {

 		glog.V(4).Infof("Waiting for the dc, rc, and pod caches to sync before starting the deployment config controller workers")

-		time.Sleep(StoreSyncedPollPeriod)

+		select {

+		case <-time.After(StoreSyncedPollPeriod):

+		case <-stopCh:

+			return

+		}

 	}

 	close(ready)

 }

@@ -102,9 +107,9 @@ func (c *DeploymentConfigController) addDeploymentConfig(obj interface{}) {

 }

 func (c *DeploymentConfigController) updateDeploymentConfig(old, cur interface{}) {

-	oldDc := old.(*deployapi.DeploymentConfig)

-	glog.V(4).Infof("Updating deployment config %q", oldDc.Name)

-	c.enqueueDeploymentConfig(cur.(*deployapi.DeploymentConfig))

+	dc := cur.(*deployapi.DeploymentConfig)

+	glog.V(4).Infof("Updating deployment config %q", dc.Name)

+	c.enqueueDeploymentConfig(dc)

 }

 func (c *DeploymentConfigController) deleteDeploymentConfig(obj interface{}) {

@@ -125,7 +130,10 @@ func (c *DeploymentConfigController) deleteDeploymentConfig(obj interface{}) {

 	c.enqueueDeploymentConfig(dc)

 }

-// addReplicationController enqueues the deployment that manages a replicationcontroller when the replicationcontroller is created.

+// addReplicationController figures out which deploymentconfig is managing this replication

+// controller and requeues the deployment config.

+// TODO: Determine if we need to resync here. Would be useful for adoption but we cannot

+// adopt right now.

 func (c *DeploymentConfigController) addReplicationController(obj interface{}) {

 	rc := obj.(*kapi.ReplicationController)

 	glog.V(4).Infof("Replication controller %q added.", rc.Name)

@@ -211,7 +219,7 @@ func (c *DeploymentConfigController) work() bool {

 		return false

 	}

-	copied, err := dcCopy(dc)

+	copied, err := deployutil.DeploymentConfigDeepCopy(dc)

 	if err != nil {

 		glog.Error(err.Error())

 		return false

@@ -237,15 +245,3 @@ func (c *DeploymentConfigController) getByKey(key string) (*deployapi.Deployment

 	return obj.(*deployapi.DeploymentConfig), nil

 }

-

-func dcCopy(dc *deployapi.DeploymentConfig) (*deployapi.DeploymentConfig, error) {

-	objCopy, err := kapi.Scheme.DeepCopy(dc)

-	if err != nil {

-		return nil, err

-	}

-	copied, ok := objCopy.(*deployapi.DeploymentConfig)

-	if !ok {

-		return nil, fmt.Errorf("expected DeploymentConfig, got %#v", objCopy)

-	}

-	return copied, nil

-}

@@ -6,8 +6,10 @@ import (

 	kapi "k8s.io/kubernetes/pkg/api"

 	kclient "k8s.io/kubernetes/pkg/client/unversioned"

 	"k8s.io/kubernetes/pkg/runtime"

+	"k8s.io/kubernetes/pkg/util/workqueue"

 	osclient "github.com/openshift/origin/pkg/client"

+	oscache "github.com/openshift/origin/pkg/client/cache"

 	deployapi "github.com/openshift/origin/pkg/deploy/api"

 	deployutil "github.com/openshift/origin/pkg/deploy/util"

 )

@@ -19,19 +21,19 @@ type DeploymentTriggerController struct {

 	dn osclient.DeploymentConfigsNamespacer

 	// rn is used for getting the latest deployment for a config.

 	rn kclient.ReplicationControllersNamespacer

+

+	// queue contains deployment configs that need to be synced.

+	queue workqueue.RateLimitingInterface

+

+	// dcStore provides a local cache for deployment configs.

+	dcStore oscache.StoreToDeploymentConfigLister

+	// dcStoreSynced makes sure the dc store is synced before reconcling any deployment config.

+	dcStoreSynced func() bool

+

 	// codec is used for decoding a config out of a deployment.

 	codec runtime.Codec

 }

-// NewDeploymentTriggerController returns a new DeploymentTriggerController.

-func NewDeploymentTriggerController(oc osclient.Interface, kc kclient.Interface, codec runtime.Codec) *DeploymentTriggerController {

-	return &DeploymentTriggerController{

-		dn:    oc,

-		rn:    kc,

-		codec: codec,

-	}

-}

-

 // fatalError is an error which can't be retried.

 type fatalError string

@@ -59,7 +61,12 @@ func (c *DeploymentTriggerController) Handle(config *deployapi.DeploymentConfig)

 		return nil

 	}

-	return c.update(config, causes)

+	copied, err := deployutil.DeploymentConfigDeepCopy(config)

+	if err != nil {

+		return err

+	}

+

+	return c.update(copied, causes)

 }

 // decodeFromLatest will try to return the decoded version of the current deploymentconfig found

@@ -2,19 +2,54 @@ package generictrigger

 import (

 	"testing"

+	"time"

 	kapi "k8s.io/kubernetes/pkg/api"

+	"k8s.io/kubernetes/pkg/client/cache"

 	ktestclient "k8s.io/kubernetes/pkg/client/unversioned/testclient"

+	"k8s.io/kubernetes/pkg/controller/framework"

 	"k8s.io/kubernetes/pkg/runtime"

+	"k8s.io/kubernetes/pkg/watch"

 	"github.com/openshift/origin/pkg/client/testclient"

 	deployapi "github.com/openshift/origin/pkg/deploy/api"

 	_ "github.com/openshift/origin/pkg/deploy/api/install"

 	testapi "github.com/openshift/origin/pkg/deploy/api/test"

 	deployutil "github.com/openshift/origin/pkg/deploy/util"

+	imageapi "github.com/openshift/origin/pkg/image/api"

 )

-var codec = kapi.Codecs.LegacyCodec(deployapi.SchemeGroupVersion)

+var (

+	codec      = kapi.Codecs.LegacyCodec(deployapi.SchemeGroupVersion)

+	mock       = &testclient.Fake{}

+	dcInformer = framework.NewSharedIndexInformer(

+		&cache.ListWatch{

+			ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

+				return mock.DeploymentConfigs(kapi.NamespaceAll).List(options)

+			},

+			WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

+				return mock.DeploymentConfigs(kapi.NamespaceAll).Watch(options)

+			},

+		},

+		&deployapi.DeploymentConfig{},

+		2*time.Minute,

+		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},

+	)

+

+	streamInformer = framework.NewSharedIndexInformer(

+		&cache.ListWatch{

+			ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

+				return mock.ImageStreams(kapi.NamespaceAll).List(options)

+			},

+			WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

+				return mock.ImageStreams(kapi.NamespaceAll).Watch(options)

+			},

+		},

+		&imageapi.ImageStream{},

+		2*time.Minute,

+		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},

+	)

+)

 // TestHandle_newConfigNoTriggers ensures that a change to a config with no

 // triggers doesn't result in a new config version bump.

@@ -22,7 +57,7 @@ func TestHandle_newConfigNoTriggers(t *testing.T) {

 	fake := &testclient.Fake{}

 	kFake := &ktestclient.Fake{}

-	controller := NewDeploymentTriggerController(fake, kFake, codec)

+	controller := NewDeploymentTriggerController(dcInformer, streamInformer, fake, kFake, codec)

 	config := testapi.OkDeploymentConfig(1)

 	config.Namespace = kapi.NamespaceDefault

@@ -51,7 +86,7 @@ func TestHandle_newConfigTriggers(t *testing.T) {

 	})

 	kFake := &ktestclient.Fake{}

-	controller := NewDeploymentTriggerController(fake, kFake, codec)

+	controller := NewDeploymentTriggerController(dcInformer, streamInformer, fake, kFake, codec)

 	config := testapi.OkDeploymentConfig(0)

 	config.Namespace = kapi.NamespaceDefault

@@ -123,7 +158,7 @@ func TestHandle_changeWithTemplateDiff(t *testing.T) {

 			return true, deployment, nil

 		})

-		controller := NewDeploymentTriggerController(fake, kFake, codec)

+		controller := NewDeploymentTriggerController(dcInformer, streamInformer, fake, kFake, codec)

 		s.modify(config)

 		if err := controller.Handle(config); err != nil {

@@ -169,7 +204,7 @@ func TestHandle_waitForImageController(t *testing.T) {

 		return true, nil, nil

 	})

-	controller := NewDeploymentTriggerController(fake, kFake, codec)

+	controller := NewDeploymentTriggerController(dcInformer, streamInformer, fake, kFake, codec)

 	config := testapi.OkDeploymentConfig(0)

 	config.Namespace = kapi.NamespaceDefault

@@ -236,7 +271,7 @@ func TestHandle_automaticImageUpdates(t *testing.T) {

 			return true, deployment, nil

 		})

-		controller := NewDeploymentTriggerController(fake, kFake, codec)

+		controller := NewDeploymentTriggerController(dcInformer, streamInformer, fake, kFake, codec)

 		config := testapi.OkDeploymentConfig(test.version)

 		config.Namespace = kapi.NamespaceDefault

@@ -4,67 +4,185 @@ import (

 	"time"

 	kapi "k8s.io/kubernetes/pkg/api"

-	"k8s.io/kubernetes/pkg/client/cache"

-	"k8s.io/kubernetes/pkg/client/record"

 	kclient "k8s.io/kubernetes/pkg/client/unversioned"

+	kcontroller "k8s.io/kubernetes/pkg/controller"

+	"k8s.io/kubernetes/pkg/controller/framework"

 	"k8s.io/kubernetes/pkg/runtime"

-	"k8s.io/kubernetes/pkg/util/flowcontrol"

 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"

-	"k8s.io/kubernetes/pkg/watch"

+	"k8s.io/kubernetes/pkg/util/wait"

+	"k8s.io/kubernetes/pkg/util/workqueue"

+	"github.com/golang/glog"

 	osclient "github.com/openshift/origin/pkg/client"

-	controller "github.com/openshift/origin/pkg/controller"

 	deployapi "github.com/openshift/origin/pkg/deploy/api"

+	imageapi "github.com/openshift/origin/pkg/image/api"

 )

-// DeploymentTriggerControllerFactory can create a DeploymentTriggerController that watches all DeploymentConfigs.

-type DeploymentTriggerControllerFactory struct {

-	// Client is an OpenShift client.

-	Client osclient.Interface

-	// KubeClient is a Kubernetes client.

-	KubeClient kclient.Interface

-	// Codec is used for encoding/decoding.

-	Codec runtime.Codec

+const (

+	// We must avoid creating processing deployment configs until the deployment config and image

+	// stream stores have synced. If it hasn't synced, to avoid a hot loop, we'll wait this long

+	// between checks.

+	StoreSyncedPollPeriod = 100 * time.Millisecond

+)

+

+// NewDeploymentTriggerController returns a new DeploymentTriggerController.

+func NewDeploymentTriggerController(dcInformer, streamInformer framework.SharedIndexInformer, oc osclient.Interface, kc kclient.Interface, codec runtime.Codec) *DeploymentTriggerController {

+	c := &DeploymentTriggerController{

+		dn: oc,

+		rn: kc,

+

+		queue: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),

+

+		codec: codec,

+	}

+

+	c.dcStore.Indexer = dcInformer.GetIndexer()

+	dcInformer.AddEventHandler(framework.ResourceEventHandlerFuncs{

+		AddFunc:    c.addDeploymentConfig,

+		UpdateFunc: c.updateDeploymentConfig,

+	})

+	c.dcStoreSynced = dcInformer.HasSynced

+

+	streamInformer.AddEventHandler(framework.ResourceEventHandlerFuncs{

+		AddFunc:    c.addImageStream,

+		UpdateFunc: c.updateImageStream,

+	})

+

+	return c

+}

+

+// Run begins watching and syncing.

+func (c *DeploymentTriggerController) Run(workers int, stopCh <-chan struct{}) {

+	defer utilruntime.HandleCrash()

+

+	// Wait for the dc store to sync before starting any work in this controller.

+	ready := make(chan struct{})

+	go c.waitForSyncedStore(ready, stopCh)

+	select {

+	case <-ready:

+	case <-stopCh:

+		return

+	}

+

+	for i := 0; i < workers; i++ {

+		go wait.Until(c.worker, time.Second, stopCh)

+	}

+	<-stopCh

+	glog.Infof("Shutting down deployment trigger controller")

+	c.queue.ShutDown()

+}

+

+func (c *DeploymentTriggerController) waitForSyncedStore(ready chan<- struct{}, stopCh <-chan struct{}) {

+	defer utilruntime.HandleCrash()

+

+	for !c.dcStoreSynced() {

+		glog.V(4).Infof("Waiting for the deployment config cache to sync before starting the trigger controller workers")

+		select {

+		case <-time.After(StoreSyncedPollPeriod):

+		case <-stopCh:

+			return

+		}

+	}

+	close(ready)

+}

+

+func (c *DeploymentTriggerController) addDeploymentConfig(obj interface{}) {

+	dc := obj.(*deployapi.DeploymentConfig)

+	c.enqueueDeploymentConfig(dc)

+}

+

+func (c *DeploymentTriggerController) updateDeploymentConfig(old, cur interface{}) {

+	dc := cur.(*deployapi.DeploymentConfig)

+	c.enqueueDeploymentConfig(dc)

+}

+

+// addImageStream enqueues the deployment configs that point to the new image stream.

+func (c *DeploymentTriggerController) addImageStream(obj interface{}) {

+	stream := obj.(*imageapi.ImageStream)

+	glog.V(4).Infof("Image stream %q added.", stream.Name)

+	dcList, err := c.dcStore.GetConfigsForImageStream(stream)

+	if err != nil {

+		return

+	}

+	for _, dc := range dcList {

+		c.enqueueDeploymentConfig(dc)

+	}

+}

+

+// updateImageStream enqueues the deployment configs that point to the updated image stream.

+func (c *DeploymentTriggerController) updateImageStream(old, cur interface{}) {

+	// A periodic relist will send update events for all known streams.

+	if kapi.Semantic.DeepEqual(old, cur) {

+		return

+	}

+

+	stream := cur.(*imageapi.ImageStream)

+	glog.V(4).Infof("Image stream %q updated.", stream.Name)

+	dcList, err := c.dcStore.GetConfigsForImageStream(stream)

+	if err != nil {

+		return

+	}

+	for _, dc := range dcList {

+		c.enqueueDeploymentConfig(dc)

+	}

+}

+

+func (c *DeploymentTriggerController) enqueueDeploymentConfig(dc *deployapi.DeploymentConfig) {

+	key, err := kcontroller.KeyFunc(dc)

+	if err != nil {

+		glog.Errorf("Couldn't get key for object %+v: %v", dc, err)

+		return

+	}

+	c.queue.Add(key)

+}

+

+func (c *DeploymentTriggerController) worker() {

+	for {

+		if quit := c.work(); quit {

+			return

+		}

+	}

 }

-// Create creates a DeploymentTriggerController.

-func (factory *DeploymentTriggerControllerFactory) Create() controller.RunnableController {

-	deploymentConfigLW := &cache.ListWatch{

-		ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

-			return factory.Client.DeploymentConfigs(kapi.NamespaceAll).List(options)

-		},

-		WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

-			return factory.Client.DeploymentConfigs(kapi.NamespaceAll).Watch(options)

-		},

-	}

-	queue := cache.NewFIFO(cache.MetaNamespaceKeyFunc)

-	cache.NewReflector(deploymentConfigLW, &deployapi.DeploymentConfig{}, queue, 2*time.Minute).Run()

-

-	eventBroadcaster := record.NewBroadcaster()

-	eventBroadcaster.StartRecordingToSink(factory.KubeClient.Events(""))

-

-	triggerController := NewDeploymentTriggerController(factory.Client, factory.KubeClient, factory.Codec)

-

-	return &controller.RetryController{

-		Queue: queue,

-		RetryManager: controller.NewQueueRetryManager(

-			queue,

-			cache.MetaNamespaceKeyFunc,

-			func(obj interface{}, err error, retries controller.Retry) bool {

-				utilruntime.HandleError(err)

-				if _, isFatal := err.(fatalError); isFatal {

-					return false

-				}

-				if retries.Count > 0 {

-					return false

-				}

-				return true

-			},

-			flowcontrol.NewTokenBucketRateLimiter(1, 10),

-		),

-		Handle: func(obj interface{}) error {

-			config := obj.(*deployapi.DeploymentConfig)

-			return triggerController.Handle(config)

-		},

+func (c *DeploymentTriggerController) work() bool {

+	key, quit := c.queue.Get()

+	if quit {

+		return true

 	}

+	defer c.queue.Done(key)

+

+	dc, err := c.getByKey(key.(string))

+	if err != nil {

+		glog.Error(err.Error())

+	}

+

+	if dc == nil {

+		return false

+	}

+

+	if err := c.Handle(dc); err != nil {

+		utilruntime.HandleError(err)

+

+		if c.queue.NumRequeues(key) < 2 {

+			c.queue.AddRateLimited(key)

+			return false

+		}

+	}

+	c.queue.Forget(key)

+	return false

+}

+

+func (c *DeploymentTriggerController) getByKey(key string) (*deployapi.DeploymentConfig, error) {

+	obj, exists, err := c.dcStore.Indexer.GetByKey(key)

+	if err != nil {

+		glog.Infof("Unable to retrieve deployment config %q from store: %v", key, err)

+		c.queue.Add(key)

+		return nil, err

+	}

+	if !exists {

+		glog.Infof("Deployment config %q has been deleted", key)

+		return nil, nil

+	}

+

+	return obj.(*deployapi.DeploymentConfig), nil

 }

@@ -105,6 +105,18 @@ func HasChangeTrigger(config *deployapi.DeploymentConfig) bool {

 	return false

 }

+func DeploymentConfigDeepCopy(dc *deployapi.DeploymentConfig) (*deployapi.DeploymentConfig, error) {

+	objCopy, err := api.Scheme.DeepCopy(dc)

+	if err != nil {

+		return nil, err

+	}

+	copied, ok := objCopy.(*deployapi.DeploymentConfig)

+	if !ok {

+		return nil, fmt.Errorf("expected DeploymentConfig, got %#v", objCopy)

+	}

+	return copied, nil

+}

+

 // DecodeDeploymentConfig decodes a DeploymentConfig from controller using codec. An error is returned

 // if the controller doesn't contain an encoded config.

 func DecodeDeploymentConfig(controller *api.ReplicationController, decoder runtime.Decoder) (*deployapi.DeploymentConfig, error) {