feat(handler): use middleware to handle requests that cannot be routed

Author: timandy

cluster/cluster_worker.go

@@ -35,7 +35,7 @@ func (w workerService) HandleRequest(_ context.Context, req *clusterpb.RequestMe
 		Route: req.Route,
 		Data:  req.Data,
 	}
-	w.node.handler.localProcess(handlerNode, req.Id, s, msg)
+	w.node.handler.localProcess(s, msg, handlerNode)
 	return &clusterpb.MemberHandleResponse{}, nil
 }
 
@@ -53,7 +53,7 @@ func (w workerService) HandleNotify(_ context.Context, req *clusterpb.NotifyMess
 		Route: req.Route,
 		Data:  req.Data,
 	}
-	w.node.handler.localProcess(handler, 0, s, msg)
+	w.node.handler.localProcess(s, msg, handler)
 	return &clusterpb.MemberHandleResponse{}, nil
 }
 

cluster/handler.go

@@ -7,7 +7,6 @@ import (
 	"math/rand"
 	"net"
 	"sort"
-	"strings"
 	"sync"
 	"time"
 
@@ -303,36 +302,20 @@ func (h *LocalHandler) processPacket(agent *agent, p *packet.Packet) error {
 
 // processMessage 处理消息, 分发到本地或远程
 func (h *LocalHandler) processMessage(agent *agent, msg *message.Message) {
-	var lastMid uint64
-	switch msg.Type {
-	case message.Request:
-		lastMid = msg.ID
-	case message.Notify:
-		lastMid = 0
-	default:
-		log.Info("Invalid message type: " + msg.Type.String())
+	if handlerNode, found := h.localHandlers[msg.Route]; found {
+		h.localProcess(agent.session, msg, handlerNode)
 		return
 	}
-	handlerNode, found := h.localHandlers[msg.Route]
-	if !found {
-		h.remoteProcess(agent.session, msg, false)
-	} else {
-		h.localProcess(handlerNode, lastMid, agent.session, msg)
-	}
+	h.remoteProcess(agent.session, msg, false)
 }
 
 // remoteProcess 处理远程消息
 func (h *LocalHandler) remoteProcess(session *session.Session, msg *message.Message, noCopy bool) {
-	index := strings.LastIndex(msg.Route, ".")
-	if index < 0 {
-		log.Info("nano/handler: invalid route %s", msg.Route)
-		return
-	}
-
-	service := msg.Route[:index]
+	service := msg.Service()
 	members := h.findMembers(service)
 	if len(members) == 0 {
-		log.Info("nano/handler: %s not found(forgot registered?)", msg.Route)
+		// 没有服务节点, 转本地处理 404
+		h.localProcess(session, msg, nil)
 		return
 	}
 
@@ -347,7 +330,11 @@ func (h *LocalHandler) remoteProcess(session *session.Session, msg *message.Mess
 		} else {
 			member := h.node.opts.RemoteServiceRoute(service, session, members)
 			if member == nil {
-				log.Info("customize remoteServiceRoute handler: %s is not found", msg.Route)
+				if env.Debug {
+					log.Info("customize remoteServiceRoute handler: %s is not found", msg.Route)
+				}
+				// 有服务节点, 但自定义路由没匹配到, 转本地处理 404
+				h.localProcess(session, msg, nil)
 				return
 			}
 			remoteAddr = member.ServiceAddr
@@ -407,21 +394,27 @@ func (h *LocalHandler) remoteProcess(session *session.Session, msg *message.Mess
 }
 
 // localProcess 本地处理
-func (h *LocalHandler) localProcess(handlerNode *npi.HandlerNode, lastMid uint64, session *session.Session, msg *message.Message) {
+func (h *LocalHandler) localProcess(session *session.Session, msg *message.Message, handlerNode *npi.HandlerNode) {
+	// 计算响应 ID
+	lastMid, err := msg.ResponseID()
+	if err != nil {
+		log.Info(err.Error() + ":" + msg.Type.String())
+		return
+	}
+
+	// 计算服务名
+	service := msg.Service()
+
+	// 执行管道任务
 	if pipe := h.pipeline; pipe != nil {
-		err := pipe.Inbound().Process(session, msg)
+		err = pipe.Inbound().Process(session, msg)
 		if err != nil {
 			log.Error("Pipeline process failed.", err)
 			return
 		}
 	}
-	index := strings.LastIndex(msg.Route, ".")
-	if index < 0 {
-		log.Info("nano/handler: invalid route %s", msg.Route)
-		return
-	}
-	service := msg.Route[:index]
 
+	// 记录日志
 	if env.Debug {
 		log.Info("UID=%d, Message={%s}, Data=%v", session.UID(), msg.String(), msg.Data)
 	}

cluster/handler_task.go

@@ -1,6 +1,8 @@
 package cluster
 
 import (
+	"github.com/lonng/nano/internal/env"
+	"github.com/lonng/nano/internal/log"
 	"github.com/lonng/nano/npi"
 	"github.com/lonng/nano/protocal/message"
 	"github.com/lonng/nano/session"
@@ -9,7 +11,7 @@ import (
 // 任务对象
 type localHandlerTask struct {
 	localHandler *LocalHandler
-	handlerNode  *npi.HandlerNode
+	handlerNode  *npi.HandlerNode // 只有该字段允许为 nil
 	session      *session.Session
 	msg          *message.Message
 	service      string
@@ -18,6 +20,7 @@ type localHandlerTask struct {
 
 // run 运行任务
 func (task *localHandlerTask) run() {
+	// 定义变量
 	h := task.localHandler
 	handlerNode := task.handlerNode
 	s := task.session
@@ -47,12 +50,15 @@ func (task *localHandlerTask) run() {
 	c.Msg = msg
 	c.Session = s
 	// 有路由
-	if handlerNode.Len() > 0 {
+	if handlerNode != nil && handlerNode.Len() > 0 {
 		c.HandlerNode = handlerNode
 		c.Next()
 		return
 	}
 	// 无路由
+	if env.Debug {
+		log.Info("nano/handler: %s not found(forgot registered?)", msg.Route)
+	}
 	c.HandlerNode = h.allNoRoutes
 	c.Next()
 }

protocal/message/message.go

@@ -2,9 +2,10 @@ package message
 
 import (
 	"fmt"
+	"strings"
 )
 
-// Message represents a unmarshalled message or a message which to be marshaled
+// Message represents an unmarshalled message or a message which to be marshaled
 type Message struct {
 	Type       Type   // message type
 	ID         uint64 // unique id, zero while notify mode
@@ -18,6 +19,27 @@ func New() *Message {
 	return &Message{}
 }
 
+// ResponseID 获取响应 ID
+func (m *Message) ResponseID() (uint64, error) {
+	switch m.Type {
+	case Request:
+		return m.ID, nil
+	case Notify:
+		return 0, nil
+	default:
+		return 0, ErrWrongMessageType
+	}
+}
+
+// Service 获取服务名
+func (m *Message) Service() string {
+	route := m.Route
+	if index := strings.LastIndex(route, "."); index != -1 {
+		return route[:index]
+	}
+	return ""
+}
+
 // String, implementation of fmt.Stringer interface
 func (m *Message) String() string {
 	return fmt.Sprintf("%s %s (%dbytes)", types[m.Type], m.Route, len(m.Data))