mirror of https://github.com/tidwall/tile38.git
use redcon for loading aof
This commit is contained in:
parent
1c8027c580
commit
d4d51a8191
|
@ -1,7 +1,6 @@
|
||||||
package controller
|
package controller
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bufio"
|
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
|
@ -13,6 +12,7 @@ import (
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/tidwall/buntdb"
|
"github.com/tidwall/buntdb"
|
||||||
|
"github.com/tidwall/redcon"
|
||||||
"github.com/tidwall/resp"
|
"github.com/tidwall/resp"
|
||||||
"github.com/tidwall/tile38/controller/log"
|
"github.com/tidwall/tile38/controller/log"
|
||||||
"github.com/tidwall/tile38/controller/server"
|
"github.com/tidwall/tile38/controller/server"
|
||||||
|
@ -53,86 +53,57 @@ func (c *Controller) loadAOF() error {
|
||||||
log.Infof("AOF loaded %d commands: %.2fs, %.0f/s, %s",
|
log.Infof("AOF loaded %d commands: %.2fs, %.0f/s, %s",
|
||||||
count, float64(d)/float64(time.Second), ps, byteSpeed)
|
count, float64(d)/float64(time.Second), ps, byteSpeed)
|
||||||
}()
|
}()
|
||||||
|
var buf []byte
|
||||||
|
var args [][]byte
|
||||||
|
var packet [0xFFFF]byte
|
||||||
var msg server.Message
|
var msg server.Message
|
||||||
rd := bufio.NewReader(c.aof)
|
|
||||||
for {
|
for {
|
||||||
var nn int
|
n, err := c.aof.Read(packet[:])
|
||||||
ch, err := rd.ReadByte()
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if err == io.EOF {
|
if err == io.EOF {
|
||||||
|
if len(buf) > 0 {
|
||||||
|
return io.ErrUnexpectedEOF
|
||||||
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
nn += 1
|
c.aofsz += n
|
||||||
if ch != '*' {
|
data := packet[:n]
|
||||||
return errInvalidAOF
|
if len(buf) > 0 {
|
||||||
|
data = append(buf, data...)
|
||||||
}
|
}
|
||||||
ns, err := rd.ReadString('\n')
|
var complete bool
|
||||||
|
for {
|
||||||
|
complete, args, _, data, err = redcon.ReadNextCommand(data, args[:0])
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
nn += len(ns)
|
if !complete {
|
||||||
if len(ns) < 2 || ns[len(ns)-2] != '\r' {
|
break
|
||||||
return errInvalidAOF
|
|
||||||
}
|
|
||||||
n, err := strconv.ParseUint(ns[:len(ns)-2], 10, 64)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if int(n) == 0 {
|
|
||||||
continue
|
|
||||||
}
|
}
|
||||||
|
if len(args) > 0 {
|
||||||
msg.Values = msg.Values[:0]
|
msg.Values = msg.Values[:0]
|
||||||
for i := 0; i < int(n); i++ {
|
for _, arg := range args {
|
||||||
ch, err := rd.ReadByte()
|
msg.Values = append(msg.Values, resp.BytesValue(arg))
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if ch != '$' {
|
|
||||||
return errInvalidAOF
|
|
||||||
}
|
|
||||||
ns, err := rd.ReadString('\n')
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if len(ns) < 2 || ns[len(ns)-2] != '\r' {
|
|
||||||
return errInvalidAOF
|
|
||||||
}
|
|
||||||
n, err := strconv.ParseUint(ns[:len(ns)-2], 10, 64)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
b := make([]byte, int(n))
|
|
||||||
_, err = io.ReadFull(rd, b)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if ch, err := rd.ReadByte(); err != nil {
|
|
||||||
return err
|
|
||||||
} else if ch != '\r' {
|
|
||||||
return errInvalidAOF
|
|
||||||
}
|
|
||||||
if ch, err := rd.ReadByte(); err != nil {
|
|
||||||
return err
|
|
||||||
} else if ch != '\n' {
|
|
||||||
return errInvalidAOF
|
|
||||||
}
|
|
||||||
msg.Values = append(msg.Values, resp.BytesValue(b))
|
|
||||||
if i == 0 {
|
|
||||||
msg.Command = qlower(b)
|
|
||||||
}
|
|
||||||
nn += 1 + len(ns) + int(n) + 2
|
|
||||||
}
|
}
|
||||||
|
msg.Command = qlower(args[0])
|
||||||
if _, _, err := c.command(&msg, nil, nil); err != nil {
|
if _, _, err := c.command(&msg, nil, nil); err != nil {
|
||||||
if commandErrIsFatal(err) {
|
if commandErrIsFatal(err) {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
c.aofsz += nn
|
|
||||||
count++
|
count++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if len(data) > 0 {
|
||||||
|
buf = append(buf[:0], data...)
|
||||||
|
} else if len(buf) > 0 {
|
||||||
|
buf = buf[:0]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func qlower(s []byte) string {
|
func qlower(s []byte) string {
|
||||||
if len(s) == 3 {
|
if len(s) == 3 {
|
||||||
if s[0] == 'S' && s[1] == 'E' && s[2] == 'T' {
|
if s[0] == 'S' && s[1] == 'E' && s[2] == 'T' {
|
||||||
|
|
Loading…
Reference in New Issue