Fix test to not sleep

pkg/extractor/batchers/fileBatcher.go

@@ -33,7 +33,6 @@ func OpenFilesToChan(filenames <-chan string, gunzip bool, concurrency int, batc
 					wg.Done()
 					out.stopFileReading(goFilename)
 				}()
-				out.startFileReading(goFilename)
 
 				var file io.ReadCloser
 				file, err := openFileToReader(goFilename, gunzip)
@@ -44,6 +43,7 @@ func OpenFilesToChan(filenames <-chan string, gunzip bool, concurrency int, batc
 				}
 				defer file.Close()
 
+				out.startFileReading(goFilename)
 				out.syncReaderToBatcher(goFilename, file, batchSize)
 			}(filename)
 		}

pkg/extractor/batchers/tailBatcher.go

@@ -21,7 +21,6 @@ func TailFilesToChan(filenames <-chan string, batchSize int, reopen, poll, tail
 					out.stopFileReading(filename)
 				}()
 
-				out.startFileReading(filename)
 				r, err := followreader.New(filename, reopen, poll)
 				if err != nil {
 					logger.Print("Unable to open file: ", err)
@@ -35,6 +34,8 @@ func TailFilesToChan(filenames <-chan string, batchSize int, reopen, poll, tail
 					}
 				}
 
+				out.startFileReading(filename)
+
 				out.syncReaderToBatcherWithTimeFlush(filename, r, batchSize, AutoFlushTimeout)
 			}(filename)
 		}

pkg/extractor/batchers/tailBatcher_test.go

@@ -40,7 +40,9 @@ func TestBatchFollowTailFile(t *testing.T) {
 
 	batcher := TailFilesToChan(filenames, 1, false, false, true)
 
-	time.Sleep(300 * time.Millisecond) // Uhg hack cause auto-flushing
+	for len(batcher.activeFiles) == 0 {
+		time.Sleep(time.Millisecond) // Semi-hack: Wait for the go-routine reader to start and the source to be drained
+	}
 
 	// And write some more data
 	const testLines = 5
@@ -58,4 +60,6 @@ func TestBatchFollowTailFile(t *testing.T) {
 			assert.Len(t, batch.Batch, 1)
 		}
 	}
+
+	assert.Len(t, batcher.BatchChan(), 0)
 }