refactor code structure for gnet

redis/server/std/pubsub_test.go

@@ -0,0 +1,50 @@
+package std
+
+import (
+	"github.com/hdt3213/godis/lib/utils"
+	"github.com/hdt3213/godis/pubsub"
+	"github.com/hdt3213/godis/redis/connection"
+	"github.com/hdt3213/godis/redis/parser"
+	"github.com/hdt3213/godis/redis/protocol/asserts"
+	"testing"
+)
+
+func TestPublish(t *testing.T) {
+	hub := pubsub.MakeHub()
+	channel := utils.RandString(5)
+	msg := utils.RandString(5)
+	conn := connection.NewFakeConn()
+	pubsub.Subscribe(hub, conn, utils.ToCmdLine(channel))
+	conn.Clean() // clean subscribe success
+	pubsub.Publish(hub, utils.ToCmdLine(channel, msg))
+	data := conn.Bytes()
+	ret, err := parser.ParseOne(data)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	asserts.AssertMultiBulkReply(t, ret, []string{
+		"message",
+		channel,
+		msg,
+	})
+
+	// unsubscribe
+	pubsub.UnSubscribe(hub, conn, utils.ToCmdLine(channel))
+	conn.Clean()
+	pubsub.Publish(hub, utils.ToCmdLine(channel, msg))
+	data = conn.Bytes()
+	if len(data) > 0 {
+		t.Error("expect no msg")
+	}
+
+	// unsubscribe all
+	pubsub.Subscribe(hub, conn, utils.ToCmdLine(channel))
+	pubsub.UnSubscribe(hub, conn, utils.ToCmdLine())
+	conn.Clean()
+	pubsub.Publish(hub, utils.ToCmdLine(channel, msg))
+	data = conn.Bytes()
+	if len(data) > 0 {
+		t.Error("expect no msg")
+	}
+}

redis/server/std/server.go

@@ -0,0 +1,125 @@
+package std
+
+/*
+ * A tcp.Handler implements redis protocol
+ */
+
+import (
+	"context"
+	"io"
+	"net"
+	"strings"
+	"sync"
+
+	"github.com/hdt3213/godis/cluster"
+	"github.com/hdt3213/godis/config"
+	"github.com/hdt3213/godis/database"
+	idatabase "github.com/hdt3213/godis/interface/database"
+	"github.com/hdt3213/godis/lib/logger"
+	"github.com/hdt3213/godis/lib/sync/atomic"
+	"github.com/hdt3213/godis/redis/connection"
+	"github.com/hdt3213/godis/redis/parser"
+	"github.com/hdt3213/godis/redis/protocol"
+	"github.com/hdt3213/godis/tcp"
+)
+
+var (
+	unknownErrReplyBytes = []byte("-ERR unknown\r\n")
+)
+
+// Handler implements tcp.Handler and serves as a redis server
+type Handler struct {
+	activeConn sync.Map // *client -> placeholder
+	db         idatabase.DB
+	closing    atomic.Boolean // refusing new client and new request
+}
+
+// MakeHandler creates a Handler instance
+func MakeHandler() *Handler {
+	var db idatabase.DB
+	if config.Properties.ClusterEnable {
+		db = cluster.MakeCluster()
+	} else {
+		db = database.NewStandaloneServer()
+	}
+	return &Handler{
+		db: db,
+	}
+}
+
+func Serve(addr string, handler *Handler) error {
+	return tcp.ListenAndServeWithSignal(&tcp.Config{
+		Address: addr,
+	}, handler)
+}
+
+func (h *Handler) closeClient(client *connection.Connection) {
+	_ = client.Close()
+	h.db.AfterClientClose(client)
+	h.activeConn.Delete(client)
+}
+
+
+// Handle receives and executes redis commands
+func (h *Handler) Handle(ctx context.Context, conn net.Conn) {
+	if h.closing.Get() {
+		// closing handler refuse new connection
+		_ = conn.Close()
+		return
+	}
+
+	client := connection.NewConn(conn)
+	h.activeConn.Store(client, struct{}{})
+
+	ch := parser.ParseStream(conn)
+	for payload := range ch {
+		if payload.Err != nil {
+			if payload.Err == io.EOF ||
+				payload.Err == io.ErrUnexpectedEOF ||
+				strings.Contains(payload.Err.Error(), "use of closed network connection") {
+				// connection closed
+				h.closeClient(client)
+				logger.Info("connection closed: " + client.RemoteAddr())
+				return
+			}
+			// protocol err
+			errReply := protocol.MakeErrReply(payload.Err.Error())
+			_, err := client.Write(errReply.ToBytes())
+			if err != nil {
+				h.closeClient(client)
+				logger.Info("connection closed: " + client.RemoteAddr())
+				return
+			}
+			continue
+		}
+		if payload.Data == nil {
+			logger.Error("empty payload")
+			continue
+		}
+		r, ok := payload.Data.(*protocol.MultiBulkReply)
+		if !ok {
+			logger.Error("require multi bulk protocol")
+			continue
+		}
+		result := h.db.Exec(client, r.Args)
+		if result != nil {
+			_, _ = client.Write(result.ToBytes())
+		} else {
+			_, _ = client.Write(unknownErrReplyBytes)
+		}
+	}
+}
+
+// Close stops handler
+func (h *Handler) Close() error {
+	logger.Info("handler shutting down...")
+	h.closing.Set(true)
+	// TODO: concurrent wait
+	h.activeConn.Range(func(key interface{}, val interface{}) bool {
+		client := key.(*connection.Connection)
+		_ = client.Close()
+		return true
+	})
+	h.db.Close()
+	return nil
+}

redis/server/std/server_test.go

@@ -0,0 +1,44 @@
+package std
+
+import (
+	"bufio"
+	"github.com/hdt3213/godis/tcp"
+	"net"
+	"testing"
+	"time"
+)
+
+func TestListenAndServe(t *testing.T) {
+	var err error
+	closeChan := make(chan struct{})
+	listener, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	addr := listener.Addr().String()
+	go tcp.ListenAndServe(listener, MakeHandler(), closeChan)
+
+	conn, err := net.Dial("tcp", addr)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	_, err = conn.Write([]byte("PING\r\n"))
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	bufReader := bufio.NewReader(conn)
+	line, _, err := bufReader.ReadLine()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if string(line) != "+PONG" {
+		t.Error("get wrong response")
+		return
+	}
+	closeChan <- struct{}{}
+	time.Sleep(time.Second)
+}