fixed bug of the recover

cluster/node/node.go

@@ -7,6 +7,7 @@ import (
 	"github.com/dobyte/due/log"
 	"github.com/dobyte/due/registry"
 	"github.com/dobyte/due/transport"
+	"github.com/dobyte/due/utils/xcall"
 	"github.com/dobyte/due/utils/xnet"
 	"sync/atomic"
 	"time"
@@ -115,17 +116,21 @@ func (n *Node) dispatch() {
 			if !ok {
 				return
 			}
-			n.events.handle(evt)
+			xcall.Call(func() {
+				n.events.handle(evt)
+			})
 		case ctx, ok := <-n.router.receive():
 			if !ok {
 				return
 			}
-			n.router.handle(ctx)
+			xcall.Call(func() {
+				n.router.handle(ctx)
+			})
 		case handle, ok := <-n.fnChan:
 			if !ok {
 				return
 			}
-			handle()
+			xcall.Call(handle)
 		}
 	}
 }

container.go

@@ -33,15 +33,6 @@ func (c *Container) Add(components ...component.Component) {
 func (c *Container) Serve() {
 	log.Debug(fmt.Sprintf("Welcome to the due framework %s, Learn more at %s", Version, Website))
 
-	defer func() {
-		switch err := recover(); err.(type) {
-		case runtime.Error:
-			log.Panicf("due runtime error: %v", err)
-		default:
-			log.Panicf("due error: %v", err)
-		}
-	}()
-
 	for _, comp := range c.components {
 		comp.Init()
 	}

packet/packer.go

@@ -8,9 +8,10 @@ import (
 )
 
 var (
-	ErrMessageIsNil  = errors.New("the message is nil")
-	ErrSeqOverflow   = errors.New("the message seq overflow")
-	ErrRouteOverflow = errors.New("the message route overflow")
+	ErrMessageIsNil   = errors.New("the message is nil")
+	ErrSeqOverflow    = errors.New("the message seq overflow")
+	ErrRouteOverflow  = errors.New("the message route overflow")
+	ErrInvalidMessage = errors.New("invalid message")
 )
 
 type Packer interface {
@@ -98,10 +99,16 @@ func (p *defaultPacker) Pack(message *Message) ([]byte, error) {
 
 // Unpack 解包消息
 func (p *defaultPacker) Unpack(data []byte) (*Message, error) {
+	ln := len(data) - p.opts.seqBytesLen - p.opts.routeBytesLen
+
+	if ln < 0 {
+		return nil, ErrInvalidMessage
+	}
+
 	var (
 		err     error
 		reader  = bytes.NewReader(data)
-		message = &Message{Buffer: make([]byte, reader.Len()-p.opts.seqBytesLen-p.opts.routeBytesLen)}
+		message = &Message{Buffer: make([]byte, ln)}
 	)
 
 	switch p.opts.seqBytesLen {

utils/xcall/call.go

@@ -0,0 +1,21 @@
+package xcall
+
+import (
+	"github.com/dobyte/due/log"
+	"runtime"
+)
+
+func Call(fn func()) {
+	defer func() {
+		if err := recover(); err != nil {
+			switch err.(type) {
+			case runtime.Error:
+				log.Panic(err)
+			default:
+				log.Panicf("panic error: %v", err)
+			}
+		}
+	}()
+
+	fn()
+}