第 5 章

[Day 5] Redis Command Handler

`Command{}` and reader

“Redis Command” is a special data type not defined in RESP. Actually, it just RESP array with multiple bulk string.

I defined a Command{} in pkg/resp/cmd.go for converting from RSEP array.

package resp

import (
	"errors"
	"fmt"
)

var (
	ErrProtocol = errors.New("Protocol error")
)

type Command struct {
	raw  Value
	cmd  BulkString
	args []BulkString
}

func ReadCommand(rd *Reader) (*Command, error) {
	v, err := rd.Read()
	if err != nil {
		return nil, err
	}

	arr, ok := v.(Array)
	if !ok {
		return nil, fmt.Errorf("%w: expected array, got %T(%+v)", ErrProtocol, v, v)
	}
	if arr.null || len(arr.data) == 0 {
		return nil, fmt.Errorf("%w: empty array", ErrProtocol)
	}

	cmd, ok := arr.data[0].(BulkString)
	if !ok {
		return nil, fmt.Errorf("%w: command expected bulk string, got %T(%+v)", ErrProtocol, arr.data[0], arr.data[0])
	}

	args := make([]BulkString, 0, len(arr.data)-1)
	for i := range arr.data[1:] {
		arg, ok := arr.data[i+1].(BulkString)
		if !ok {
			return nil, fmt.Errorf("%w: argument [%d] expected bulk string, got %T(%+v)", ErrProtocol, i, arr.data[i+1], arr.data[i+1])
		}
		args = append(args, arg)
	}

	return &Command{
		raw:  v,
		cmd:  cmd,
		args: args,
	}, nil
}

Apply `resp.ReadCommand` in server

-	cmd, err := rd.Read()
+	cmd, err := resp.ReadCommand(rd)

func (s *simpleSrv) serve(conn net.Conn) {
	defer conn.Close()

	rd := resp.NewReader(conn)
	cmd, err := resp.ReadCommand(rd)
	if err != nil {
		slog.Error("failed to read from conn:", slog.Any("error", err))
		return
	}

	slog.Info("Read RESP command:", slog.Any("command", cmd))

	if _, err := conn.Write(resp.SimpleString("OK").Marshal()); err != nil {
		slog.Error("failed to write to conn:", slog.Any("error", err))
		return
	}
}

Server and Handler

In a small service, handler is a part of server, because server “handles” connections. However, when service scaling, the duty of server and handler are different.

Server: Listen a port, and accept connections.
Handler: Read requests from connections, do something and write response to connections.

`Handler{}` interface and implementation

I created internal/server/handler.go and add Handler{} interface.

// file: internal/server/handler.go
package server

type Handler interface {
	ServeRESP(context.Context, *resp.Reader) (resp.Value, error)
}

func NewHandler() Handler {
	return &simpleHandler{}
}

type simpleHandler struct {}

func (h *simpleHandler) ServeRESP(ctx context.Context, rd *resp.Reader) (resp.Value, error) {
	cmd, err := resp.ReadCommand(rd)
	if err != nil {
		return nil, err
	}

	slog.Info("Read RESP command:", slog.Any("command", cmd))

	return resp.SimpleString("OK"), nil
}

Update internal/server/server.go:

// file: internal/server/server.go
func NewServer() Server {
	return &simpleSrv{
		handler: NewHandler(),
	}
}

func (s *simpleSrv) serve(conn net.Conn) {
	defer conn.Close()

	rd := resp.NewReader(conn)

	ret, err := s.handler.ServeRESP(context.Background(), rd)
	if err != nil {
		return
	}
	if ret == nil { // avoid panic
		ret = resp.SimpleString("OK")
	}

	if _, err := conn.Write(ret.Marshal()); err != nil {
		slog.Error("failed to write to conn:", slog.Any("error", err))
		return
	}
}

Dependency Injection

I used NewHandler() in NewServer() constructor, it makes Server{} depends on Handler{}, it’s a bad smell for this pattern, because no matter what I do, the Server{} requires Handler{} anyway.

There’s a solution: dependency injection, the NewServer() construct needs Handler{} as argument, we can free to choose other handler implementation when constructing Server{}.

func NewServer(handler Handler) Server {
	return &simpleSrv{
		handler: handler,
	}
}

And we need to update cmd/olivine/app.go:

func NewApp() *App {
	return &App{
		srv: server.NewServer(server.NewHandler()),
	}
}

Note: when the service scaling, we can use dependency injection library / tool to mangae dependencies, but it is no needed yet.

Multiple Commands in single connection

As I used redis-cli to check this server, I found a strange behavior.

$ redis-cli -p 16879
(127.0.0.1:16879) > ping
Error: Server closed the connection
(not connected) > ping
OK
(127.0.0.1:16879) > ping
Error: Server closed the connection

The s.serve() uses defer conn.Close() for each connection be processed in internal/server/server.go. We need to reconnect to server for every command.

It takes a lot of resources for multiple TCP handshakes, and redis-cli is also designed for multiple commands in single connection for a client.

Use a for loop for handling commands:

// file: internal/server/server.go
func (s *simpleSrv) serve(conn net.Conn) {
	defer conn.Close()

	rd := resp.NewReader(conn)

	for {
		ret, err := s.handler.ServeRESP(context.Background(), rd)
		if err != nil {
			return
		}

		if _, err := conn.Write(ret.Marshal()); err != nil {
			slog.Error("failed to write to conn:", slog.Any("error", err))
			return
		}
	}

}

Error Handling

The ret, err := s.handler.ServeRESP(context.Background(), rd) will return and close connection immediately when error occurred, that’s no my expectation.

The err from handler.ServeRESP() could be 2 types:

Client Error: “Unknown command”, “Mismatch arguments”, …
Server Error: “Connection failed”, “Out of memory”, …

Connections can be closed when server error, but clients usually hope that keep connection alive if client error.

Defines the 2 types of error in internal/server/handler.go:

var (
	ErrClient = errors.New("client error")
	ErrServer = errors.New("server error")
)

func (h *simpleHandler) ServeRESP(ctx context.Context, rd *resp.Reader) (resp.Value, error) {
	cmd, err := resp.ReadCommand(rd)
	if err != nil {
		if errors.Is(err, resp.ErrProtocol) {
			return resp.NewSimpleError(err), fmt.Errorf("%w: %w", ErrClient, err)
		} else {
			return nil, fmt.Errorf("%w: %w", ErrServer, err)
		}
	}

	slog.Info("Read RESP command:", slog.Any("command", cmd))

Command Factory

Factory Pattern is an important design pattern, because it can make developers focus on single feature(redis command, in this service).

Add a factory in simpleHandler{}:

type HandlerFunc func(context.Context, *resp.Command) (resp.Value, error)

func NewHandler() Handler {
	return &simpleHandler{
		factory: make(map[string]HandlerFunc),
	}
}

type simpleHandler struct {
	factory map[string]HandlerFunc
}

Define a function exec() for simpleHandler{}:

func (h *simpleHandler) ServeRESP(ctx context.Context, rd *resp.Reader) (resp.Value, error) {
	cmd, err := resp.ReadCommand(rd)
	if err != nil {
		if errors.Is(err, resp.ErrProtocol) {
			return resp.NewSimpleError(err), fmt.Errorf("%w: %w", ErrClient, err)
		} else {
			return nil, fmt.Errorf("%w: %w", ErrServer, err)
		}
	}

	slog.Info("Read RESP command:", slog.Any("command", cmd))

	return h.exec(ctx, cmd)
}

func (h *simpleHandler) exec(ctx context.Context, cmd *resp.Command) (resp.Value, error) {
	f, ok := h.factory[cmd.Command()]
	if !ok {
		err := fmt.Errorf("%w: unknown command: %s", ErrClient, cmd.Command())
		return resp.NewSimpleError(err), err
	}

	return f(ctx, cmd)
}

The *resp.Command.Command() is not defined now, we need to add it in pkg/resp/cmd.go:

func (cmd *Command) Command() string {
	return strings.ToUpper(string(cmd.cmd.data))
}

Command{} and reader

Apply resp.ReadCommand in server