Add recover to msg handlers.

Add more tests to stanmsg.

Author: huangjunwen

natsrpc/natsrpc_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log"
 	"math"
 	"os"
@@ -65,9 +66,9 @@ func newRawMessage(s string) *json.RawMessage {
 	return &r
 }
 
-func TestNatsRPC(t *testing.T) {
+func TestCall(t *testing.T) {
 	log.Printf("\n")
-	log.Printf(">>> TestNatsRPC.\n")
+	log.Printf(">>> TestCall.\n")
 	var err error
 	assert := assert.New(t)
 
@@ -329,6 +330,18 @@ func TestNatsRPC(t *testing.T) {
 		}
 	)
 
+	var (
+		panicSpec = MustRPCSpec(
+			svcName,
+			"panic",
+			func() interface{} { return &emptypb.Empty{} },
+			func() interface{} { return &emptypb.Empty{} },
+		)
+		panicHandler = func(ctx context.Context, in interface{}) (interface{}, error) {
+			panic(fmt.Errorf("Something goes wrong"))
+		}
+	)
+
 	// Regist handlers.
 	if err := server.RegistHandler(sqrtSpec, sqrtHandler); err != nil {
 		log.Panic(err)
@@ -348,6 +361,9 @@ func TestNatsRPC(t *testing.T) {
 	if err := server.RegistHandler(blockSpec, blockHandler); err != nil {
 		log.Panic(err)
 	}
+	if err := server.RegistHandler(panicSpec, panicHandler); err != nil {
+		log.Panic(err)
+	}
 
 	{
 		// Regist and deregist
@@ -617,7 +633,7 @@ func TestNatsRPC(t *testing.T) {
 			ExpectOutput: nil,
 			ExpectError:  true,
 		},
-		// block handlers.
+		// block to test handler busy.
 		{
 			Client:     jsonClient,
 			ClientName: "json",
@@ -639,6 +655,17 @@ func TestNatsRPC(t *testing.T) {
 			ExpectOutput: nil,
 			ExpectError:  true,
 		},
+		// handler panic.
+		{
+			Client:     pbClient,
+			ClientName: "pb",
+			Spec:       panicSpec,
+			GenInput: func() (context.Context, interface{}) {
+				return context.Background(), &emptypb.Empty{}
+			},
+			ExpectOutput: nil,
+			ExpectError:  true,
+		},
 	}
 	for i, testCase := range testCases {
 		testCase := testCase
@@ -672,9 +699,9 @@ func TestNatsRPC(t *testing.T) {
 
 }
 
-func TestPbJsonAndFanout(t *testing.T) {
+func TestFanout(t *testing.T) {
 	log.Printf("\n")
-	log.Printf(">>> TestPbJsonAndFanout.\n")
+	log.Printf(">>> TestFanout.\n")
 	var err error
 	assert := assert.New(t)
 

natsrpc/server.go

@@ -219,6 +219,15 @@ func (sc *ServerConn) msgHandler(svcName string, mm *methodMap) nats.MsgHandler
 		}
 
 		if err := sc.runner.Submit(func() {
+			defer func() {
+				if e := recover(); e != nil {
+					err, ok := e.(error)
+					if !ok {
+						err = fmt.Errorf("%+v", e)
+					}
+					logger().Error(err, "handler panic")
+				}
+			}()
 
 			info := mm.Lookup(methodName)
 			if info == nil {

stanmsg/durconn.go

@@ -361,11 +361,20 @@ func (dc *DurConn) subscribe(sub *subscription, sc stan.Conn) error {
 
 func (dc *DurConn) msgHandler(sub *subscription) stan.MsgHandler {
 
-	logger := dc.logger.WithValues("subject", sub.spec.SubjectName, "queue", sub.queue)
+	logger := dc.logger.WithValues("subject", sub.spec.SubjectName(), "queue", sub.queue)
 
 	return func(stanMsg *stan.Msg) {
 
 		if err := dc.runner.Submit(func() {
+			defer func() {
+				if e := recover(); e != nil {
+					err, ok := e.(error)
+					if !ok {
+						err = fmt.Errorf("%+v", e)
+					}
+					logger.Error(err, "handler panic")
+				}
+			}()
 
 			m := &nppbmsg.MessageWithMD{}
 			if err := proto.Unmarshal(stanMsg.Data, m); err != nil {