FileInput should have own limiter algorithm

input_file.go

@@ -8,15 +8,17 @@ import (
 )
 
 type FileInput struct {
-	data    chan []byte
-	path    string
-	decoder *gob.Decoder
+	data        chan []byte
+	path        string
+	decoder     *gob.Decoder
+	speedFactor float64
 }
 
 func NewFileInput(path string) (i *FileInput) {
 	i = new(FileInput)
 	i.data = make(chan []byte)
 	i.path = path
+	i.speedFactor = 1
 	i.Init(path)
 
 	go i.emit()
@@ -57,8 +59,16 @@ func (i *FileInput) emit() {
 		}
 
 		if lastTime != 0 {
-			time.Sleep(time.Duration(raw.Timestamp - lastTime))
+			timeDiff := raw.Timestamp - lastTime
+
+			// We can speedup or slowdown execution based on speedFactor
+			if i.speedFactor != 1 {
+				timeDiff = int64(float64(raw.Timestamp-lastTime) / i.speedFactor)
+			}
+
+			time.Sleep(time.Duration(timeDiff))
 		}
+
 		lastTime = raw.Timestamp
 
 		i.data <- raw.Request

limiter.go

@@ -36,10 +36,20 @@ func NewLimiter(plugin interface{}, options string) io.ReadWriter {
 	l.plugin = plugin
 	l.currentTime = time.Now().UnixNano()
 
+	// FileInput have its own rate limiting. Unlike other inputs we not just dropping requests, we can slow down or speed up request emittion.
+	if fi, ok := l.plugin.(*FileInput); ok && l.isPercent {
+		fi.speedFactor = float64(l.limit) / float64(100)
+	}
+
 	return l
 }
 
 func (l *Limiter) isLimited() bool {
+	// File input have its own limiting algorithm
+	if _, ok := l.plugin.(*FileInput); ok {
+		return false
+	}
+
 	if l.isPercent {
 		return l.limit <= rand.Intn(100)
 	}
@@ -71,9 +81,9 @@ func (l *Limiter) Write(data []byte) (n int, err error) {
 func (l *Limiter) Read(data []byte) (n int, err error) {
 	n, err = l.plugin.(io.Reader).Read(data)
 
-	if l.isLimited() {		
+	if l.isLimited() {
 		return 0, nil
-	}	
+	}
 
 	return
 }

limiter_test.go

@@ -1,3 +1,5 @@
+// +build !race
+
 package main
 
 import (