clean up

Author: Pryz

proxy.go

@@ -30,18 +30,18 @@ type ReverseProxy struct {
 
 // ServeRedis satisfies the Handler interface.
 func (proxy *ReverseProxy) ServeRedis(w ResponseWriter, r *Request) {
-for _, cmd := range r.Cmds {
-switch cmd.Cmd {
+if len(r.Cmds) == 1 {
+switch r.Cmds[0].Cmd {
 case "SUBSCRIBE", "PSUBSCRIBE":
 var channels []string
 var channel string
 
-for cmd.Args.Next(&channel) {
+for r.Cmds[0].Args.Next(&channel) {
 // TOOD: limit the number of channels read here
 channels = append(channels, channel)
 }
 
-if err := cmd.Args.Close(); err != nil {
+if err := r.Cmds[0].Args.Close(); err != nil {
 proxy.log(err)
 return
 }
@@ -52,13 +52,11 @@ func (proxy *ReverseProxy) ServeRedis(w ResponseWriter, r *Request) {
 return
 }
 
-proxy.servePubSub(conn, rw, cmd.Cmd, channels...)
-// TOD: figure out a way to pass the connection back in regular mode
-
-default:
-proxy.serveRequest(w, r)
+proxy.servePubSub(conn, rw, r.Cmds[0].Cmd, channels...)
+// TOD0: figure out a way to pass the connection back in regular mode
 }
 }
+proxy.serveRequest(w, r)
 }
 
 func (proxy *ReverseProxy) serveRequest(w ResponseWriter, req *Request) {

response.go

@@ -2,6 +2,7 @@ package redis
 
 import (
 "io"
+"strings"
 
 "github.com/segmentio/objconv/resp"
 )
@@ -48,7 +49,7 @@ func (res *Response) Close() error {
 }
 
 func newResponse(parser *resp.Parser, req *Request, done chan<- error) *Response {
-argsDone := make(chan error, 1)
+argsDone := make(chan error, len(req.Cmds))
 
 args := make([]Args, len(req.Cmds))
 for i := 0; i < len(req.Cmds); i++ {
@@ -60,12 +61,15 @@ func newResponse(parser *resp.Parser, req *Request, done chan<- error) *Response
 for range req.Cmds {
 if e := <-argsDone; err == nil && e != nil {
 err = e
+if redisErr, ok := e.(*resp.Error); ok {
+if strings.HasPrefix(redisErr.Error(), "EXECABORT") {
+break
+}
+}
 }
 
 }
-if done != nil {
-done <- err
-}
+done <- err
 }()
 
 return &Response{

server.go

@@ -305,21 +305,41 @@ func (s *Server) serveConnection(ctx context.Context, c *serverConn, config serv
 }
 
 func (s *Server) serveRequest(res *responseWriter, req *Request) (err error) {
-for _, cmd := range req.Cmds {
-switch cmd.Cmd {
+if len(req.Cmds) == 1 {
+switch req.Cmds[0].Cmd {
 case "PING":
 msg := "PONG"
 req.ParseArgs(&msg)
 res.Write(msg)
-
-default:
-err = s.serveRedis(res, req)
-cmd.Args.Close()
+return
 }
+}
+//for _, cmd := range req.Cmds {
+//	switch cmd.Cmd {
+//	case "PING":
+//	msg := "PONG"
+//	req.ParseArgs(&msg)
+//	res.Write(msg)
+
+//	default:
+//	err = s.serveRedis(res, req)
+//	cmd.Args.Close()
+//	}
+
+//	if err == nil {
+//	err = res.Flush()
+//	}
+//}
+err = s.serveRedis(res, req)
+if err == nil {
+err = res.Flush()
+}
+for _, cmd := range req.Cmds {
+cmd.Args.Close()
+}
 
-if err == nil {
-err = res.Flush()
-}
+if err == nil {
+err = res.Flush()
 }
 
 return
@@ -504,9 +524,11 @@ func readRequest(ctx context.Context, conn *serverConn, done chan<- error) (*Req
 
 cmds := []Command{cmd}
 if cmd.Cmd == "MULTI" {
+argsDone := make(chan error, 1)
+
 for cmd.Cmd != "EXEC" {
 cmd.Args = nil
-args = newByteArgsReader(&conn.p, nil)
+args = newByteArgsReader(&conn.p, argsDone)
 
 if !args.Next(&cmd.Cmd) {
 break
@@ -524,6 +546,16 @@ func readRequest(ctx context.Context, conn *serverConn, done chan<- error) (*Req
 cmd.Args = List(a...)
 cmds = append(cmds, cmd)
 }
+
+go func() {
+var err error
+for range cmds {
+if e := <-argsDone; err == nil && e != nil {
+err = e
+}
+}
+done <- err
+}()
 }
 
 req := &Request{