Merge pull request #294 from github/close-streamer-connection

Close streamer connection

RELEASE_VERSION

@@ -0,0 +1 @@
+1.0.28

build.sh

@@ -2,7 +2,7 @@
 #
 #
 
-RELEASE_VERSION="1.0.26"
+RELEASE_VERSION=$(cat RELEASE_VERSION)
 
 function build {
     osname=$1

go/base/context.go

@@ -155,6 +155,7 @@ type MigrationContext struct {
 	AllEventsUpToLockProcessedInjectedFlag int64
 	CleanupImminentFlag                    int64
 	UserCommandedUnpostponeFlag            int64
+	CutOverCompleteFlag                    int64
 	PanicAbort                             chan error
 
 	OriginalTableColumnsOnApplier    *sql.ColumnList

go/binlog/gomysql_reader.go

@@ -118,6 +118,9 @@ func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEven
 
 // StreamEvents
 func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChannel chan<- *BinlogEntry) error {
+	if canStopStreaming() {
+		return nil
+	}
 	for {
 		if canStopStreaming() {
 			break
@@ -148,3 +151,8 @@ func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesCha
 
 	return nil
 }
+
+func (this *GoMySQLReader) Close() error {
+	this.binlogSyncer.Close()
+	return nil
+}

go/logic/migrator.go

@@ -171,7 +171,7 @@ func (this *Migrator) consumeRowCopyComplete() {
 }
 
 func (this *Migrator) canStopStreaming() bool {
-	return false
+	return atomic.LoadInt64(&this.migrationContext.CutOverCompleteFlag) != 0
 }
 
 // onChangelogStateEvent is called when a binlog event operation on the changelog table is intercepted.
@@ -345,6 +345,7 @@ func (this *Migrator) Migrate() (err error) {
 	if err := this.cutOver(); err != nil {
 		return err
 	}
+	atomic.StoreInt64(&this.migrationContext.CutOverCompleteFlag, 1)
 
 	if err := this.finalCleanup(); err != nil {
 		return nil
@@ -1058,6 +1059,9 @@ func (this *Migrator) finalCleanup() error {
 			log.Errore(err)
 		}
 	}
+	if err := this.eventsStreamer.Close(); err != nil {
+		log.Errore(err)
+	}
 
 	if err := this.retryOperation(this.applier.DropChangelogTable); err != nil {
 		return err

go/logic/streamer.go

@@ -217,3 +217,9 @@ func (this *EventsStreamer) StreamEvents(canStopStreaming func() bool) error {
 		}
 	}
 }
+
+func (this *EventsStreamer) Close() (err error) {
+	err = this.binlogReader.Close()
+	log.Infof("Closed streamer connection. err=%+v", err)
+	return err
+}