diff --git a/ledis/ledis.go b/ledis/ledis.go index e132ab4..040f362 100644 --- a/ledis/ledis.go +++ b/ledis/ledis.go @@ -11,10 +11,13 @@ import ( type Config struct { DataDB leveldb.Config `json:"data_db"` - BinLog replication.BinLogConfig `json:"binlog"` + BinLog replication.BinLogConfig `json:"binlog"` + RelayLog replication.RelayLogConfig `json:"relaylog"` } type DB struct { + l *Ledis + db *leveldb.DB index uint8 @@ -33,7 +36,10 @@ type Ledis struct { ldb *leveldb.DB dbs [MaxDBNumber]*DB - binlog *replication.Log + 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) { diff --git a/ledis/ledis_db.go b/ledis/ledis_db.go index a45f59b..cae79f9 100644 --- a/ledis/ledis_db.go +++ b/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 { - eliminator.active() - time.Sleep(1 * time.Second) + select { + case <-tick.C: + eliminator.active() + case <-db.l.quit: + break + } } + + tick.Stop() }() } diff --git a/ledis/replication.go b/ledis/replication.go index 2559dba..111f96b 100644 --- a/ledis/replication.go +++ b/ledis/replication.go @@ -11,7 +11,8 @@ import ( ) var ( - errInvalidBinLogEvent = errors.New("invalid binglog event") + 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 }