diff --git a/input_file.go b/input_file.go index ad2495c..043ca3e 100644 --- a/input_file.go +++ b/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 diff --git a/limiter.go b/limiter.go index 192cfa2..f53fd60 100644 --- a/limiter.go +++ b/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 } diff --git a/limiter_test.go b/limiter_test.go index ee56cc9..f1757d8 100644 --- a/limiter_test.go +++ b/limiter_test.go @@ -1,3 +1,5 @@ +// +build !race + package main import (