2014-07-31 10:38:20 +04:00
|
|
|
package server
|
|
|
|
|
|
|
|
import (
|
|
|
|
"bufio"
|
|
|
|
"errors"
|
2015-02-03 09:15:11 +03:00
|
|
|
"fmt"
|
2014-09-24 08:34:21 +04:00
|
|
|
"github.com/siddontang/go/hack"
|
2014-09-24 05:46:36 +04:00
|
|
|
"github.com/siddontang/go/log"
|
2014-09-24 09:29:27 +04:00
|
|
|
"github.com/siddontang/go/num"
|
2015-03-23 14:50:37 +03:00
|
|
|
"github.com/siddontang/goredis"
|
2014-07-31 10:38:20 +04:00
|
|
|
"github.com/siddontang/ledisdb/ledis"
|
|
|
|
"io"
|
|
|
|
"net"
|
2014-07-31 14:28:19 +04:00
|
|
|
"runtime"
|
2014-07-31 10:38:20 +04:00
|
|
|
"strconv"
|
2014-10-30 06:11:45 +03:00
|
|
|
"time"
|
2014-07-31 10:38:20 +04:00
|
|
|
)
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
var errReadRequest = errors.New("invalid request protocol")
|
2014-11-11 10:21:41 +03:00
|
|
|
var errClientQuit = errors.New("remote client quit")
|
2014-07-31 14:28:19 +04:00
|
|
|
|
|
|
|
type respClient struct {
|
2014-08-25 10:18:23 +04:00
|
|
|
*client
|
2014-07-31 14:28:19 +04:00
|
|
|
|
2014-07-31 10:38:20 +04:00
|
|
|
conn net.Conn
|
2014-10-29 19:01:19 +03:00
|
|
|
|
2015-03-23 14:50:37 +03:00
|
|
|
respReader *goredis.RespReader
|
2014-11-11 10:21:41 +03:00
|
|
|
|
|
|
|
activeQuit bool
|
2014-07-31 10:38:20 +04:00
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
type respWriter struct {
|
2014-07-31 10:38:20 +04:00
|
|
|
buff *bufio.Writer
|
|
|
|
}
|
|
|
|
|
2014-11-11 10:21:41 +03:00
|
|
|
func (app *App) addRespClient(c *respClient) {
|
|
|
|
app.rcm.Lock()
|
|
|
|
app.rcs[c] = struct{}{}
|
|
|
|
app.rcm.Unlock()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (app *App) delRespClient(c *respClient) {
|
|
|
|
app.rcm.Lock()
|
|
|
|
delete(app.rcs, c)
|
|
|
|
app.rcm.Unlock()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (app *App) closeAllRespClients() {
|
|
|
|
app.rcm.Lock()
|
|
|
|
|
|
|
|
for c := range app.rcs {
|
|
|
|
c.conn.Close()
|
|
|
|
}
|
|
|
|
|
|
|
|
app.rcm.Unlock()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (app *App) respClientNum() int {
|
|
|
|
app.rcm.Lock()
|
|
|
|
n := len(app.rcs)
|
|
|
|
app.rcm.Unlock()
|
|
|
|
return n
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func newClientRESP(conn net.Conn, app *App) {
|
|
|
|
c := new(respClient)
|
2014-07-31 10:38:20 +04:00
|
|
|
|
2014-08-25 10:18:23 +04:00
|
|
|
c.client = newClient(app)
|
2014-07-31 14:28:19 +04:00
|
|
|
c.conn = conn
|
2014-07-31 10:38:20 +04:00
|
|
|
|
2014-11-11 10:21:41 +03:00
|
|
|
c.activeQuit = false
|
|
|
|
|
2014-10-25 10:48:52 +04:00
|
|
|
if tcpConn, ok := conn.(*net.TCPConn); ok {
|
|
|
|
tcpConn.SetReadBuffer(app.cfg.ConnReadBufferSize)
|
|
|
|
tcpConn.SetWriteBuffer(app.cfg.ConnWriteBufferSize)
|
|
|
|
}
|
2014-10-28 12:58:37 +03:00
|
|
|
|
2015-03-23 14:50:37 +03:00
|
|
|
br := bufio.NewReaderSize(conn, app.cfg.ConnReadBufferSize)
|
|
|
|
c.respReader = goredis.NewRespReader(br)
|
2014-07-31 10:38:20 +04:00
|
|
|
|
2014-10-25 10:48:52 +04:00
|
|
|
c.resp = newWriterRESP(conn, app.cfg.ConnWriteBufferSize)
|
2014-08-25 10:18:23 +04:00
|
|
|
c.remoteAddr = conn.RemoteAddr().String()
|
2014-07-31 10:38:20 +04:00
|
|
|
|
2014-11-11 10:21:41 +03:00
|
|
|
app.connWait.Add(1)
|
|
|
|
|
|
|
|
app.addRespClient(c)
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
go c.run()
|
2014-07-31 10:38:20 +04:00
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (c *respClient) run() {
|
|
|
|
defer func() {
|
|
|
|
if e := recover(); e != nil {
|
|
|
|
buf := make([]byte, 4096)
|
|
|
|
n := runtime.Stack(buf, false)
|
|
|
|
buf = buf[0:n]
|
|
|
|
|
2015-01-12 05:32:03 +03:00
|
|
|
log.Fatalf("client run panic %s:%v", buf, e)
|
2014-07-31 14:28:19 +04:00
|
|
|
}
|
|
|
|
|
2014-11-11 10:21:41 +03:00
|
|
|
c.client.close()
|
|
|
|
|
|
|
|
c.conn.Close()
|
2014-08-25 10:18:23 +04:00
|
|
|
|
2015-03-14 04:10:00 +03:00
|
|
|
// if c.tx != nil {
|
|
|
|
// c.tx.Rollback()
|
|
|
|
// c.tx = nil
|
|
|
|
// }
|
2014-09-27 16:11:36 +04:00
|
|
|
|
2014-11-11 10:21:41 +03:00
|
|
|
c.app.removeSlave(c.client, c.activeQuit)
|
|
|
|
|
|
|
|
c.app.delRespClient(c)
|
|
|
|
|
|
|
|
c.app.connWait.Done()
|
2014-07-31 14:28:19 +04:00
|
|
|
}()
|
2014-07-31 10:38:20 +04:00
|
|
|
|
2014-11-11 10:21:41 +03:00
|
|
|
select {
|
|
|
|
case <-c.app.quit:
|
|
|
|
//check app closed
|
|
|
|
return
|
|
|
|
default:
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
2014-10-30 06:11:45 +03:00
|
|
|
kc := time.Duration(c.app.cfg.ConnKeepaliveInterval) * time.Second
|
2014-07-31 14:28:19 +04:00
|
|
|
for {
|
2014-10-30 06:11:45 +03:00
|
|
|
if kc > 0 {
|
|
|
|
c.conn.SetReadDeadline(time.Now().Add(kc))
|
|
|
|
}
|
|
|
|
|
2015-03-23 14:50:37 +03:00
|
|
|
c.cmd = ""
|
|
|
|
c.args = nil
|
|
|
|
|
|
|
|
reqData, err := c.respReader.ParseRequest()
|
2014-11-10 08:09:37 +03:00
|
|
|
if err == nil {
|
2014-11-11 10:21:41 +03:00
|
|
|
err = c.handleRequest(reqData)
|
2014-11-10 08:09:37 +03:00
|
|
|
}
|
2014-10-30 04:03:58 +03:00
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
if err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
2014-07-31 10:38:20 +04:00
|
|
|
}
|
|
|
|
|
2014-11-11 10:21:41 +03:00
|
|
|
func (c *respClient) handleRequest(reqData [][]byte) error {
|
2014-07-31 14:28:19 +04:00
|
|
|
if len(reqData) == 0 {
|
2014-08-25 10:18:23 +04:00
|
|
|
c.cmd = ""
|
|
|
|
c.args = reqData[0:0]
|
2014-07-31 14:28:19 +04:00
|
|
|
} else {
|
2014-10-30 07:48:52 +03:00
|
|
|
c.cmd = hack.String(lowerSlice(reqData[0]))
|
2014-08-25 10:18:23 +04:00
|
|
|
c.args = reqData[1:]
|
2014-07-31 14:28:19 +04:00
|
|
|
}
|
2015-03-21 15:33:57 +03:00
|
|
|
|
2015-03-22 03:39:20 +03:00
|
|
|
if c.cmd == "xselect" {
|
|
|
|
err := c.handleXSelectCmd()
|
2015-03-21 15:33:57 +03:00
|
|
|
if err != nil {
|
|
|
|
c.resp.writeError(err)
|
|
|
|
c.resp.flush()
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-08-25 10:18:23 +04:00
|
|
|
if c.cmd == "quit" {
|
2014-11-11 10:21:41 +03:00
|
|
|
c.activeQuit = true
|
2014-08-25 10:18:23 +04:00
|
|
|
c.resp.writeStatus(OK)
|
|
|
|
c.resp.flush()
|
2014-08-04 07:06:28 +04:00
|
|
|
c.conn.Close()
|
2014-11-11 10:21:41 +03:00
|
|
|
return errClientQuit
|
2014-08-04 07:06:28 +04:00
|
|
|
}
|
2014-07-31 14:28:19 +04:00
|
|
|
|
2014-08-25 10:18:23 +04:00
|
|
|
c.perform()
|
2014-08-01 05:38:08 +04:00
|
|
|
|
2015-03-21 15:33:57 +03:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2015-03-22 03:39:20 +03:00
|
|
|
// XSELECT db THEN command
|
|
|
|
func (c *respClient) handleXSelectCmd() error {
|
2015-03-21 15:33:57 +03:00
|
|
|
if len(c.args) <= 2 {
|
|
|
|
// invalid command format
|
2015-03-22 03:39:20 +03:00
|
|
|
return fmt.Errorf("invalid format for XSELECT, must XSELECT db THEN your command")
|
2015-03-21 15:33:57 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
if hack.String(upperSlice(c.args[1])) != "THEN" {
|
|
|
|
// invalid command format, just resturn here
|
2015-03-22 03:39:20 +03:00
|
|
|
return fmt.Errorf("invalid format for XSELECT, must XSELECT db THEN your command")
|
2015-03-21 15:33:57 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
index, err := strconv.Atoi(hack.String(c.args[0]))
|
|
|
|
if err != nil {
|
2015-03-22 03:39:20 +03:00
|
|
|
return fmt.Errorf("invalid db for XSELECT, err %v", err)
|
2015-03-21 15:33:57 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
db, err := c.app.ldb.Select(index)
|
|
|
|
if err != nil {
|
2015-03-22 03:39:20 +03:00
|
|
|
return fmt.Errorf("invalid db for XSELECT, err %v", err)
|
2015-03-21 15:33:57 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
c.db = db
|
2014-10-29 19:01:19 +03:00
|
|
|
|
2015-03-21 15:33:57 +03:00
|
|
|
c.cmd = hack.String(lowerSlice(c.args[2]))
|
|
|
|
c.args = c.args[3:]
|
2014-10-29 19:01:19 +03:00
|
|
|
|
2014-11-11 10:21:41 +03:00
|
|
|
return nil
|
2014-07-31 10:38:20 +04:00
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
// response writer
|
2014-07-31 10:38:20 +04:00
|
|
|
|
2014-10-25 10:48:52 +04:00
|
|
|
func newWriterRESP(conn net.Conn, size int) *respWriter {
|
2014-07-31 14:28:19 +04:00
|
|
|
w := new(respWriter)
|
2014-10-25 10:48:52 +04:00
|
|
|
w.buff = bufio.NewWriterSize(conn, size)
|
2014-07-31 10:38:20 +04:00
|
|
|
return w
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) writeError(err error) {
|
2015-03-23 14:50:37 +03:00
|
|
|
w.buff.Write(hack.Slice("-"))
|
2014-07-31 10:38:20 +04:00
|
|
|
if err != nil {
|
|
|
|
w.buff.WriteByte(' ')
|
2014-09-24 08:34:21 +04:00
|
|
|
w.buff.Write(hack.Slice(err.Error()))
|
2014-07-31 10:38:20 +04:00
|
|
|
}
|
|
|
|
w.buff.Write(Delims)
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) writeStatus(status string) {
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.WriteByte('+')
|
2014-09-24 08:34:21 +04:00
|
|
|
w.buff.Write(hack.Slice(status))
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Write(Delims)
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) writeInteger(n int64) {
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.WriteByte(':')
|
2014-09-24 09:29:27 +04:00
|
|
|
w.buff.Write(num.FormatInt64ToSlice(n))
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Write(Delims)
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) writeBulk(b []byte) {
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.WriteByte('$')
|
|
|
|
if b == nil {
|
|
|
|
w.buff.Write(NullBulk)
|
|
|
|
} else {
|
2014-09-24 08:34:21 +04:00
|
|
|
w.buff.Write(hack.Slice(strconv.Itoa(len(b))))
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Write(Delims)
|
|
|
|
w.buff.Write(b)
|
|
|
|
}
|
|
|
|
|
|
|
|
w.buff.Write(Delims)
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) writeArray(lst []interface{}) {
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.WriteByte('*')
|
|
|
|
if lst == nil {
|
|
|
|
w.buff.Write(NullArray)
|
|
|
|
w.buff.Write(Delims)
|
|
|
|
} else {
|
2014-09-24 08:34:21 +04:00
|
|
|
w.buff.Write(hack.Slice(strconv.Itoa(len(lst))))
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Write(Delims)
|
|
|
|
|
|
|
|
for i := 0; i < len(lst); i++ {
|
|
|
|
switch v := lst[i].(type) {
|
|
|
|
case []interface{}:
|
|
|
|
w.writeArray(v)
|
2014-08-26 19:21:45 +04:00
|
|
|
case [][]byte:
|
|
|
|
w.writeSliceArray(v)
|
2014-07-31 10:38:20 +04:00
|
|
|
case []byte:
|
|
|
|
w.writeBulk(v)
|
|
|
|
case nil:
|
|
|
|
w.writeBulk(nil)
|
|
|
|
case int64:
|
|
|
|
w.writeInteger(v)
|
|
|
|
default:
|
2015-02-03 09:15:11 +03:00
|
|
|
panic(fmt.Sprintf("invalid array type %T %v", lst[i], v))
|
2014-07-31 10:38:20 +04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) writeSliceArray(lst [][]byte) {
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.WriteByte('*')
|
|
|
|
if lst == nil {
|
|
|
|
w.buff.Write(NullArray)
|
|
|
|
w.buff.Write(Delims)
|
|
|
|
} else {
|
2014-09-24 08:34:21 +04:00
|
|
|
w.buff.Write(hack.Slice(strconv.Itoa(len(lst))))
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Write(Delims)
|
|
|
|
|
|
|
|
for i := 0; i < len(lst); i++ {
|
|
|
|
w.writeBulk(lst[i])
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) writeFVPairArray(lst []ledis.FVPair) {
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.WriteByte('*')
|
|
|
|
if lst == nil {
|
|
|
|
w.buff.Write(NullArray)
|
|
|
|
w.buff.Write(Delims)
|
|
|
|
} else {
|
2014-09-24 08:34:21 +04:00
|
|
|
w.buff.Write(hack.Slice(strconv.Itoa(len(lst) * 2)))
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Write(Delims)
|
|
|
|
|
|
|
|
for i := 0; i < len(lst); i++ {
|
|
|
|
w.writeBulk(lst[i].Field)
|
|
|
|
w.writeBulk(lst[i].Value)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) writeScorePairArray(lst []ledis.ScorePair, withScores bool) {
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.WriteByte('*')
|
|
|
|
if lst == nil {
|
|
|
|
w.buff.Write(NullArray)
|
|
|
|
w.buff.Write(Delims)
|
|
|
|
} else {
|
|
|
|
if withScores {
|
2014-09-24 08:34:21 +04:00
|
|
|
w.buff.Write(hack.Slice(strconv.Itoa(len(lst) * 2)))
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Write(Delims)
|
|
|
|
} else {
|
2014-09-24 08:34:21 +04:00
|
|
|
w.buff.Write(hack.Slice(strconv.Itoa(len(lst))))
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Write(Delims)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
for i := 0; i < len(lst); i++ {
|
|
|
|
w.writeBulk(lst[i].Member)
|
|
|
|
|
|
|
|
if withScores {
|
2014-09-24 09:29:27 +04:00
|
|
|
w.writeBulk(num.FormatInt64ToSlice(lst[i].Score))
|
2014-07-31 10:38:20 +04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) writeBulkFrom(n int64, rb io.Reader) {
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.WriteByte('$')
|
2014-09-24 08:34:21 +04:00
|
|
|
w.buff.Write(hack.Slice(strconv.FormatInt(n, 10)))
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Write(Delims)
|
|
|
|
|
|
|
|
io.Copy(w.buff, rb)
|
|
|
|
w.buff.Write(Delims)
|
|
|
|
}
|
|
|
|
|
2014-07-31 14:28:19 +04:00
|
|
|
func (w *respWriter) flush() {
|
2014-07-31 10:38:20 +04:00
|
|
|
w.buff.Flush()
|
|
|
|
}
|