Feature/multi (#101)

* log client count only in debug mode

* removed unwanted printf

* client count print only on debug

* set default level to info

* well why not some asciiart

* typo

* goimport

* added new errors

* added new errors

* added multi and exec

* command table updated with multi and exec

* updated exec command to handle a multi transaction

* updated exec command to handle a multi transaction

* run test coverage

* run test coverage

.gitignore

@@ -5,3 +5,4 @@ bin
 build
 vendor
 dist
+coverage.txt

.travis.yml

@@ -8,6 +8,7 @@ before_install:
 
 script:
   - mage check
+  - mage testcover
 
 after_success:
   - bash <(curl -s https://codecov.io/bash)

internal/client/client.go

@@ -68,9 +68,11 @@ type Client struct {
 	// Multi will indicate that client is in multi mode or not
 	Multi bool
 
-	// Pending will indicate number of pending commands needs to send
-	// If zero or 1 it will write reply immediately to the connection
-	Pending int
+	// MultiError indicates whether a command failed during a transaction
+	MultiError bool
+
+	// Commands store a list of queued commands in a multi transaction
+	Commands []*Command
 
 	// Writer is used to write out data to client connection
 	*bufio.Writer
@@ -180,14 +182,6 @@ func (client *Client) WriteInteger(n int) {
 
 // WriteProtocolReply will write a protocol reply
 func (client *Client) WriteProtocolReply(reply protocol.Reply) {
-	// if we are in multi mode and we still have commands to be processed wait
-	// cache the reply too
-	if client.Pending > 0 {
-		// TODO cache reply
-		client.Pending--
-		return
-	}
-
 	// ok we are clear to send
 	_, err := client.Write(reply.ToBytes())
 	if err != nil {

internal/client/command-table.go

@@ -28,7 +28,9 @@ package client
 // CommandTable holds all commands that are supported by kache
 var CommandTable = map[string]Command{
 	// server
-	"ping": {ModifyKeySpace: false, Fn: Ping, MinArgs: 0, MaxArgs: 1},
+	"ping":  {ModifyKeySpace: false, Fn: Ping, MinArgs: 0, MaxArgs: 1},
+	"multi": {ModifyKeySpace: true, Fn: Multi, MinArgs: 0, MaxArgs: 0},
+	"exec":  {ModifyKeySpace: true, Fn: Exec, MinArgs: 0, MaxArgs: 0},
 
 	// key space
 	"exists": {ModifyKeySpace: false, Fn: Exists, MinArgs: 1, MaxArgs: 1},

internal/client/command.go

@@ -26,22 +26,22 @@
 package client
 
 import (
+	"strings"
+
 	"github.com/kasvith/kache/internal/protocol"
+	"github.com/kasvith/kache/internal/resp/resp2"
 )
 
+// CommandFunc holds a function signature which can be used as a command.
+type CommandFunc func(*Client, []string)
+
 // Command holds a command structure which is used to execute a kache command
 type Command struct {
 	ModifyKeySpace bool
 	Fn             CommandFunc
 	MinArgs        int // 0
 	MaxArgs        int // -1 ~ +inf, -1 mean infinite
-}
-
-// CommandFunc holds a function signature which can be used as a command.
-type CommandFunc func(*Client, []string)
-
-// DBCommand is a command that executes on a given db
-type DBCommand struct {
+	Args           []string
 }
 
 // GetCommand will fetch the command from command table
@@ -57,14 +57,31 @@ func GetCommand(cmd string) (*Command, error) {
 func Execute(client *Client, cmd string, args []string) {
 	command, err := GetCommand(cmd)
 	if err != nil {
+		if client.Multi {
+			client.MultiError = true
+			client.Commands = []*Command{}
+		}
 		client.WriteError(err)
 		return
 	}
 
 	if argsLen := len(args); (command.MinArgs > 0 && argsLen < command.MinArgs) || (command.MaxArgs != -1 && argsLen > command.MaxArgs) {
+		if client.Multi {
+			client.MultiError = true
+			client.Commands = []*Command{}
+		}
 		client.WriteError(&protocol.ErrWrongNumberOfArgs{Cmd: cmd})
 		return
 	}
 
+	if client.Multi && strings.ToLower(cmd) != "exec" {
+		// store args for later use
+		command.Args = args
+		client.Commands = append(client.Commands, command)
+		client.WriteProtocolReply(resp2.NewSimpleStringReply("QUEUED"))
+		return
+	}
+
+	// execute command directly
 	command.Fn(client, args)
 }

internal/client/server.go

@@ -25,7 +25,10 @@
 
 package client
 
-import "github.com/kasvith/kache/internal/resp/resp2"
+import (
+	"github.com/kasvith/kache/internal/protocol"
+	"github.com/kasvith/kache/internal/resp/resp2"
+)
 
 // Ping will return PONG when no argument found or will echo the given argument
 func Ping(client *Client, args []string) {
@@ -36,3 +39,32 @@ func Ping(client *Client, args []string) {
 
 	client.WriteProtocolReply(resp2.NewSimpleStringReply(args[0]))
 }
+
+// Multi command will put client in multi mode where can execute multiple commands at once
+func Multi(client *Client, args []string) {
+	client.Multi = true
+	client.WriteProtocolReply(resp2.NewSimpleStringReply("OK"))
+}
+
+// Exec command will execute a multi transaction
+func Exec(client *Client, args []string) {
+	if !client.Multi {
+		client.WriteError(protocol.ErrExecWithoutMulti{})
+		return
+	}
+
+	client.Multi = false
+	if client.MultiError {
+		client.MultiError = false
+		client.Commands = []*Command{}
+		client.WriteError(protocol.ErrExecAbortTransaction{})
+		return
+	}
+
+	for _, cmd := range client.Commands {
+		cmd.Fn(client, cmd.Args)
+	}
+
+	// clear all commands
+	client.Commands = []*Command{}
+}

internal/protocol/errors.go

@@ -250,3 +250,29 @@ func (ErrUnknownProtocol) Error() string {
 func (ErrUnknownProtocol) Recoverable() bool {
 	return true
 }
+
+// ErrExecAbortTransaction is used to indicate whether an error is occurred while preparing a multi transaction
+type ErrExecAbortTransaction struct {
+}
+
+// Recoverable whether error is recoverable or not
+func (ErrExecAbortTransaction) Recoverable() bool {
+	return true
+}
+
+func (ErrExecAbortTransaction) Error() string {
+	return "EXECABORT Transaction discarded because of previous errors."
+}
+
+// ErrExecWithoutMulti is used to indicate that multi is not enabled
+type ErrExecWithoutMulti struct {
+}
+
+// Recoverable whether error is recoverable or not
+func (ErrExecWithoutMulti) Recoverable() bool {
+	return true
+}
+
+func (ErrExecWithoutMulti) Error() string {
+	return "ERR EXEC without MULTI"
+}