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

Cleaned up shutdown code

Browse source
This commit is contained in:
Xavi Ramirez 2017-09-20 21:28:43 +00:00
parent 7f2afe9d8f
commit b07ad4c1c2
1 changed files with 30 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
kcl/kcl.go
View file

@ -141,7 +141,7 @@ func New(
recordProcessor: recordProcessor,
nextCheckpointPair: SequencePair{},
wasAskedToShutdown: false,
doGracefulShutdown: false,
}
}
@ -152,7 +152,7 @@ type KCLProcess struct {
recordProcessor RecordProcessor
nextCheckpointPair SequencePair
wasAskedToShutdown bool
doGracefulShutdown bool
}
func (kclp *KCLProcess) Checkpoint(pair SequencePair) {
@ -166,7 +166,7 @@ func (kclp *KCLProcess) Checkpoint(pair SequencePair) {
func (kclp *KCLProcess) Shutdown() {
kclp.ioHandler.writeError("Checkpoint shutdown")
kclp.wasAskedToShutdown = true
kclp.doGracefulShutdown = true
}
func (kclp *KCLProcess) handleCheckpointAction(action ActionCheckpoint) error {
@ -224,15 +224,15 @@ func (kclp *KCLProcess) reportDone(responseFor string) error {
})
}
func (kclp *KCLProcess) handleLine(line string) error {
func (kclp *KCLProcess) handleLine(line string) (string, error) {
action, err := kclp.ioHandler.loadAction(line)
if err != nil {
return err
return "", err
}
switch action := action.(type) {
case ActionCheckpoint:
err = kclp.handleCheckpointAction(action)
return "checkpoint", kclp.handleCheckpointAction(action)
case ActionShutdown:
kclp.ioHandler.writeError("Received shutdown action...")
@ -244,27 +244,31 @@ func (kclp *KCLProcess) handleLine(line string) error {
kclp.ioHandler.writeError("Reporting shutdown done")
err := kclp.sendCheckpoint(nil, nil) // nil sequence num is signal to shutdown
if err != nil {
kclp.ioHandler.writeError(fmt.Sprintf("ERR shutdown ask: %+#v", err))
if kclp.doGracefulShutdown {
err := kclp.sendCheckpoint(nil, nil) // nil sequence num is signal to shutdown
if err != nil {
kclp.ioHandler.writeError(fmt.Sprintf("ERR shutdown ask: %+#v", err))
}
}
return kclp.reportDone("shutdown")
return "shutdown", kclp.reportDone("shutdown")
case ActionInitialize:
err = kclp.recordProcessor.Initialize(action.ShardID, kclp)
if err == nil {
err = kclp.reportDone(action.Action)
err := kclp.recordProcessor.Initialize(action.ShardID, kclp)
if err != nil {
return "", err
}
case ActionProcessRecords:
err = kclp.recordProcessor.ProcessRecords(action.Records)
if err == nil {
err = kclp.reportDone(action.Action)
}
default:
err = fmt.Errorf("unknown action to dispatch: %+#v", action)
}
return err
return "initialize", kclp.reportDone(action.Action)
case ActionProcessRecords:
err := kclp.recordProcessor.ProcessRecords(action.Records)
if err != nil {
return "", err
}
return "process-record", kclp.reportDone(action.Action)
default:
return "", fmt.Errorf("unknown action to dispatch: %+#v", action)
}
}
func (kclp *KCLProcess) Run() {
@ -281,9 +285,11 @@ func (kclp *KCLProcess) Run() {
continue
}
err = kclp.handleLine(line)
action, err := kclp.handleLine(line)
if err != nil {
kclp.ioHandler.writeError(fmt.Sprintf("ERR Handle line: %+#v", err))
} else if action == "shutdown" {
return
}
kclp.ckpmux.Lock()
@ -297,7 +303,7 @@ func (kclp *KCLProcess) Run() {
} else {
kclp.nextCheckpointPair = SequencePair{}
}
} else if kclp.wasAskedToShutdown {
} else if kclp.doGracefulShutdown {
err := kclp.sendCheckpoint(nil, nil) // nil sequence num is signal to shutdown
if err != nil {
kclp.ioHandler.writeError(fmt.Sprintf("ERR shutdown ask: %+#v", err))