update for latest micro

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

go.mod

@@ -2,4 +2,12 @@ module go.unistack.org/micro-wrapper-validator/v3
 
 go 1.17
 
-require go.unistack.org/micro/v3 v3.10.14
+require go.unistack.org/micro/v3 v3.10.66
+
+require (
+	github.com/golang/protobuf v1.5.4 // indirect
+	golang.org/x/sys v0.19.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be // indirect
+	google.golang.org/grpc v1.63.2 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
+)

go.sum

@@ -1,8 +1,21 @@
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg=
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
 go.unistack.org/micro/v3 v3.10.14 h1:7fgLpwGlCN67twhwtngJDEQvrMkUBDSA5vzZqxIDqNE=
 go.unistack.org/micro/v3 v3.10.14/go.mod h1:uMAc0U/x7dmtICCrblGf0ZLgYegu3VwQAquu+OFCw1Q=
+go.unistack.org/micro/v3 v3.10.66 h1:tiG8HnyTC71IZWSC2qT/DmLhJinZJL9qvw+4Fvpm3d4=
+go.unistack.org/micro/v3 v3.10.66/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg=
+golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
+golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be h1:LG9vZxsWGOmUKieR8wPAUR3u3MpnYFQZROPIMaXh7/A=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be/go.mod h1:WtryC6hu0hhx87FDGxWCDptyssuo68sk10vYjF+T9fY=
+google.golang.org/grpc v1.63.2 h1:MUeiw1B2maTVZthpU5xvASfTh3LDbxHd6IJ6QQVU+xM=
+google.golang.org/grpc v1.63.2/go.mod h1:WAX/8DgncnokcFUldAxq7GeB5DXHDbMF+lLvDomNkRA=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

validator.go

@@ -45,11 +45,25 @@ type Options struct {
 	ServerErrorFn          ServerErrorFunc
 	PublishErrorFn         PublishErrorFunc
 	SubscribeErrorFn       SubscribeErrorFunc
+	ClientValidateResponse bool
+	ServerValidateResponse bool
 }
 
 // Option func signature
 type Option func(*Options)
 
+func ClientValidateResponse(b bool) Option {
+	return func(o *Options) {
+		o.ClientValidateResponse = b
+	}
+}
+
+func ServerValidateResponse(b bool) Option {
+	return func(o *Options) {
+		o.ClientValidateResponse = b
+	}
+}
+
 func ClientReqErrorFn(fn ClientErrorFunc) Option {
 	return func(o *Options) {
 		o.ClientErrorFn = fn
@@ -87,108 +101,94 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
+func NewHook(opts ...Option) *hook {
+	return &hook{opts: NewOptions(opts...)}
+}
+
 type validator interface {
 	Validate() error
 }
 
-type wrapper struct {
+type hook struct {
-	client.Client
 	opts Options
 }
 
-func NewClientWrapper(opts ...Option) client.Wrapper {
+func (w *hook) ClientCall(next client.FuncCall) client.FuncCall {
-	return func(c client.Client) client.Client {
+	return func(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
-		handler := &wrapper{
-			Client: c,
-			opts:   NewOptions(opts...),
-		}
-		return handler
-	}
-}
-
-func NewClientCallWrapper(opts ...Option) client.CallWrapper {
-	options := NewOptions(opts...)
-	return func(fn client.CallFunc) client.CallFunc {
-		return func(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
-			if v, ok := req.Body().(validator); ok {
-				if verr := v.Validate(); verr != nil {
-					return options.ClientErrorFn(req, nil, verr)
-				}
-			}
-			err := fn(ctx, addr, req, rsp, opts)
-			if v, ok := rsp.(validator); ok {
-				if verr := v.Validate(); verr != nil {
-					return options.ClientErrorFn(req, rsp, verr)
-				}
-			}
-			return err
-		}
-	}
-}
-
-func (w *wrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
 		if v, ok := req.Body().(validator); ok {
 			if err := v.Validate(); err != nil {
 				return w.opts.ClientErrorFn(req, nil, err)
 			}
 		}
-	err := w.Client.Call(ctx, req, rsp, opts...)
+		err := next(ctx, req, rsp, opts...)
-	if v, ok := rsp.(validator); ok {
+		if v, ok := rsp.(validator); ok && w.opts.ClientValidateResponse {
 			if verr := v.Validate(); verr != nil {
 				return w.opts.ClientErrorFn(req, rsp, verr)
 			}
 		}
 		return err
+	}
 }
 
-func (w *wrapper) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
+func (w *hook) ClientStream(next client.FuncStream) client.FuncStream {
+	return func(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
 		if v, ok := req.Body().(validator); ok {
 			if err := v.Validate(); err != nil {
 				return nil, w.opts.ClientErrorFn(req, nil, err)
 			}
 		}
-	return w.Client.Stream(ctx, req, opts...)
+		return next(ctx, req, opts...)
+	}
 }
 
-func (w *wrapper) Publish(ctx context.Context, msg client.Message, opts ...client.PublishOption) error {
+func (w *hook) ClientPublish(next client.FuncPublish) client.FuncPublish {
+	return func(ctx context.Context, msg client.Message, opts ...client.PublishOption) error {
 		if v, ok := msg.Payload().(validator); ok {
 			if err := v.Validate(); err != nil {
 				return w.opts.PublishErrorFn(msg, err)
 			}
 		}
-	return w.Client.Publish(ctx, msg, opts...)
+		return next(ctx, msg, opts...)
+	}
 }
 
-func NewServerHandlerWrapper(opts ...Option) server.HandlerWrapper {
+func (w *hook) ClientBatchPublish(next client.FuncBatchPublish) client.FuncBatchPublish {
-	options := NewOptions(opts...)
+	return func(ctx context.Context, msgs []client.Message, opts ...client.PublishOption) error {
-	return func(fn server.HandlerFunc) server.HandlerFunc {
+		for _, msg := range msgs {
+			if v, ok := msg.Payload().(validator); ok {
+				if err := v.Validate(); err != nil {
+					return w.opts.PublishErrorFn(msg, err)
+				}
+			}
+		}
+		return next(ctx, msgs, opts...)
+	}
+}
+
+func (w *hook) ServerHandler(next server.FuncHandler) server.FuncHandler {
 	return func(ctx context.Context, req server.Request, rsp interface{}) error {
 		if v, ok := req.Body().(validator); ok {
 			if err := v.Validate(); err != nil {
-					return options.ClientErrorFn(req, nil, err)
+				return w.opts.ServerErrorFn(req, nil, err)
 			}
 		}
-			err := fn(ctx, req, rsp)
+		err := next(ctx, req, rsp)
-			if v, ok := rsp.(validator); ok {
+		if v, ok := rsp.(validator); ok && w.opts.ServerValidateResponse {
 			if verr := v.Validate(); verr != nil {
-					return options.ClientErrorFn(req, rsp, err)
+				return w.opts.ServerErrorFn(req, rsp, verr)
 			}
 		}
 		return err
 	}
-	}
 }
 
-func NewServerSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
+func (w *hook) ServerSubscriber(next server.FuncSubHandler) server.FuncSubHandler {
-	options := NewOptions(opts...)
-	return func(fn server.SubscriberFunc) server.SubscriberFunc {
 	return func(ctx context.Context, msg server.Message) error {
 		if v, ok := msg.Body().(validator); ok {
 			if err := v.Validate(); err != nil {
-					return options.SubscribeErrorFn(msg, err)
+				return w.opts.SubscribeErrorFn(msg, err)
 			}
 		}
-			return fn(ctx, msg)
+		return next(ctx, msg)
-		}
 	}
 }

validator_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"testing"
 
+	memory "go.unistack.org/micro/v3/broker/memory"
 	"go.unistack.org/micro/v3/client"
 	"go.unistack.org/micro/v3/codec"
 	"go.unistack.org/micro/v3/server"
@@ -20,10 +21,6 @@ type Message struct {
 
 func (m *Message) Validate() error {
 	return fmt.Errorf("SSS")
-	if len(m.Name) == 0 || m.Name != "test" {
-		return fmt.Errorf("name is empty")
-	}
-	return nil
 }
 
 func (h *Handler) Sub(ctx context.Context, req *Message) error {
@@ -34,12 +31,26 @@ func TestValidator(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	b := memory.NewBroker()
+
+	if err := b.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := b.Connect(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	w := NewHook()
 	// create server
 	srv := server.NewServer(
 		server.Name("helloworld"),
 		server.Codec("application/json", codec.NewCodec()),
-		server.WrapHandler(NewServerHandlerWrapper()),
+		server.Broker(b),
-		server.WrapSubscriber(NewServerSubscriberWrapper()),
+		server.Hooks(
+			server.HookHandler(w.ServerHandler),
+			server.HookSubHandler(w.ServerSubscriber),
+		),
 		server.Context(ctx),
 	)
 
@@ -61,17 +72,23 @@ func TestValidator(t *testing.T) {
 	cli := client.NewClient(
 		client.ContentType("application/json"),
 		client.Codec("application/json", codec.NewCodec()),
-		client.Wrap(NewClientWrapper()),
+		client.Broker(b),
+		client.Hooks(
+			client.HookCall(w.ClientCall),
+			client.HookStream(w.ClientStream),
+			client.HookPublish(w.ClientPublish),
+			client.HookBatchPublish(w.ClientBatchPublish),
+		),
 	)
 
+	if err := cli.Init(); err != nil {
+		t.Fatal(err)
+	}
+
 	if err := cli.Publish(ctx, cli.NewMessage("test", &Message{Name: "test1"}, client.WithMessageContentType("application/json"))); err == nil {
 		t.Fatalf("validator not works as message with bad contents")
 	}
 
-	if err := cli.Publish(ctx, cli.NewMessage("test", &Message{Name: "test"}, client.WithMessageContentType("application/json"))); err == nil {
-		t.Fatal(err)
-	}
-
 	// stop server
 	if err := srv.Stop(); err != nil {
 		t.Fatal(err)