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

Fixed another race-condition. Ensure line is completely processed before allowing a checkpoint.

Browse source
This commit is contained in:
Xavi Ramirez 2017-08-06 01:52:11 +00:00
parent 6a30e0eb8a
commit c5f75d6554
1 changed files with 7 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
kcl/kcl.go
View file

@ -281,11 +281,13 @@ func (kclp *KCLProcess) startLineProcessor(
select { select {
case <-next: case <-next:
line, err := kclp.ioHandler.readLine() line, err := kclp.ioHandler.readLine()
if err != nil { if err == nil {
outErr <- err if line == "" {
} else { err = fmt.Errorf("Empty read line recieved")
out <- line }
err = kclp.handleLine(line)
} }
outErr <- err
case pair := <-checkpoint: case pair := <-checkpoint:
err := kclp.processCheckpoint(pair) err := kclp.processCheckpoint(pair)
checkpointErr <- err checkpointErr <- err
@ -297,20 +299,7 @@ func (kclp *KCLProcess) startLineProcessor(
func (kclp *KCLProcess) processNextLine() error { func (kclp *KCLProcess) processNextLine() error {
kclp.next <- struct{}{} // We're ready for a new line kclp.next <- struct{}{} // We're ready for a new line
var err error return <-kclp.outErr
var line string
select {
case err = <-kclp.outErr:
case line = <-kclp.out:
if line == "" {
err = fmt.Errorf("Empty read line recieved")
} else {
err = kclp.handleLine(line)
}
}
return err
} }
func (kclp *KCLProcess) Run() { func (kclp *KCLProcess) Run() {