diff --git a/server/cmd_replication_test.go b/server/cmd_replication_test.go index 07db0c7..76bf2c2 100644 --- a/server/cmd_replication_test.go +++ b/server/cmd_replication_test.go @@ -120,4 +120,14 @@ func TestReplication(t *testing.T) { t.Fatal(err) } + slave.tryReSlaveof() + + time.Sleep(1 * time.Second) + + slave.ldb.WaitReplication() + + if err = checkDataEqual(master, slave); err != nil { + t.Fatal(err) + } + } diff --git a/server/cmd_server.go b/server/cmd_server.go index ab8051f..198003c 100644 --- a/server/cmd_server.go +++ b/server/cmd_server.go @@ -68,6 +68,9 @@ func flushallCommand(c *client) error { return err } + //we will restart the replication from master if possible + c.app.tryReSlaveof() + c.resp.writeStatus(OK) return nil } diff --git a/server/replication.go b/server/replication.go index 2c409bf..dcd1587 100644 --- a/server/replication.go +++ b/server/replication.go @@ -94,12 +94,12 @@ func (m *master) startReplication(masterAddr string, restart bool) error { m.app.ldb.SetReadOnly(true) + m.wg.Add(1) go m.runReplication(restart) return nil } func (m *master) runReplication(restart bool) { - m.wg.Add(1) defer m.wg.Done() for { @@ -245,6 +245,8 @@ func (app *App) slaveof(masterAddr string, restart bool) error { return fmt.Errorf("slaveof must enable replication") } + app.cfg.SlaveOf = masterAddr + if len(masterAddr) == 0 { if err := app.m.stopReplication(); err != nil { return err @@ -258,6 +260,21 @@ func (app *App) slaveof(masterAddr string, restart bool) error { return nil } +func (app *App) tryReSlaveof() error { + app.m.Lock() + defer app.m.Unlock() + + if !app.ldb.ReplicationUsed() { + return nil + } + + if len(app.cfg.SlaveOf) == 0 { + return nil + } else { + return app.m.startReplication(app.cfg.SlaveOf, true) + } +} + func (app *App) addSlave(c *client) { app.slock.Lock() defer app.slock.Unlock()