Simplify stopping of deferred executor.

deferred_executor.go

@@ -33,8 +33,7 @@ import (
 // their order.
 type DeferredExecutor struct {
 	queue     chan func()
-	closeChan chan bool
+	closed    chan struct{}
-	closed    chan bool
 	closeOnce sync.Once
 }
 
@@ -43,28 +42,24 @@ func NewDeferredExecutor(queueSize int) *DeferredExecutor {
 		queueSize = 0
 	}
 	result := &DeferredExecutor{
-		queue:     make(chan func(), queueSize),
+		queue:  make(chan func(), queueSize),
-		closeChan: make(chan bool, 1),
+		closed: make(chan struct{}),
-		closed:    make(chan bool, 1),
 	}
 	go result.run()
 	return result
 }
 
 func (e *DeferredExecutor) run() {
-loop:
+	defer close(e.closed)
+
 	for {
-		select {
+		f := <-e.queue
-		case f := <-e.queue:
+		if f == nil {
-			if f == nil {
+			break
-				break loop
-			}
-			f()
-		case <-e.closeChan:
-			break loop
 		}
+
+		f()
 	}
-	e.closed <- true
 }
 
 func getFunctionName(i interface{}) string {
@@ -83,14 +78,9 @@ func (e *DeferredExecutor) Execute(f func()) {
 }
 
 func (e *DeferredExecutor) Close() {
-	select {
+	e.closeOnce.Do(func() {
-	case e.closeChan <- true:
+		close(e.queue)
-		e.closeOnce.Do(func() {
+	})
-			close(e.queue)
-		})
-	default:
-		// Already closed.
-	}
 }
 
 func (e *DeferredExecutor) waitForStop() {

deferred_executor_test.go

@@ -109,3 +109,12 @@ func TestDeferredExecutor_DeferAfterClose(t *testing.T) {
 		t.Error("method should not have been called")
 	})
 }
+
+func TestDeferredExecutor_WaitForStopTwice(t *testing.T) {
+	e := NewDeferredExecutor(64)
+	defer e.waitForStop()
+
+	e.Close()
+
+	e.waitForStop()
+}