goreplay/input_file.go

package main

import (
	"log"
	"os"
	"bufio"
	"bytes"
	"strconv"
	"time"
)

// FileInput can read requests generated by FileOutput
type FileInput struct {
	data        chan []byte
	path        string
	file        *os.File
	speedFactor float64
}

// NewFileInput constructor for FileInput. Accepts file path as argument.
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()

	return
}

func (i *FileInput) init(path string) {
	file, err := os.Open(path)

	if err != nil {
		log.Fatal(i, "Cannot open file %q. Error: %s", path, err)
	}

	i.file = file
}

func (i *FileInput) Read(data []byte) (int, error) {
	buf := <-i.data
	copy(data, buf)

	return len(buf), nil
}

func (i *FileInput) String() string {
	return "File input: " + i.path
}

func scanSeparator(data []byte, atEOF bool) (advance int, token []byte, err error) {
	if atEOF && len(data) == 0 {
   		return 0, nil, nil
   	}

	if i := bytes.Index(data, []byte(fileSeparator)); i >= 0 {
		// We have a full newline-terminated line.
		return i + len(fileSeparator), data[0:i], nil
	}

	if atEOF {
   		return len(data), data, nil
	}
   return 0, nil, nil
}

func (i *FileInput) emit() {
	var lastTime int64

	// reader := bufio.NewReader(conn)
	scanner := bufio.NewScanner(i.file)
	scanner.Split(scanSeparator)

	for scanner.Scan() {
		buf := scanner.Bytes()
		meta := payloadMeta(buf)

		if meta[0][0] == '1' && lastTime != 0 {
			ts, _ := strconv.ParseInt(string(meta[2]), 10, 64)
			timeDiff := ts - lastTime

			if i.speedFactor != 1 {
				timeDiff = int64(float64(timeDiff) / i.speedFactor)
			}

			time.Sleep(time.Duration(timeDiff))

			lastTime = ts
		}

		i.data <- buf
	}
}