async replication refactoring

internal/app/app.go

@@ -2136,23 +2136,8 @@ func (app *App) waitForCatchUp(node *mysql.Node, gtidset gtids.GTIDSet, timeout
 		if app.dcs.Get(pathCurrentSwitch, switchover) == dcs.ErrNotFound {
 			return false, nil
 		}
-		if app.config.ASync && switchover.Cause == CauseAuto {
-			app.logger.Infof("async mode is active and this is auto switch so we checking new master delay")
-			ts, err := app.GetMdbReplMonTS()
-			if err != nil {
-				app.logger.Errorf("failed to get mdb repl mon ts: %v", err)
-				continue
-			}
-			delay, err := node.CalcMdbReplMonTSDelay(ts)
-			if err != nil {
-				app.logger.Errorf("failed to calc mdb repl mon ts: %v", err)
-				continue
-			}
-			if delay < app.config.AsyncAllowedLag {
-				app.logger.Infof("async allowed lag is %d and current lag on host %s is %d, so we don't wait for catch up any more",
-					app.config.AsyncAllowedLag, node.Host(), delay)
-				return true, nil
-			}
+		if app.CheckAsyncSwitchAllowed(node, switchover) {
+			return true, nil
 		}
 		time.Sleep(sleep)
 		if time.Now().After(deadline) {
@@ -2257,15 +2242,6 @@ func (app *App) getNodePositions(activeNodes []string) ([]nodePosition, error) {
 	return positions, util.CombineErrors(errs)
 }
 
-func (app *App) updateMdbReplMonTS(master string) error {
-	masterNode := app.cluster.Get(master)
-	ts, err := masterNode.GetMdbReplMonTS()
-	if err != nil {
-		return fmt.Errorf("failed to get master mdb_repl_mon timestamp: %v", err)
-	}
-	return app.SetMdbReplMonTS(ts)
-}
-
 /*
 Run enters the main application loop
 When Run exits mysync process is over

internal/app/async.go

@@ -0,0 +1,37 @@
+package app
+
+import (
+	"fmt"
+	"github.com/yandex/mysync/internal/mysql"
+)
+
+func (app *App) CheckAsyncSwitchAllowed(node *mysql.Node, switchover *Switchover) bool {
+	if app.config.ASync && switchover.Cause == CauseAuto {
+		app.logger.Infof("async mode is active and this is auto switch so we checking new master delay")
+		ts, err := app.GetMdbReplMonTS()
+		if err != nil {
+			app.logger.Errorf("failed to get mdb repl mon ts: %v", err)
+			return false
+		}
+		delay, err := node.CalcMdbReplMonTSDelay(ts)
+		if err != nil {
+			app.logger.Errorf("failed to calc mdb repl mon ts: %v", err)
+			return false
+		}
+		if delay < app.config.AsyncAllowedLag {
+			app.logger.Infof("async allowed lag is %d and current lag on host %s is %d, so we don't wait for catch up any more",
+				app.config.AsyncAllowedLag, node.Host(), delay)
+			return true
+		}
+	}
+	return false
+}
+
+func (app *App) updateMdbReplMonTS(master string) error {
+	masterNode := app.cluster.Get(master)
+	ts, err := masterNode.GetMdbReplMonTS()
+	if err != nil {
+		return fmt.Errorf("failed to get master mdb_repl_mon timestamp: %v", err)
+	}
+	return app.SetMdbReplMonTS(ts)
+}