cshared: Make goroutine for Collect to be able to exit

Signed-off-by: Hiroshi Hatake <[email protected]>

cshared.go

@@ -201,8 +201,16 @@ func prepareInputCollector() (err error) {
 
 	go func(theChannel chan<- Message) {
 		defer theInputLock.Unlock()
+		for {
+			select {
+			case <-runCtx.Done():
+				log.Printf("goroutine will be stopping: name=%q\n", theName)
+				return
+			default:
+				err = theInput.Collect(runCtx, theChannel)
+			}
+		}
 
-		err := theInput.Collect(runCtx, theChannel)
 		if err != nil {
 			fmt.Fprintf(os.Stderr,
 				"collect error: %s\n", err.Error())

cshared_test.go

@@ -93,17 +93,16 @@ func TestInputCallbackDangle(t *testing.T) {
 	cdone := make(chan bool)
 	ptr := unsafe.Pointer(nil)
 
+	// prepare channel for input explicitly.
+	err := prepareInputCollector()
+	if err != nil {
+		t.Fail()
+	}
+
 	go func(test *testing.T) {
 		t := time.NewTicker(collectInterval)
 		defer t.Stop()
 
-		// prepare channel for input explicitly.
-		err := prepareInputCollector()
-		if err != nil {
-			test.Fail()
-			return
-		}
-
 		FLBPluginInputCallback(&ptr, nil)
 		for {
 			select {