luciano/amazon-kinesis-client-go
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fixed edge case where the beginnings of messages would sometimes get chopped off as a result of creating a new buffered reader each time we'd read a line.

Browse source
This commit is contained in:
Xavi Ramirez 2017-08-20 03:26:32 +00:00
parent d63effaddd
commit 22915133fe
1 changed files with 11 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
kcl/kcl.go
View file

@ -23,11 +23,19 @@ type Checkpointer interface {
} }
type ioHandler struct { type ioHandler struct {
inputFile io.Reader input *bufio.Reader
outputFile io.Writer outputFile io.Writer
errorFile io.Writer errorFile io.Writer
} }
func newIOHandler(inputFile io.Reader, outputFile, errorFile io.Writer) ioHandler {
return ioHandler{
input: bufio.NewReader(inputFile),
outputFile: outputFile,
errorFile: errorFile,
}
}
func (i ioHandler) writeLine(line string) { func (i ioHandler) writeLine(line string) {
fmt.Fprintf(i.outputFile, "\n%s\n", line) fmt.Fprintf(i.outputFile, "\n%s\n", line)
} }
@ -37,8 +45,7 @@ func (i ioHandler) writeError(message string) {
} }
func (i ioHandler) readLine() (string, error) { func (i ioHandler) readLine() (string, error) {
bio := bufio.NewReader(i.inputFile) line, err := i.input.ReadString('\n')
line, err := bio.ReadString('\n')
if err != nil { if err != nil {
return "", err return "", err
} }
@ -130,13 +137,8 @@ func (i ioHandler) writeAction(action interface{}) error {
func New( func New(
inputFile io.Reader, outputFile, errorFile io.Writer, recordProcessor RecordProcessor, inputFile io.Reader, outputFile, errorFile io.Writer, recordProcessor RecordProcessor,
) *KCLProcess { ) *KCLProcess {
i := ioHandler{
inputFile: inputFile,
outputFile: outputFile,
errorFile: errorFile,
}
return &KCLProcess{ return &KCLProcess{
ioHandler: i, ioHandler: newIOHandler(inputFile, outputFile, errorFile),
recordProcessor: recordProcessor, recordProcessor: recordProcessor,
nextCheckpointPair: SequencePair{}, nextCheckpointPair: SequencePair{},