Revert "add stopChan to storeWatcher.Run()"

This reverts commit d90ccfc.

store/app_service_store_integration_test.go

@@ -13,7 +13,6 @@ var _ = Describe("AppServiceStoreIntegration", func() {
 		incomingChan  chan appservice.AppServices
 		outAddChan    <-chan appservice.AppService
 		outRemoveChan <-chan appservice.AppService
-		stopChan chan struct{}
 	)
 
 	BeforeEach(func() {
@@ -23,17 +22,12 @@ var _ = Describe("AppServiceStoreIntegration", func() {
 		c := cache.NewAppServiceCache()
 		var watcher *AppServiceStoreWatcher
 		watcher, outAddChan, outRemoveChan = NewAppServiceStoreWatcher(adapter, c)
-		stopChan = make(chan struct{})
-		go watcher.Run(stopChan)
+		go watcher.Run()
 
 		store := NewAppServiceStore(adapter, watcher)
 		go store.Run(incomingChan)
 	})
 
-	AfterEach(func() {
-		close(stopChan)
-	})
-
 	It("should receive, store, and republish AppServices", func(done Done) {
 		appServices := appservice.AppServices{AppId: "12345", Urls: []string{"syslog://foo"}}
 		incomingChan <- appServices

store/app_service_store_watcher.go

@@ -56,7 +56,7 @@ func (w *AppServiceStoreWatcher) Exists(appService appservice.AppService) bool {
 	return w.cache.Exists(appService)
 }
 
-func (w *AppServiceStoreWatcher) Run(outerStopChan <-chan struct{}) {
+func (w *AppServiceStoreWatcher) Run() {
 	defer func() {
 		close(w.outAddChan)
 		close(w.outRemoveChan)
@@ -70,17 +70,17 @@ func (w *AppServiceStoreWatcher) Run(outerStopChan <-chan struct{}) {
 	InnerLoop:
 		for {
 			select {
-			case err, ok := <-errChan:
-				if !ok {
-					break InnerLoop
+			case err := <-errChan:
+				if err == nil {
+					return
 				}
 				cfcomponent.Logger.Errorf("AppStoreWatcher: Got error while waiting for ETCD events: %s", err.Error())
 				close(stopChan)
 				break InnerLoop
 
 			case event, ok := <-events:
 				if !ok {
-					break InnerLoop
+					return
 				}
 
 				cfcomponent.Logger.Debugf("AppStoreWatcher: Got an event from store %s", event.Type)
@@ -96,12 +96,11 @@ func (w *AppServiceStoreWatcher) Run(outerStopChan <-chan struct{}) {
 				case storeadapter.ExpireEvent:
 					w.deleteEvent(*(event.PrevNode))
 				}
-
-			case <-outerStopChan:
-				return
 			}
 		}
+
 	}
+
 }
 
 func (w *AppServiceStoreWatcher) registerExistingServicesFromStore() {

store/app_service_store_watcher_test.go

@@ -23,7 +23,6 @@ var _ = Describe("AppServiceStoreWatcher", func() {
 	var adapter storeadapter.StoreAdapter
 	var outAddChan <-chan appservice.AppService
 	var outRemoveChan <-chan appservice.AppService
-	var stopChan chan struct{}
 
 	var app1Service1 appservice.AppService
 	var app1Service2 appservice.AppService
@@ -49,7 +48,6 @@ var _ = Describe("AppServiceStoreWatcher", func() {
 
 		c := cache.NewAppServiceCache()
 		watcher, outAddChan, outRemoveChan = NewAppServiceStoreWatcher(adapter, c)
-		stopChan = make(chan struct{})
 	})
 
 	AfterEach(func() {
@@ -60,11 +58,10 @@ var _ = Describe("AppServiceStoreWatcher", func() {
 
 	Describe("Shutdown", func() {
 		It("should close the outgoing channels", func() {
-			go watcher.Run(stopChan)
+			go watcher.Run()
 			time.Sleep(500 * time.Millisecond)
 			adapter.Disconnect()
 
-			close(stopChan)
 			Eventually(outRemoveChan).Should(BeClosed())
 			Eventually(outAddChan).Should(BeClosed())
 		})
@@ -73,8 +70,7 @@ var _ = Describe("AppServiceStoreWatcher", func() {
 	Describe("Loading watcher state on startup", func() {
 		Context("when the store is empty", func() {
 			It("should not send anything on the output channels", func() {
-				go watcher.Run(stopChan)
-				defer close(stopChan)
+				go watcher.Run()
 
 				Consistently(outAddChan).Should(BeEmpty())
 				Consistently(outRemoveChan).Should(BeEmpty())
@@ -89,8 +85,7 @@ var _ = Describe("AppServiceStoreWatcher", func() {
 			})
 
 			It("should send all the AppServices on the output add channel", func(done Done) {
-				go watcher.Run(stopChan)
-				defer close(stopChan)
+				go watcher.Run()
 
 				appServices := drainOutgoingChannel(outAddChan, 3)
 
@@ -106,23 +101,16 @@ var _ = Describe("AppServiceStoreWatcher", func() {
 	})
 
 	Describe("when the store has data and watcher is bootstrapped", func() {
-		var stopChan chan struct{}
-
 		BeforeEach(func(done Done) {
 			adapter.Create(buildNode(app1Service1))
 			adapter.Create(buildNode(app1Service2))
 			adapter.Create(buildNode(app2Service1))
 
-			stopChan = make(chan struct{})
-			go watcher.Run(stopChan)
+			go watcher.Run()
 			drainOutgoingChannel(outAddChan, 3)
 			close(done)
 		})
 
-		AfterEach(func() {
-			close(stopChan)
-		})
-
 		It("does not send updates when the data has already been processed", func(done Done) {
 			adapter.Create(buildNode(app1Service1))
 			adapter.Create(buildNode(app1Service2))

store/app_service_store_watcher_unit_test.go

@@ -11,18 +11,12 @@ import (
 var _ = Describe("AppServiceStoreWatcherUnit", func() {
 	Context("when there is an error", func() {
 		var adapter *FakeAdapter
-		var stopChan chan struct{}
 
 		BeforeEach(func() {
 			adapter = &FakeAdapter{}
 			watcher, _, _ := NewAppServiceStoreWatcher(adapter, cache.NewAppServiceCache())
 
-			stopChan = make(chan struct{})
-			go watcher.Run(stopChan)
-		})
-
-		AfterEach(func() {
-			close(stopChan)
+			go watcher.Run()
 		})
 
 		It("calls watch again", func() {