ledisdb/server/cmd_replication_test.go

package server

import (
	"fmt"
	"os"
	"reflect"
	"testing"
	"time"

	"github.com/siddontang/goredis"
	"github.com/siddontang/ledisdb/config"
)

func checkDataEqual(master *App, slave *App) error {
	mdb, _ := master.ldb.Select(0)
	sdb, _ := slave.ldb.Select(0)

	mkeys, _ := mdb.Scan(KV, nil, 100, true, "")
	skeys, _ := sdb.Scan(KV, nil, 100, true, "")

	if len(mkeys) != len(skeys) {
		return fmt.Errorf("keys number not equal %d != %d", len(mkeys), len(skeys))
	} else if !reflect.DeepEqual(mkeys, skeys) {
		return fmt.Errorf("keys not equal")
	} else {
		for _, k := range mkeys {
			v1, _ := mdb.Get(k)
			v2, _ := sdb.Get(k)
			if !reflect.DeepEqual(v1, v2) {
				return fmt.Errorf("value not equal")
			}
		}
	}

	return nil
}

func TestReplication(t *testing.T) {
	data_dir := "/tmp/test_replication"
	os.RemoveAll(data_dir)

	masterCfg := config.NewConfigDefault()
	masterCfg.DataDir = fmt.Sprintf("%s/master", data_dir)
	masterCfg.Addr = "127.0.0.1:11182"
	masterCfg.UseReplication = true
	masterCfg.Replication.Sync = true
	masterCfg.Replication.WaitSyncTime = 5000

	var master *App
	var slave *App
	var err error
	master, err = NewApp(masterCfg)
	if err != nil {
		t.Fatal(err)
	}
	defer master.Close()

	slaveCfg := config.NewConfigDefault()
	slaveCfg.DataDir = fmt.Sprintf("%s/slave", data_dir)
	slaveCfg.Addr = "127.0.0.1:11183"
	slaveCfg.SlaveOf = masterCfg.Addr
	slaveCfg.UseReplication = true

	slave, err = NewApp(slaveCfg)
	if err != nil {
		t.Fatal(err)
	}
	defer slave.Close()

	go master.Run()

	time.Sleep(1 * time.Second)
	go slave.Run()

	db, _ := master.ldb.Select(0)

	value := make([]byte, 10)

	db.Set([]byte("a"), value)
	db.Set([]byte("b"), value)
	db.Set([]byte("c"), value)
	db.Set([]byte("d"), value)

	time.Sleep(1 * time.Second)
	if err = checkDataEqual(master, slave); err != nil {
		t.Fatal(err)
	}

	db.Set([]byte("a1"), value)
	db.Set([]byte("b1"), value)
	db.Set([]byte("c1"), value)
	db.Set([]byte("d1"), value)

	//time.Sleep(1 * time.Second)
	slave.ldb.WaitReplication()

	if err = checkDataEqual(master, slave); err != nil {
		t.Fatal(err)
	}

	slave.slaveof("", false, false)

	db.Set([]byte("a2"), value)
	db.Set([]byte("b2"), value)
	db.Set([]byte("c2"), value)
	db.Set([]byte("d2"), value)

	db.Set([]byte("a3"), value)
	db.Set([]byte("b3"), value)
	db.Set([]byte("c3"), value)
	db.Set([]byte("d3"), value)

	if err = checkDataEqual(master, slave); err == nil {
		t.Fatal("must error")
	}

	slave.slaveof(masterCfg.Addr, false, false)

	time.Sleep(1 * time.Second)

	if err = checkDataEqual(master, slave); err != nil {
		t.Fatal(err)
	}

	mStat, _ := master.ldb.ReplicationStat()
	sStat, _ := slave.ldb.ReplicationStat()

	if err = checkTestRole(masterCfg.Addr, []interface{}{
		[]byte("master"),
		int64(mStat.LastID),
		[]interface{}{
			[]interface{}{
				[]byte("127.0.0.1"),
				[]byte("11183"),
				[]byte(fmt.Sprintf("%d", sStat.LastID)),
			}},
	}); err != nil {
		t.Fatal(err)
	}

	if err = checkTestRole(slaveCfg.Addr, []interface{}{
		[]byte("slave"),
		[]byte("127.0.0.1"),
		int64(11182),
		[]byte("connected"),
		int64(sStat.LastID),
	}); err != nil {
		t.Fatal(err)
	}

	slave.tryReSlaveof()

	time.Sleep(1 * time.Second)

	slave.ldb.WaitReplication()

	if err = checkDataEqual(master, slave); err != nil {
		t.Fatal(err)
	}
}

func checkTestRole(addr string, checkRoles []interface{}) error {
	conn, _ := goredis.Connect(addr)
	defer conn.Close()
	roles, err := goredis.MultiBulk(conn.Do("ROLE"))
	if err != nil {
		return err
	} else if !reflect.DeepEqual(roles, checkRoles) {
		return fmt.Errorf("%v != %v", roles, checkRoles)
	}
	return nil
}