various fixes (#1267)

* logger: remove Panic log level

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* server/grpc: add missing Unlock in Subscribe error

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* server: minor code change

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* server/grpc: extend test suite with pub/sub testing

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* server/grpc: fix invalid check and allow subscriber error to be returned

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* server/grpc: add pubsub tests

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

* client/grpc: check for nil req/rsp

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

client/grpc/grpc.go

@@ -383,6 +383,11 @@ func (g *grpcClient) NewRequest(service, method string, req interface{}, reqOpts
 }
 
 func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
+	if req == nil {
+		return errors.InternalServerError("go.micro.client", "req is nil")
+	} else if rsp == nil {
+		return errors.InternalServerError("go.micro.client", "rsp is nil")
+	}
 	// make a copy of call opts
 	callOpts := g.opts.CallOptions
 	for _, opt := range opts {

logger/level.go

@@ -16,8 +16,6 @@ const (
 	WarnLevel
 	// ErrorLevel level. Logs. Used for errors that should definitely be noted.
 	ErrorLevel
-	// PanicLevel level, logs the message and then panics.
-	PanicLevel
 	// FatalLevel level. Logs and then calls `logger.Exit(1)`. highest level of severity.
 	FatalLevel
 )
@@ -34,8 +32,6 @@ func (l Level) String() string {
 		return "warn"
 	case ErrorLevel:
 		return "error"
-	case PanicLevel:
-		return "panic"
 	case FatalLevel:
 		return "fatal"
 	}
@@ -61,12 +57,10 @@ func GetLevel(levelStr string) (Level, error) {
 		return WarnLevel, nil
 	case ErrorLevel.String():
 		return ErrorLevel, nil
-	case PanicLevel.String():
-		return PanicLevel, nil
 	case FatalLevel.String():
 		return FatalLevel, nil
 	}
-	return InfoLevel, fmt.Errorf("Unknown Level String: '%s', defaulting to NoLevel", levelStr)
+	return InfoLevel, fmt.Errorf("Unknown Level String: '%s', defaulting to InfoLevel", levelStr)
 }
 
 func Info(args ...interface{}) {
@@ -109,14 +103,6 @@ func Errorf(template string, args ...interface{}) {
 	DefaultLogger.Logf(ErrorLevel, template, args...)
 }
 
-func Panic(args ...interface{}) {
-	DefaultLogger.Log(PanicLevel, args...)
-}
-
-func Panicf(template string, args ...interface{}) {
-	DefaultLogger.Logf(PanicLevel, template, args...)
-}
-
 func Fatal(args ...interface{}) {
 	DefaultLogger.Log(FatalLevel, args...)
 }

server/grpc/grpc.go

@@ -555,8 +555,8 @@ func (g *grpcServer) Subscribe(sb server.Subscriber) error {
 	}
 
 	g.Lock()
-
 	if _, ok = g.subscribers[sub]; ok {
+		g.Unlock()
 		return fmt.Errorf("subscriber %v already exists", sub)
 	}
 

server/grpc/grpc_test.go

@@ -1,12 +1,19 @@
-package grpc
+package grpc_test
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
+	"github.com/micro/go-micro/v2"
+	bmemory "github.com/micro/go-micro/v2/broker/memory"
+	"github.com/micro/go-micro/v2/client"
+	gcli "github.com/micro/go-micro/v2/client/grpc"
 	"github.com/micro/go-micro/v2/errors"
-	"github.com/micro/go-micro/v2/registry/memory"
+	rmemory "github.com/micro/go-micro/v2/registry/memory"
 	"github.com/micro/go-micro/v2/server"
+	gsrv "github.com/micro/go-micro/v2/server/grpc"
+	tgrpc "github.com/micro/go-micro/v2/transport/grpc"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/status"
 
@@ -14,7 +21,17 @@ import (
 )
 
 // server is used to implement helloworld.GreeterServer.
-type testServer struct{}
+type testServer struct {
+	msgCount int
+}
+
+func (s *testServer) Handle(ctx context.Context, msg *pb.Request) error {
+	s.msgCount++
+	return nil
+}
+func (s *testServer) HandleError(ctx context.Context, msg *pb.Request) error {
+	return fmt.Errorf("fake")
+}
 
 // TestHello implements helloworld.GreeterServer
 func (s *testServer) Call(ctx context.Context, req *pb.Request, rsp *pb.Response) error {
@@ -26,14 +43,75 @@ func (s *testServer) Call(ctx context.Context, req *pb.Request, rsp *pb.Response
 	return nil
 }
 
-func TestGRPCServer(t *testing.T) {
-	r := memory.NewRegistry()
-	s := NewServer(
+/*
+func BenchmarkServer(b *testing.B) {
+	r := rmemory.NewRegistry()
+	br := bmemory.NewBroker()
+	tr := tgrpc.NewTransport()
+	s := gsrv.NewServer(
+		server.Broker(br),
 		server.Name("foo"),
 		server.Registry(r),
+		server.Transport(tr),
 	)
+	c := gcli.NewClient(
+		client.Registry(r),
+		client.Broker(br),
+		client.Transport(tr),
+	)
+	ctx := context.TODO()
 
-	pb.RegisterTestHandler(s, &testServer{})
+	h := &testServer{}
+	pb.RegisterTestHandler(s, h)
+	if err := s.Start(); err != nil {
+		b.Fatalf("failed to start: %v", err)
+	}
+
+	// check registration
+	services, err := r.GetService("foo")
+	if err != nil || len(services) == 0 {
+		b.Fatalf("failed to get service: %v # %d", err, len(services))
+	}
+
+	defer func() {
+		if err := s.Stop(); err != nil {
+			b.Fatalf("failed to stop: %v", err)
+		}
+	}()
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		c.Call()
+	}
+
+}
+*/
+func TestGRPCServer(t *testing.T) {
+	r := rmemory.NewRegistry()
+	b := bmemory.NewBroker()
+	tr := tgrpc.NewTransport()
+	s := gsrv.NewServer(
+		server.Broker(b),
+		server.Name("foo"),
+		server.Registry(r),
+		server.Transport(tr),
+	)
+	c := gcli.NewClient(
+		client.Registry(r),
+		client.Broker(b),
+		client.Transport(tr),
+	)
+	ctx := context.TODO()
+
+	h := &testServer{}
+	pb.RegisterTestHandler(s, h)
+
+	if err := micro.RegisterSubscriber("test_topic", s, h.Handle); err != nil {
+		t.Fatal(err)
+	}
+	if err := micro.RegisterSubscriber("error_topic", s, h.HandleError); err != nil {
+		t.Fatal(err)
+	}
 
 	if err := s.Start(); err != nil {
 		t.Fatalf("failed to start: %v", err)
@@ -51,6 +129,22 @@ func TestGRPCServer(t *testing.T) {
 		}
 	}()
 
+	pub := micro.NewEvent("test_topic", c)
+	pubErr := micro.NewEvent("error_topic", c)
+	cnt := 4
+	for i := 0; i < cnt; i++ {
+		if err = pub.Publish(ctx, &pb.Request{Name: fmt.Sprintf("msg %d", i)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if h.msgCount != cnt {
+		t.Fatalf("pub/sub not work, or invalid message count %d", h.msgCount)
+	}
+	if err = pubErr.Publish(ctx, &pb.Request{}); err == nil {
+		t.Fatal("this must return error, as we return error from handler")
+	}
+
 	cc, err := grpc.Dial(s.Options().Address, grpc.WithInsecure())
 	if err != nil {
 		t.Fatalf("failed to dial server: %v", err)

server/grpc/subscriber.go

@@ -246,18 +246,19 @@ func (g *grpcServer) createSubHandler(sb *subscriber, opts server.Options) broke
 				if g.wg != nil {
 					defer g.wg.Done()
 				}
-				results <- fn(ctx, &rpcMessage{
+				err := fn(ctx, &rpcMessage{
 					topic:       sb.topic,
 					contentType: ct,
 					payload:     req.Interface(),
 					header:      msg.Header,
 					body:        msg.Body,
 				})
+				results <- err
 			}()
 		}
 		var errors []string
 		for i := 0; i < len(sb.handlers); i++ {
-			if rerr := <-results; err != nil {
+			if rerr := <-results; rerr != nil {
 				errors = append(errors, rerr.Error())
 			}
 		}

server/rpc_router.go

@@ -505,7 +505,6 @@ func (router *router) Subscribe(s Subscriber) error {
 }
 
 func (router *router) ProcessMessage(ctx context.Context, msg Message) (err error) {
-
 	defer func() {
 		// recover any panics
 		if r := recover(); r != nil {
@@ -516,16 +515,13 @@ func (router *router) ProcessMessage(ctx context.Context, msg Message) (err erro
 	}()
 
 	router.su.RLock()
-
 	// get the subscribers by topic
 	subs, ok := router.subscribers[msg.Topic()]
-	if !ok {
-		router.su.RUnlock()
-		return nil
-	}
-
 	// unlock since we only need to get the subs
 	router.su.RUnlock()
+	if !ok {
+		return nil
+	}
 
 	var errResults []string