add relay log, add quit channel

ledis/ledis.go

@@ -12,9 +12,12 @@ type Config struct {
 	DataDB leveldb.Config `json:"data_db"`
 
 	BinLog   replication.BinLogConfig   `json:"binlog"`
+	RelayLog replication.RelayLogConfig `json:"relaylog"`
 }
 
 type DB struct {
+	l *Ledis
+
 	db *leveldb.DB
 
 	index uint8
@@ -34,6 +37,9 @@ type Ledis struct {
 	dbs [MaxDBNumber]*DB
 
 	binlog   *replication.Log
+	relaylog *replication.Log
+
+	quit chan struct{}
 }
 
 func Open(configJson json.RawMessage) (*Ledis, error) {
@@ -53,6 +59,9 @@ func OpenWithConfig(cfg *Config) (*Ledis, error) {
 	}
 
 	l := new(Ledis)
+
+	l.quit = make(chan struct{})
+
 	l.ldb = ldb
 
 	if len(cfg.BinLog.Path) > 0 {
@@ -64,6 +73,15 @@ func OpenWithConfig(cfg *Config) (*Ledis, error) {
 		l.binlog = nil
 	}
 
+	if len(cfg.RelayLog.Path) > 0 {
+		l.relaylog, err = replication.NewRelayLogWithConfig(&cfg.RelayLog)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		l.relaylog = nil
+	}
+
 	for i := uint8(0); i < MaxDBNumber; i++ {
 		l.dbs[i] = newDB(l, i)
 	}
@@ -74,6 +92,8 @@ func OpenWithConfig(cfg *Config) (*Ledis, error) {
 func newDB(l *Ledis, index uint8) *DB {
 	d := new(DB)
 
+	d.l = l
+
 	d.db = l.ldb
 
 	d.index = index
@@ -89,7 +109,19 @@ func newDB(l *Ledis, index uint8) *DB {
 }
 
 func (l *Ledis) Close() {
+	close(l.quit)
+
 	l.ldb.Close()
+
+	if l.binlog != nil {
+		l.binlog.Close()
+		l.binlog = nil
+	}
+
+	if l.relaylog != nil {
+		l.relaylog.Close()
+		l.relaylog = nil
+	}
 }
 
 func (l *Ledis) Select(index int) (*DB, error) {

ledis/ledis_db.go

@@ -28,9 +28,16 @@ func (db *DB) activeExpireCycle() {
 	eliminator.regRetireContext(kvExpType, db.kvTx, db.delete)
 
 	go func() {
+		tick := time.NewTicker(1 * time.Second)
 		for {
+			select {
+			case <-tick.C:
 				eliminator.active()
-			time.Sleep(1 * time.Second)
+			case <-db.l.quit:
+				break
 			}
+		}
+
+		tick.Stop()
 	}()
 }

ledis/replication.go

@@ -12,6 +12,7 @@ import (
 
 var (
 	errInvalidBinLogEvent   = errors.New("invalid binglog event")
+	errRelayLogNotSupported = errors.New("write relay log not supported")
 )
 
 func (l *Ledis) replicateEvent(event []byte) error {
@@ -132,5 +133,11 @@ func (l *Ledis) RepliateRelayLog(relayLog string, offset int64) (int64, error) {
 }
 
 func (l *Ledis) WriteRelayLog(data []byte) error {
-	return nil
+	if l.relaylog == nil {
+		return errRelayLogNotSupported
+	}
+
+	err := l.relaylog.Log(data)
+
+	return err
 }