util/reflect: add StructFieldNameByTag

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

broker/broker.go

@@ -48,6 +48,17 @@ type Broker interface {
 	String() string
 }
 
+type (
+	FuncPublish        func(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error
+	HookPublish        func(next FuncPublish) FuncPublish
+	FuncBatchPublish   func(ctx context.Context, msgs []*Message, opts ...PublishOption) error
+	HookBatchPublish   func(next FuncBatchPublish) FuncBatchPublish
+	FuncSubscribe      func(ctx context.Context, topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)
+	HookSubscribe      func(next FuncSubscribe) FuncSubscribe
+	FuncBatchSubscribe func(ctx context.Context, topic string, h BatchHandler, opts ...SubscribeOption) (Subscriber, error)
+	HookBatchSubscribe func(next FuncBatchSubscribe) FuncBatchSubscribe
+)
+
 // Handler is used to process messages via a subscription of a topic.
 type Handler func(Event) error
 

broker/memory/memory.go

@@ -7,6 +7,7 @@ import (
 	"go.unistack.org/micro/v3/broker"
 	"go.unistack.org/micro/v3/logger"
 	"go.unistack.org/micro/v3/metadata"
+	"go.unistack.org/micro/v3/options"
 	maddr "go.unistack.org/micro/v3/util/addr"
 	"go.unistack.org/micro/v3/util/id"
 	mnet "go.unistack.org/micro/v3/util/net"
@@ -14,9 +15,13 @@ import (
 )
 
 type memoryBroker struct {
-	subscribers map[string][]*memorySubscriber
-	addr        string
-	opts        broker.Options
+	funcPublish        broker.FuncPublish
+	funcBatchPublish   broker.FuncBatchPublish
+	funcSubscribe      broker.FuncSubscribe
+	funcBatchSubscribe broker.FuncBatchSubscribe
+	subscribers        map[string][]*memorySubscriber
+	addr               string
+	opts               broker.Options
 	sync.RWMutex
 	connected bool
 }
@@ -98,15 +103,42 @@ func (m *memoryBroker) Init(opts ...broker.Option) error {
 	for _, o := range opts {
 		o(&m.opts)
 	}
+
+	m.funcPublish = m.fnPublish
+	m.funcBatchPublish = m.fnBatchPublish
+	m.funcSubscribe = m.fnSubscribe
+	m.funcBatchSubscribe = m.fnBatchSubscribe
+
+	m.opts.Hooks.EachNext(func(hook options.Hook) {
+		switch h := hook.(type) {
+		case broker.HookPublish:
+			m.funcPublish = h(m.funcPublish)
+		case broker.HookBatchPublish:
+			m.funcBatchPublish = h(m.funcBatchPublish)
+		case broker.HookSubscribe:
+			m.funcSubscribe = h(m.funcSubscribe)
+		case broker.HookBatchSubscribe:
+			m.funcBatchSubscribe = h(m.funcBatchSubscribe)
+		}
+	})
+
 	return nil
 }
 
 func (m *memoryBroker) Publish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error {
+	return m.funcPublish(ctx, topic, msg, opts...)
+}
+
+func (m *memoryBroker) fnPublish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error {
 	msg.Header.Set(metadata.HeaderTopic, topic)
 	return m.publish(ctx, []*broker.Message{msg}, opts...)
 }
 
 func (m *memoryBroker) BatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
+	return m.funcBatchPublish(ctx, msgs, opts...)
+}
+
+func (m *memoryBroker) fnBatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
 	return m.publish(ctx, msgs, opts...)
 }
 
@@ -202,6 +234,10 @@ func (m *memoryBroker) publish(ctx context.Context, msgs []*broker.Message, opts
 }
 
 func (m *memoryBroker) BatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
+	return m.funcBatchSubscribe(ctx, topic, handler, opts...)
+}
+
+func (m *memoryBroker) fnBatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
 	m.RLock()
 	if !m.connected {
 		m.RUnlock()
@@ -247,6 +283,10 @@ func (m *memoryBroker) BatchSubscribe(ctx context.Context, topic string, handler
 }
 
 func (m *memoryBroker) Subscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
+	return m.funcSubscribe(ctx, topic, handler, opts...)
+}
+
+func (m *memoryBroker) fnSubscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
 	m.RLock()
 	if !m.connected {
 		m.RUnlock()

broker/memory/memory_test.go

@@ -13,6 +13,10 @@ func TestMemoryBatchBroker(t *testing.T) {
 	b := NewBroker()
 	ctx := context.Background()
 
+	if err := b.Init(); err != nil {
+		t.Fatalf("Unexpected init error %v", err)
+	}
+
 	if err := b.Connect(ctx); err != nil {
 		t.Fatalf("Unexpected connect error %v", err)
 	}
@@ -59,6 +63,10 @@ func TestMemoryBroker(t *testing.T) {
 	b := NewBroker()
 	ctx := context.Background()
 
+	if err := b.Init(); err != nil {
+		t.Fatalf("Unexpected init error %v", err)
+	}
+
 	if err := b.Connect(ctx); err != nil {
 		t.Fatalf("Unexpected connect error %v", err)
 	}

broker/noop.go

@@ -3,14 +3,25 @@ package broker
 import (
 	"context"
 	"strings"
+
+	"go.unistack.org/micro/v3/options"
 )
 
 type NoopBroker struct {
-	opts Options
+	funcPublish        FuncPublish
+	funcBatchPublish   FuncBatchPublish
+	funcSubscribe      FuncSubscribe
+	funcBatchSubscribe FuncBatchSubscribe
+	opts               Options
 }
 
 func NewBroker(opts ...Option) *NoopBroker {
 	b := &NoopBroker{opts: NewOptions(opts...)}
+	b.funcPublish = b.fnPublish
+	b.funcBatchPublish = b.fnBatchPublish
+	b.funcSubscribe = b.fnSubscribe
+	b.funcBatchSubscribe = b.fnBatchSubscribe
+
 	return b
 }
 
@@ -30,6 +41,25 @@ func (b *NoopBroker) Init(opts ...Option) error {
 	for _, opt := range opts {
 		opt(&b.opts)
 	}
+
+	b.funcPublish = b.fnPublish
+	b.funcBatchPublish = b.fnBatchPublish
+	b.funcSubscribe = b.fnSubscribe
+	b.funcBatchSubscribe = b.fnBatchSubscribe
+
+	b.opts.Hooks.EachNext(func(hook options.Hook) {
+		switch h := hook.(type) {
+		case HookPublish:
+			b.funcPublish = h(b.funcPublish)
+		case HookBatchPublish:
+			b.funcBatchPublish = h(b.funcBatchPublish)
+		case HookSubscribe:
+			b.funcSubscribe = h(b.funcSubscribe)
+		case HookBatchSubscribe:
+			b.funcBatchSubscribe = h(b.funcBatchSubscribe)
+		}
+	})
+
 	return nil
 }
 
@@ -45,14 +75,22 @@ func (b *NoopBroker) Address() string {
 	return strings.Join(b.opts.Addrs, ",")
 }
 
-func (b *NoopBroker) BatchPublish(_ context.Context, _ []*Message, _ ...PublishOption) error {
+func (b *NoopBroker) fnBatchPublish(_ context.Context, _ []*Message, _ ...PublishOption) error {
 	return nil
 }
 
-func (b *NoopBroker) Publish(_ context.Context, _ string, _ *Message, _ ...PublishOption) error {
+func (b *NoopBroker) BatchPublish(ctx context.Context, msgs []*Message, opts ...PublishOption) error {
+	return b.funcBatchPublish(ctx, msgs, opts...)
+}
+
+func (b *NoopBroker) fnPublish(_ context.Context, _ string, _ *Message, _ ...PublishOption) error {
 	return nil
 }
 
+func (b *NoopBroker) Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error {
+	return b.funcPublish(ctx, topic, msg, opts...)
+}
+
 type NoopSubscriber struct {
 	ctx          context.Context
 	topic        string
@@ -61,14 +99,22 @@ type NoopSubscriber struct {
 	opts         SubscribeOptions
 }
 
-func (b *NoopBroker) BatchSubscribe(ctx context.Context, topic string, handler BatchHandler, opts ...SubscribeOption) (Subscriber, error) {
+func (b *NoopBroker) fnBatchSubscribe(ctx context.Context, topic string, handler BatchHandler, opts ...SubscribeOption) (Subscriber, error) {
 	return &NoopSubscriber{ctx: ctx, topic: topic, opts: NewSubscribeOptions(opts...), batchHandler: handler}, nil
 }
 
-func (b *NoopBroker) Subscribe(ctx context.Context, topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+func (b *NoopBroker) BatchSubscribe(ctx context.Context, topic string, handler BatchHandler, opts ...SubscribeOption) (Subscriber, error) {
+	return b.funcBatchSubscribe(ctx, topic, handler, opts...)
+}
+
+func (b *NoopBroker) fnSubscribe(ctx context.Context, topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
 	return &NoopSubscriber{ctx: ctx, topic: topic, opts: NewSubscribeOptions(opts...), handler: handler}, nil
 }
 
+func (b *NoopBroker) Subscribe(ctx context.Context, topic string, handler Handler, opts ...SubscribeOption) (Subscriber, error) {
+	return b.funcSubscribe(ctx, topic, handler, opts...)
+}
+
 func (s *NoopSubscriber) Options() SubscribeOptions {
 	return s.opts
 }
@@ -77,6 +123,6 @@ func (s *NoopSubscriber) Topic() string {
 	return s.topic
 }
 
-func (s *NoopSubscriber) Unsubscribe(ctx context.Context) error {
+func (s *NoopSubscriber) Unsubscribe(_ context.Context) error {
 	return nil
 }

broker/noop_test.go

@@ -0,0 +1,35 @@
+package broker
+
+import (
+	"context"
+	"testing"
+)
+
+type testHook struct {
+	f bool
+}
+
+func (t *testHook) Publish1(fn FuncPublish) FuncPublish {
+	return func(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error {
+		t.f = true
+		return fn(ctx, topic, msg, opts...)
+	}
+}
+
+func TestNoopHook(t *testing.T) {
+	h := &testHook{}
+
+	b := NewBroker(Hooks(HookPublish(h.Publish1)))
+
+	if err := b.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := b.Publish(context.TODO(), "", nil); err != nil {
+		t.Fatal(err)
+	}
+
+	if !h.f {
+		t.Fatal("hook not works")
+	}
+}

broker/options.go

@@ -8,6 +8,7 @@ import (
 	"go.unistack.org/micro/v3/codec"
 	"go.unistack.org/micro/v3/logger"
 	"go.unistack.org/micro/v3/meter"
+	"go.unistack.org/micro/v3/options"
 	"go.unistack.org/micro/v3/register"
 	"go.unistack.org/micro/v3/sync"
 	"go.unistack.org/micro/v3/tracer"
@@ -37,10 +38,13 @@ type Options struct {
 	Name string
 	// Addrs holds the broker address
 	Addrs []string
-
+	// Wait waits for a collection of goroutines to finish
 	Wait *sync.WaitGroup
-
+	// GracefulTimeout contains time to wait to finish in flight requests
 	GracefulTimeout time.Duration
+	// Hooks can be run before broker Publish/BatchPublish and
+	// Subscribe/BatchSubscribe methods
+	Hooks options.Hooks
 }
 
 // NewOptions create new Options
@@ -230,6 +234,13 @@ func Name(n string) Option {
 	}
 }
 
+// Hooks sets hook runs before action
+func Hooks(h ...options.Hook) Option {
+	return func(o *Options) {
+		o.Hooks = append(o.Hooks, h...)
+	}
+}
+
 // SubscribeContext set context
 func SubscribeContext(ctx context.Context) SubscribeOption {
 	return func(o *SubscribeOptions) {

client/client.go

@@ -44,6 +44,17 @@ type Client interface {
 	String() string
 }
 
+type (
+	FuncCall         func(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error
+	HookCall         func(next FuncCall) FuncCall
+	FuncStream       func(ctx context.Context, req Request, opts ...CallOption) (Stream, error)
+	HookStream       func(next FuncStream) FuncStream
+	FuncPublish      func(ctx context.Context, msg Message, opts ...PublishOption) error
+	HookPublish      func(next FuncPublish) FuncPublish
+	FuncBatchPublish func(ctx context.Context, msg []Message, opts ...PublishOption) error
+	HookBatchPublish func(next FuncBatchPublish) FuncBatchPublish
+)
+
 // Message is the interface for publishing asynchronously
 type Message interface {
 	Topic() string

client/client_call_options_test.go

@@ -1,26 +0,0 @@
-package client
-
-import (
-	"context"
-	"testing"
-	"time"
-)
-
-func TestNewClientCallOptions(t *testing.T) {
-	var flag bool
-	w := func(fn CallFunc) CallFunc {
-		flag = true
-		return fn
-	}
-	c := NewClientCallOptions(NewClient(),
-		WithAddress("127.0.0.1"),
-		WithCallWrapper(w),
-		WithRequestTimeout(1*time.Millisecond),
-		WithRetries(0),
-		WithBackoff(BackoffInterval(10*time.Millisecond, 100*time.Millisecond)),
-	)
-	_ = c.Call(context.TODO(), c.NewRequest("service", "endpoint", nil), nil)
-	if !flag {
-		t.Fatalf("NewClientCallOptions not works")
-	}
-}

client/noop.go

@@ -4,13 +4,17 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strconv"
 	"time"
 
 	"go.unistack.org/micro/v3/broker"
 	"go.unistack.org/micro/v3/codec"
 	"go.unistack.org/micro/v3/errors"
 	"go.unistack.org/micro/v3/metadata"
+	"go.unistack.org/micro/v3/options"
 	"go.unistack.org/micro/v3/selector"
+	"go.unistack.org/micro/v3/semconv"
+	"go.unistack.org/micro/v3/tracer"
 )
 
 // DefaultCodecs will be used to encode/decode data
@@ -19,7 +23,11 @@ var DefaultCodecs = map[string]codec.Codec{
 }
 
 type noopClient struct {
-	opts Options
+	funcPublish      FuncPublish
+	funcBatchPublish FuncBatchPublish
+	funcCall         FuncCall
+	funcStream       FuncStream
+	opts             Options
 }
 
 type noopMessage struct {
@@ -40,16 +48,14 @@ type noopRequest struct {
 
 // NewClient returns new noop client
 func NewClient(opts ...Option) Client {
-	nc := &noopClient{opts: NewOptions(opts...)}
-	// wrap in reverse
+	n := &noopClient{opts: NewOptions(opts...)}
 
-	c := Client(nc)
+	n.funcCall = n.fnCall
+	n.funcStream = n.fnStream
+	n.funcPublish = n.fnPublish
+	n.funcBatchPublish = n.fnBatchPublish
 
-	for i := len(nc.opts.Wrappers); i > 0; i-- {
-		c = nc.opts.Wrappers[i-1](c)
-	}
-
-	return c
+	return n
 }
 
 func (n *noopClient) Name() string {
@@ -101,10 +107,13 @@ func (n *noopResponse) Read() ([]byte, error) {
 	return nil, nil
 }
 
-type noopStream struct{}
+type noopStream struct {
+	err error
+	ctx context.Context
+}
 
 func (n *noopStream) Context() context.Context {
-	return context.Background()
+	return n.ctx
 }
 
 func (n *noopStream) Request() Request {
@@ -132,15 +141,21 @@ func (n *noopStream) RecvMsg(interface{}) error {
 }
 
 func (n *noopStream) Error() error {
-	return nil
+	return n.err
 }
 
 func (n *noopStream) Close() error {
-	return nil
+	if sp, ok := tracer.SpanFromContext(n.ctx); ok && sp != nil {
+		if n.err != nil {
+			sp.SetStatus(tracer.SpanStatusError, n.err.Error())
+		}
+		sp.Finish()
+	}
+	return n.err
 }
 
 func (n *noopStream) CloseSend() error {
-	return nil
+	return n.err
 }
 
 func (n *noopMessage) Topic() string {
@@ -173,6 +188,25 @@ func (n *noopClient) Init(opts ...Option) error {
 	for _, o := range opts {
 		o(&n.opts)
 	}
+
+	n.funcCall = n.fnCall
+	n.funcStream = n.fnStream
+	n.funcPublish = n.fnPublish
+	n.funcBatchPublish = n.fnBatchPublish
+
+	n.opts.Hooks.EachNext(func(hook options.Hook) {
+		switch h := hook.(type) {
+		case HookCall:
+			n.funcCall = h(n.funcCall)
+		case HookStream:
+			n.funcStream = h(n.funcStream)
+		case HookPublish:
+			n.funcPublish = h(n.funcPublish)
+		case HookBatchPublish:
+			n.funcBatchPublish = h(n.funcBatchPublish)
+		}
+	})
+
 	return nil
 }
 
@@ -185,6 +219,31 @@ func (n *noopClient) String() string {
 }
 
 func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error {
+	ts := time.Now()
+	n.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
+	var sp tracer.Span
+	ctx, sp = n.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
+		tracer.WithSpanKind(tracer.SpanKindClient),
+		tracer.WithSpanLabels("endpoint", req.Endpoint()),
+	)
+	err := n.funcCall(ctx, req, rsp, opts...)
+	n.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Dec()
+	te := time.Since(ts)
+	n.opts.Meter.Summary(semconv.ClientRequestLatencyMicroseconds, "endpoint", req.Endpoint()).Update(te.Seconds())
+	n.opts.Meter.Histogram(semconv.ClientRequestDurationSeconds, "endpoint", req.Endpoint()).Update(te.Seconds())
+
+	if me := errors.FromError(err); me == nil {
+		sp.Finish()
+		n.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "success", "code", strconv.Itoa(int(200))).Inc()
+	} else {
+		sp.SetStatus(tracer.SpanStatusError, err.Error())
+		n.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "failure", "code", strconv.Itoa(int(me.Code))).Inc()
+	}
+
+	return err
+}
+
+func (n *noopClient) fnCall(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error {
 	// make a copy of call opts
 	callOpts := n.opts.CallOptions
 	for _, opt := range opts {
@@ -213,11 +272,8 @@ func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opt
 	}
 
 	// make copy of call method
-	hcall := n.call
-
-	// wrap the call in reverse
-	for i := len(callOpts.CallWrappers); i > 0; i-- {
-		hcall = callOpts.CallWrappers[i-1](hcall)
+	hcall := func(ctx context.Context, addr string, req Request, rsp interface{}, opts CallOptions) error {
+		return nil
 	}
 
 	// use the router passed as a call option, or fallback to the rpc clients router
@@ -316,10 +372,6 @@ func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opt
 	return gerr
 }
 
-func (n *noopClient) call(ctx context.Context, addr string, req Request, rsp interface{}, opts CallOptions) error {
-	return nil
-}
-
 func (n *noopClient) NewRequest(service, endpoint string, req interface{}, opts ...RequestOption) Request {
 	return &noopRequest{service: service, endpoint: endpoint}
 }
@@ -330,6 +382,31 @@ func (n *noopClient) NewMessage(topic string, msg interface{}, opts ...MessageOp
 }
 
 func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption) (Stream, error) {
+	ts := time.Now()
+	n.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
+	var sp tracer.Span
+	ctx, sp = n.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
+		tracer.WithSpanKind(tracer.SpanKindClient),
+		tracer.WithSpanLabels("endpoint", req.Endpoint()),
+	)
+	stream, err := n.funcStream(ctx, req, opts...)
+	n.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Dec()
+	te := time.Since(ts)
+	n.opts.Meter.Summary(semconv.ClientRequestLatencyMicroseconds, "endpoint", req.Endpoint()).Update(te.Seconds())
+	n.opts.Meter.Histogram(semconv.ClientRequestDurationSeconds, "endpoint", req.Endpoint()).Update(te.Seconds())
+
+	if me := errors.FromError(err); me == nil {
+		sp.Finish()
+		n.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "success", "code", strconv.Itoa(int(200))).Inc()
+	} else {
+		sp.SetStatus(tracer.SpanStatusError, err.Error())
+		n.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "failure", "code", strconv.Itoa(int(me.Code))).Inc()
+	}
+
+	return stream, err
+}
+
+func (n *noopClient) fnStream(ctx context.Context, req Request, opts ...CallOption) (Stream, error) {
 	var err error
 
 	// make a copy of call opts
@@ -470,14 +547,22 @@ func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption
 }
 
 func (n *noopClient) stream(ctx context.Context, addr string, req Request, opts CallOptions) (Stream, error) {
-	return &noopStream{}, nil
+	return &noopStream{ctx: ctx}, nil
 }
 
 func (n *noopClient) BatchPublish(ctx context.Context, ps []Message, opts ...PublishOption) error {
+	return n.funcBatchPublish(ctx, ps, opts...)
+}
+
+func (n *noopClient) fnBatchPublish(ctx context.Context, ps []Message, opts ...PublishOption) error {
 	return n.publish(ctx, ps, opts...)
 }
 
 func (n *noopClient) Publish(ctx context.Context, p Message, opts ...PublishOption) error {
+	return n.funcPublish(ctx, p, opts...)
+}
+
+func (n *noopClient) fnPublish(ctx context.Context, p Message, opts ...PublishOption) error {
 	return n.publish(ctx, []Message{p}, opts...)
 }
 
@@ -538,6 +623,13 @@ func (n *noopClient) publish(ctx context.Context, ps []Message, opts ...PublishO
 		msgs = append(msgs, &broker.Message{Header: md, Body: body})
 	}
 
+	if len(msgs) == 1 {
+		return n.opts.Broker.Publish(ctx, msgs[0].Header[metadata.HeaderTopic], msgs[0],
+			broker.PublishContext(options.Context),
+			broker.PublishBodyOnly(options.BodyOnly),
+		)
+	}
+
 	return n.opts.Broker.BatchPublish(ctx, msgs,
 		broker.PublishContext(options.Context),
 		broker.PublishBodyOnly(options.BodyOnly),

client/noop_test.go

@@ -0,0 +1,35 @@
+package client
+
+import (
+	"context"
+	"testing"
+)
+
+type testHook struct {
+	f bool
+}
+
+func (t *testHook) Publish(fn FuncPublish) FuncPublish {
+	return func(ctx context.Context, msg Message, opts ...PublishOption) error {
+		t.f = true
+		return fn(ctx, msg, opts...)
+	}
+}
+
+func TestNoopHook(t *testing.T) {
+	h := &testHook{}
+
+	c := NewClient(Hooks(HookPublish(h.Publish)))
+
+	if err := c.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := c.Publish(context.TODO(), c.NewMessage("", nil, MessageContentType("application/octet-stream"))); err != nil {
+		t.Fatal(err)
+	}
+
+	if !h.f {
+		t.Fatal("hook not works")
+	}
+}

client/options.go

@@ -12,6 +12,7 @@ import (
 	"go.unistack.org/micro/v3/metadata"
 	"go.unistack.org/micro/v3/meter"
 	"go.unistack.org/micro/v3/network/transport"
+	"go.unistack.org/micro/v3/options"
 	"go.unistack.org/micro/v3/register"
 	"go.unistack.org/micro/v3/router"
 	"go.unistack.org/micro/v3/selector"
@@ -59,6 +60,9 @@ type Options struct {
 	PoolTTL time.Duration
 	// ContextDialer used to connect
 	ContextDialer func(context.Context, string) (net.Conn, error)
+	// Hooks can be run before broker Publish/BatchPublish and
+	// Subscribe/BatchSubscribe methods
+	Hooks options.Hooks
 }
 
 // NewCallOptions creates new call options struct
@@ -92,8 +96,6 @@ type CallOptions struct {
 	Address []string
 	// SelectOptions selector options
 	SelectOptions []selector.SelectOption
-	// CallWrappers call wrappers
-	CallWrappers []CallWrapper
 	// StreamTimeout stream timeout
 	StreamTimeout time.Duration
 	// RequestTimeout request timeout
@@ -185,7 +187,7 @@ func NewOptions(opts ...Option) Options {
 	options := Options{
 		Context:     context.Background(),
 		ContentType: DefaultContentType,
-		Codecs:      make(map[string]codec.Codec),
+		Codecs:      DefaultCodecs,
 		CallOptions: CallOptions{
 			Context:        context.Background(),
 			Backoff:        DefaultBackoff,
@@ -306,20 +308,6 @@ func Selector(s selector.Selector) Option {
 	}
 }
 
-// Wrap adds a wrapper to the list of options passed into the client
-func Wrap(w Wrapper) Option {
-	return func(o *Options) {
-		o.Wrappers = append(o.Wrappers, w)
-	}
-}
-
-// WrapCall adds a wrapper to the list of CallFunc wrappers
-func WrapCall(cw ...CallWrapper) Option {
-	return func(o *Options) {
-		o.CallOptions.CallWrappers = append(o.CallOptions.CallWrappers, cw...)
-	}
-}
-
 // Backoff is used to set the backoff function used when retrying Calls
 func Backoff(fn BackoffFunc) Option {
 	return func(o *Options) {
@@ -450,13 +438,6 @@ func WithAddress(a ...string) CallOption {
 	}
 }
 
-// WithCallWrapper is a CallOption which adds to the existing CallFunc wrappers
-func WithCallWrapper(cw ...CallWrapper) CallOption {
-	return func(o *CallOptions) {
-		o.CallWrappers = append(o.CallWrappers, cw...)
-	}
-}
-
 // WithBackoff is a CallOption which overrides that which
 // set in Options.CallOptions
 func WithBackoff(fn BackoffFunc) CallOption {
@@ -591,3 +572,10 @@ func RequestContentType(ct string) RequestOption {
 		o.ContentType = ct
 	}
 }
+
+// Hooks sets hook runs before action
+func Hooks(h ...options.Hook) Option {
+	return func(o *Options) {
+		o.Hooks = append(o.Hooks, h...)
+	}
+}

config/config.go

@@ -50,6 +50,13 @@ type Config interface {
 	String() string
 }
 
+type (
+	FuncLoad func(ctx context.Context, opts ...LoadOption) error
+	HookLoad func(next FuncLoad) FuncLoad
+	FuncSave func(ctx context.Context, opts ...SaveOption) error
+	HookSave func(next FuncSave) FuncSave
+)
+
 // Watcher is the config watcher
 type Watcher interface {
 	// Next blocks until update happens or error returned

config/default.go

@@ -9,13 +9,16 @@ import (
 
 	"dario.cat/mergo"
 	"github.com/google/uuid"
+	"go.unistack.org/micro/v3/options"
 	mid "go.unistack.org/micro/v3/util/id"
 	rutil "go.unistack.org/micro/v3/util/reflect"
 	mtime "go.unistack.org/micro/v3/util/time"
 )
 
 type defaultConfig struct {
-	opts Options
+	funcLoad FuncLoad
+	funcSave FuncSave
+	opts     Options
 }
 
 func (c *defaultConfig) Options() Options {
@@ -31,6 +34,18 @@ func (c *defaultConfig) Init(opts ...Option) error {
 		return err
 	}
 
+	c.funcLoad = c.fnLoad
+	c.funcSave = c.fnSave
+
+	c.opts.Hooks.EachNext(func(hook options.Hook) {
+		switch h := hook.(type) {
+		case HookLoad:
+			c.funcLoad = h(c.funcLoad)
+		case HookSave:
+			c.funcSave = h(c.funcSave)
+		}
+	})
+
 	if err := DefaultAfterInit(c.opts.Context, c); err != nil && !c.opts.AllowFail {
 		return err
 	}
@@ -39,11 +54,17 @@ func (c *defaultConfig) Init(opts ...Option) error {
 }
 
 func (c *defaultConfig) Load(ctx context.Context, opts ...LoadOption) error {
+	return c.funcLoad(ctx, opts...)
+}
+
+func (c *defaultConfig) fnLoad(ctx context.Context, opts ...LoadOption) error {
+	var err error
+
 	if c.opts.SkipLoad != nil && c.opts.SkipLoad(ctx, c) {
 		return nil
 	}
 
-	if err := DefaultBeforeLoad(ctx, c); err != nil && !c.opts.AllowFail {
+	if err = DefaultBeforeLoad(ctx, c); err != nil && !c.opts.AllowFail {
 		return err
 	}
 
@@ -233,6 +254,7 @@ func fillValue(value reflect.Value, val string) error {
 		}
 		value.Set(reflect.ValueOf(v))
 	}
+
 	return nil
 }
 
@@ -295,7 +317,11 @@ func fillValues(valueOf reflect.Value, tname string) error {
 	return nil
 }
 
-func (c *defaultConfig) Save(ctx context.Context, _ ...SaveOption) error {
+func (c *defaultConfig) Save(ctx context.Context, opts ...SaveOption) error {
+	return c.funcSave(ctx, opts...)
+}
+
+func (c *defaultConfig) fnSave(ctx context.Context, opts ...SaveOption) error {
 	if c.opts.SkipSave != nil && c.opts.SkipSave(ctx, c) {
 		return nil
 	}
@@ -319,7 +345,7 @@ func (c *defaultConfig) Name() string {
 	return c.opts.Name
 }
 
-func (c *defaultConfig) Watch(ctx context.Context, opts ...WatchOption) (Watcher, error) {
+func (c *defaultConfig) Watch(_ context.Context, _ ...WatchOption) (Watcher, error) {
 	return nil, ErrWatcherNotImplemented
 }
 
@@ -329,5 +355,9 @@ func NewConfig(opts ...Option) Config {
 	if len(options.StructTag) == 0 {
 		options.StructTag = "default"
 	}
-	return &defaultConfig{opts: options}
+	c := &defaultConfig{opts: options}
+	c.funcLoad = c.fnLoad
+	c.funcSave = c.fnSave
+
+	return c
 }

config/default_test.go

@@ -41,6 +41,35 @@ func (c *cfgStructValue) Validate() error {
 	return nil
 }
 
+type testHook struct {
+	f bool
+}
+
+func (t *testHook) Load(fn config.FuncLoad) config.FuncLoad {
+	return func(ctx context.Context, opts ...config.LoadOption) error {
+		t.f = true
+		return fn(ctx, opts...)
+	}
+}
+
+func TestHook(t *testing.T) {
+	h := &testHook{}
+
+	c := config.NewConfig(config.Struct(h), config.Hooks(config.HookLoad(h.Load)))
+
+	if err := c.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := c.Load(context.TODO()); err != nil {
+		t.Fatal(err)
+	}
+
+	if !h.f {
+		t.Fatal("hook not works")
+	}
+}
+
 func TestDefault(t *testing.T) {
 	ctx := context.Background()
 	conf := &cfg{IntValue: 10}

config/options.go

@@ -7,6 +7,7 @@ import (
 	"go.unistack.org/micro/v3/codec"
 	"go.unistack.org/micro/v3/logger"
 	"go.unistack.org/micro/v3/meter"
+	"go.unistack.org/micro/v3/options"
 	"go.unistack.org/micro/v3/tracer"
 )
 
@@ -46,6 +47,8 @@ type Options struct {
 	SkipLoad func(context.Context, Config) bool
 	// SkipSave runs only if condition returns true
 	SkipSave func(context.Context, Config) bool
+	// Hooks can be run before/after config Save/Load
+	Hooks options.Hooks
 }
 
 // Option function signature
@@ -288,3 +291,10 @@ func WatchStruct(src interface{}) WatchOption {
 		o.Struct = src
 	}
 }
+
+// Hooks sets hook runs before action
+func Hooks(h ...options.Hook) Option {
+	return func(o *Options) {
+		o.Hooks = append(o.Hooks, h...)
+	}
+}

errors/errors.go

@@ -262,6 +262,10 @@ func CodeIn(err interface{}, codes ...int32) bool {
 
 // FromError try to convert go error to *Error
 func FromError(err error) *Error {
+	if err == nil {
+		return nil
+	}
+
 	if verr, ok := err.(*Error); ok && verr != nil {
 		return verr
 	}

logger/options.go

@@ -182,3 +182,10 @@ func WithMicroKeys() Option {
 		o.ErrorKey = "error"
 	}
 }
+
+// WithAddCallerSkipCount add skip count for copy logger
+func WithAddCallerSkipCount(n int) Option {
+	return func(o *Options) {
+		o.CallerSkipCount += n
+	}
+}

options/hooks.go

@@ -1,4 +1,4 @@
-package options // import "go.unistack.org/micro/v3/options"
+package options
 
 // Hook func interface
 type Hook interface{}

semconv/cache.go

@@ -0,0 +1,12 @@
+package semconv
+
+var (
+	// CacheRequestDurationSeconds specifies meter metric name
+	CacheRequestDurationSeconds = "cache_request_duration_seconds"
+	// ClientRequestLatencyMicroseconds specifies meter metric name
+	CacheRequestLatencyMicroseconds = "cache_request_latency_microseconds"
+	// CacheRequestTotal specifies meter metric name
+	CacheRequestTotal = "cache_request_total"
+	// CacheRequestInflight specifies meter metric name
+	CacheRequestInflight = "cache_request_inflight"
+)

server/handler.go

@@ -1,59 +0,0 @@
-package server
-
-import (
-	"reflect"
-
-	"go.unistack.org/micro/v3/register"
-)
-
-type rpcHandler struct {
-	opts      HandlerOptions
-	handler   interface{}
-	name      string
-	endpoints []*register.Endpoint
-}
-
-func newRPCHandler(handler interface{}, opts ...HandlerOption) Handler {
-	options := NewHandlerOptions(opts...)
-
-	typ := reflect.TypeOf(handler)
-	hdlr := reflect.ValueOf(handler)
-	name := reflect.Indirect(hdlr).Type().Name()
-
-	var endpoints []*register.Endpoint
-
-	for m := 0; m < typ.NumMethod(); m++ {
-		if e := register.ExtractEndpoint(typ.Method(m)); e != nil {
-			e.Name = name + "." + e.Name
-
-			for k, v := range options.Metadata[e.Name] {
-				e.Metadata[k] = v
-			}
-
-			endpoints = append(endpoints, e)
-		}
-	}
-
-	return &rpcHandler{
-		name:      name,
-		handler:   handler,
-		endpoints: endpoints,
-		opts:      options,
-	}
-}
-
-func (r *rpcHandler) Name() string {
-	return r.name
-}
-
-func (r *rpcHandler) Handler() interface{} {
-	return r.handler
-}
-
-func (r *rpcHandler) Endpoints() []*register.Endpoint {
-	return r.endpoints
-}
-
-func (r *rpcHandler) Options() HandlerOptions {
-	return r.opts
-}

server/noop.go

@@ -1,14 +1,22 @@
 package server
 
 import (
+	"bytes"
+	"context"
 	"fmt"
+	"reflect"
+	"runtime/debug"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
 	"go.unistack.org/micro/v3/broker"
 	"go.unistack.org/micro/v3/codec"
+	"go.unistack.org/micro/v3/errors"
 	"go.unistack.org/micro/v3/logger"
+	"go.unistack.org/micro/v3/metadata"
+	"go.unistack.org/micro/v3/options"
 	"go.unistack.org/micro/v3/register"
 	maddr "go.unistack.org/micro/v3/util/addr"
 	mnet "go.unistack.org/micro/v3/util/net"
@@ -24,6 +32,58 @@ const (
 	defaultContentType = "application/json"
 )
 
+type rpcHandler struct {
+	opts      HandlerOptions
+	handler   interface{}
+	name      string
+	endpoints []*register.Endpoint
+}
+
+func newRPCHandler(handler interface{}, opts ...HandlerOption) Handler {
+	options := NewHandlerOptions(opts...)
+
+	typ := reflect.TypeOf(handler)
+	hdlr := reflect.ValueOf(handler)
+	name := reflect.Indirect(hdlr).Type().Name()
+
+	var endpoints []*register.Endpoint
+
+	for m := 0; m < typ.NumMethod(); m++ {
+		if e := register.ExtractEndpoint(typ.Method(m)); e != nil {
+			e.Name = name + "." + e.Name
+
+			for k, v := range options.Metadata[e.Name] {
+				e.Metadata[k] = v
+			}
+
+			endpoints = append(endpoints, e)
+		}
+	}
+
+	return &rpcHandler{
+		name:      name,
+		handler:   handler,
+		endpoints: endpoints,
+		opts:      options,
+	}
+}
+
+func (r *rpcHandler) Name() string {
+	return r.name
+}
+
+func (r *rpcHandler) Handler() interface{} {
+	return r.handler
+}
+
+func (r *rpcHandler) Endpoints() []*register.Endpoint {
+	return r.endpoints
+}
+
+func (r *rpcHandler) Options() HandlerOptions {
+	return r.opts
+}
+
 type noopServer struct {
 	h           Handler
 	wg          *sync.WaitGroup
@@ -94,6 +154,35 @@ func (n *noopServer) Subscribe(sb Subscriber) error {
 	return nil
 }
 
+type rpcMessage struct {
+	payload     interface{}
+	codec       codec.Codec
+	header      metadata.Metadata
+	topic       string
+	contentType string
+	body        []byte
+}
+
+func (r *rpcMessage) ContentType() string {
+	return r.contentType
+}
+
+func (r *rpcMessage) Topic() string {
+	return r.topic
+}
+
+func (r *rpcMessage) Body() interface{} {
+	return r.payload
+}
+
+func (r *rpcMessage) Header() metadata.Metadata {
+	return r.header
+}
+
+func (r *rpcMessage) Codec() codec.Codec {
+	return r.codec
+}
+
 func (n *noopServer) NewHandler(h interface{}, opts ...HandlerOption) Handler {
 	return newRPCHandler(h, opts...)
 }
@@ -185,7 +274,7 @@ func (n *noopServer) Register() error {
 
 	if !registered {
 		if config.Logger.V(logger.InfoLevel) {
-			config.Logger.Infof(n.opts.Context, "register [%s] Registering node: %s", config.Register.String(), service.Nodes[0].ID)
+			config.Logger.Info(n.opts.Context, fmt.Sprintf("register [%s] Registering node: %s", config.Register.String(), service.Nodes[0].ID))
 		}
 	}
 
@@ -223,7 +312,7 @@ func (n *noopServer) Deregister() error {
 	}
 
 	if config.Logger.V(logger.InfoLevel) {
-		config.Logger.Infof(n.opts.Context, "deregistering node: %s", service.Nodes[0].ID)
+		config.Logger.Info(n.opts.Context, fmt.Sprintf("deregistering node: %s", service.Nodes[0].ID))
 	}
 
 	if err := DefaultDeregisterFunc(service, config); err != nil {
@@ -254,11 +343,11 @@ func (n *noopServer) Deregister() error {
 			go func(s broker.Subscriber) {
 				defer wg.Done()
 				if config.Logger.V(logger.InfoLevel) {
-					config.Logger.Infof(n.opts.Context, "unsubscribing from topic: %s", s.Topic())
+					config.Logger.Info(n.opts.Context, "unsubscribing from topic: "+s.Topic())
 				}
 				if err := s.Unsubscribe(ncx); err != nil {
 					if config.Logger.V(logger.ErrorLevel) {
-						config.Logger.Errorf(n.opts.Context, "unsubscribing from topic: %s err: %v", s.Topic(), err)
+						config.Logger.Error(n.opts.Context, "unsubscribing from topic: "+s.Topic(), err)
 					}
 				}
 			}(subs[idx])
@@ -294,7 +383,7 @@ func (n *noopServer) Start() error {
 	config.Address = addr
 
 	if config.Logger.V(logger.InfoLevel) {
-		config.Logger.Infof(n.opts.Context, "server [noop] Listening on %s", config.Address)
+		config.Logger.Info(n.opts.Context, "server [noop] Listening on "+config.Address)
 	}
 
 	n.Lock()
@@ -308,13 +397,13 @@ func (n *noopServer) Start() error {
 		// connect to the broker
 		if err := config.Broker.Connect(config.Context); err != nil {
 			if config.Logger.V(logger.ErrorLevel) {
-				config.Logger.Errorf(n.opts.Context, "broker [%s] connect error: %v", config.Broker.String(), err)
+				config.Logger.Error(n.opts.Context, fmt.Sprintf("broker [%s] connect error", config.Broker.String()), err)
 			}
 			return err
 		}
 
 		if config.Logger.V(logger.InfoLevel) {
-			config.Logger.Infof(n.opts.Context, "broker [%s] Connected to %s", config.Broker.String(), config.Broker.Address())
+			config.Logger.Info(n.opts.Context, fmt.Sprintf("broker [%s] Connected to %s", config.Broker.String(), config.Broker.Address()))
 		}
 	}
 
@@ -322,13 +411,13 @@ func (n *noopServer) Start() error {
 	// nolint: nestif
 	if err := config.RegisterCheck(config.Context); err != nil {
 		if config.Logger.V(logger.ErrorLevel) {
-			config.Logger.Errorf(n.opts.Context, "server %s-%s register check error: %s", config.Name, config.ID, err)
+			config.Logger.Error(n.opts.Context, fmt.Sprintf("server %s-%s register check error", config.Name, config.ID), err)
 		}
 	} else {
 		// announce self to the world
 		if err := n.Register(); err != nil {
 			if config.Logger.V(logger.ErrorLevel) {
-				config.Logger.Errorf(n.opts.Context, "server register error: %v", err)
+				config.Logger.Error(n.opts.Context, "server register error", err)
 			}
 		}
 	}
@@ -361,23 +450,23 @@ func (n *noopServer) Start() error {
 				// nolint: nestif
 				if rerr != nil && registered {
 					if config.Logger.V(logger.ErrorLevel) {
-						config.Logger.Errorf(n.opts.Context, "server %s-%s register check error: %s, deregister it", config.Name, config.ID, rerr)
+						config.Logger.Error(n.opts.Context, fmt.Sprintf("server %s-%s register check error, deregister it", config.Name, config.ID), rerr)
 					}
 					// deregister self in case of error
 					if err := n.Deregister(); err != nil {
 						if config.Logger.V(logger.ErrorLevel) {
-							config.Logger.Errorf(n.opts.Context, "server %s-%s deregister error: %s", config.Name, config.ID, err)
+							config.Logger.Error(n.opts.Context, fmt.Sprintf("server %s-%s deregister error", config.Name, config.ID), err)
 						}
 					}
 				} else if rerr != nil && !registered {
 					if config.Logger.V(logger.ErrorLevel) {
-						config.Logger.Errorf(n.opts.Context, "server %s-%s register check error: %s", config.Name, config.ID, rerr)
+						config.Logger.Errorf(n.opts.Context, fmt.Sprintf("server %s-%s register check error", config.Name, config.ID), rerr)
 					}
 					continue
 				}
 				if err := n.Register(); err != nil {
 					if config.Logger.V(logger.ErrorLevel) {
-						config.Logger.Errorf(n.opts.Context, "server %s-%s register error: %s", config.Name, config.ID, err)
+						config.Logger.Error(n.opts.Context, fmt.Sprintf("server %s-%s register error", config.Name, config.ID), err)
 					}
 				}
 			// wait for exit
@@ -389,7 +478,7 @@ func (n *noopServer) Start() error {
 		// deregister self
 		if err := n.Deregister(); err != nil {
 			if config.Logger.V(logger.ErrorLevel) {
-				config.Logger.Errorf(n.opts.Context, "server deregister error: ", err)
+				config.Logger.Error(n.opts.Context, "server deregister error", err)
 			}
 		}
 
@@ -402,12 +491,12 @@ func (n *noopServer) Start() error {
 		ch <- nil
 
 		if config.Logger.V(logger.InfoLevel) {
-			config.Logger.Infof(n.opts.Context, "broker [%s] Disconnected from %s", config.Broker.String(), config.Broker.Address())
+			config.Logger.Info(n.opts.Context, fmt.Sprintf("broker [%s] Disconnected from %s", config.Broker.String(), config.Broker.Address()))
 		}
 		// disconnect broker
 		if err := config.Broker.Disconnect(config.Context); err != nil {
 			if config.Logger.V(logger.ErrorLevel) {
-				config.Logger.Errorf(n.opts.Context, "broker [%s] disconnect error: %v", config.Broker.String(), err)
+				config.Logger.Error(n.opts.Context, fmt.Sprintf("broker [%s] disconnect error", config.Broker.String()), err)
 			}
 		}
 	}()
@@ -423,36 +512,33 @@ func (n *noopServer) Start() error {
 func (n *noopServer) subscribe() error {
 	config := n.Options()
 
-	cx := config.Context
-	var err error
-	var sub broker.Subscriber
+	subCtx := config.Context
 
 	for sb := range n.subscribers {
-		if sb.Options().Context != nil {
-			cx = sb.Options().Context
+
+		if cx := sb.Options().Context; cx != nil {
+			subCtx = cx
+		}
+
+		opts := []broker.SubscribeOption{
+			broker.SubscribeContext(subCtx),
+			broker.SubscribeAutoAck(sb.Options().AutoAck),
+			broker.SubscribeBodyOnly(sb.Options().BodyOnly),
 		}
 
-		opts := []broker.SubscribeOption{broker.SubscribeContext(cx), broker.SubscribeAutoAck(sb.Options().AutoAck)}
 		if queue := sb.Options().Queue; len(queue) > 0 {
 			opts = append(opts, broker.SubscribeGroup(queue))
 		}
 
-		if sb.Options().Batch {
-			// batch processing handler
-			sub, err = config.Broker.BatchSubscribe(cx, sb.Topic(), n.createBatchSubHandler(sb, config), opts...)
-		} else {
-			// single processing handler
-			sub, err = config.Broker.Subscribe(cx, sb.Topic(), n.createSubHandler(sb, config), opts...)
+		if config.Logger.V(logger.InfoLevel) {
+			config.Logger.Info(n.opts.Context, "subscribing to topic: "+sb.Topic())
 		}
 
+		sub, err := config.Broker.Subscribe(subCtx, sb.Topic(), n.createSubHandler(sb, config), opts...)
 		if err != nil {
 			return err
 		}
 
-		if config.Logger.V(logger.InfoLevel) {
-			config.Logger.Infof(n.opts.Context, "subscribing to topic: %s", sb.Topic())
-		}
-
 		n.subscribers[sb] = []broker.Subscriber{sub}
 	}
 
@@ -478,3 +564,218 @@ func (n *noopServer) Stop() error {
 
 	return err
 }
+
+func newSubscriber(topic string, sub interface{}, opts ...SubscriberOption) Subscriber {
+	var endpoints []*register.Endpoint
+	var handlers []*handler
+
+	options := NewSubscriberOptions(opts...)
+
+	if typ := reflect.TypeOf(sub); typ.Kind() == reflect.Func {
+		h := &handler{
+			method: reflect.ValueOf(sub),
+		}
+
+		switch typ.NumIn() {
+		case 1:
+			h.reqType = typ.In(0)
+		case 2:
+			h.ctxType = typ.In(0)
+			h.reqType = typ.In(1)
+		}
+
+		handlers = append(handlers, h)
+		ep := &register.Endpoint{
+			Name:     "Func",
+			Request:  register.ExtractSubValue(typ),
+			Metadata: metadata.New(2),
+		}
+		ep.Metadata.Set("topic", topic)
+		ep.Metadata.Set("subscriber", "true")
+		endpoints = append(endpoints, ep)
+	} else {
+		hdlr := reflect.ValueOf(sub)
+		name := reflect.Indirect(hdlr).Type().Name()
+
+		for m := 0; m < typ.NumMethod(); m++ {
+			method := typ.Method(m)
+			h := &handler{
+				method: method.Func,
+			}
+
+			switch method.Type.NumIn() {
+			case 2:
+				h.reqType = method.Type.In(1)
+			case 3:
+				h.ctxType = method.Type.In(1)
+				h.reqType = method.Type.In(2)
+			}
+
+			handlers = append(handlers, h)
+			ep := &register.Endpoint{
+				Name:     name + "." + method.Name,
+				Request:  register.ExtractSubValue(method.Type),
+				Metadata: metadata.New(2),
+			}
+			ep.Metadata.Set("topic", topic)
+			ep.Metadata.Set("subscriber", "true")
+			endpoints = append(endpoints, ep)
+		}
+	}
+
+	return &subscriber{
+		rcvr:       reflect.ValueOf(sub),
+		typ:        reflect.TypeOf(sub),
+		topic:      topic,
+		subscriber: sub,
+		handlers:   handlers,
+		endpoints:  endpoints,
+		opts:       options,
+	}
+}
+
+//nolint:gocyclo
+func (n *noopServer) createSubHandler(sb *subscriber, opts Options) broker.Handler {
+	return func(p broker.Event) (err error) {
+		defer func() {
+			if r := recover(); r != nil {
+				n.RLock()
+				config := n.opts
+				n.RUnlock()
+				if config.Logger.V(logger.ErrorLevel) {
+					config.Logger.Error(n.opts.Context, "panic recovered: ", r)
+					config.Logger.Error(n.opts.Context, string(debug.Stack()))
+				}
+				err = errors.InternalServerError(n.opts.Name+".subscriber", "panic recovered: %v", r)
+			}
+		}()
+
+		msg := p.Message()
+		// if we don't have headers, create empty map
+		if msg.Header == nil {
+			msg.Header = metadata.New(2)
+		}
+
+		ct := msg.Header["Content-Type"]
+		if len(ct) == 0 {
+			msg.Header.Set(metadata.HeaderContentType, defaultContentType)
+			ct = defaultContentType
+		}
+		cf, err := n.newCodec(ct)
+		if err != nil {
+			return err
+		}
+
+		hdr := metadata.New(len(msg.Header))
+		for k, v := range msg.Header {
+			hdr.Set(k, v)
+		}
+
+		ctx := metadata.NewIncomingContext(sb.opts.Context, hdr)
+
+		results := make(chan error, len(sb.handlers))
+
+		for i := 0; i < len(sb.handlers); i++ {
+			handler := sb.handlers[i]
+
+			var isVal bool
+			var req reflect.Value
+
+			if handler.reqType.Kind() == reflect.Ptr {
+				req = reflect.New(handler.reqType.Elem())
+			} else {
+				req = reflect.New(handler.reqType)
+				isVal = true
+			}
+			if isVal {
+				req = req.Elem()
+			}
+
+			if err = cf.ReadBody(bytes.NewBuffer(msg.Body), req.Interface()); err != nil {
+				return err
+			}
+
+			fn := func(ctx context.Context, msg Message) error {
+				var vals []reflect.Value
+				if sb.typ.Kind() != reflect.Func {
+					vals = append(vals, sb.rcvr)
+				}
+				if handler.ctxType != nil {
+					vals = append(vals, reflect.ValueOf(ctx))
+				}
+
+				vals = append(vals, reflect.ValueOf(msg.Body()))
+
+				returnValues := handler.method.Call(vals)
+				if rerr := returnValues[0].Interface(); rerr != nil {
+					return rerr.(error)
+				}
+				return nil
+			}
+
+			opts.Hooks.EachNext(func(hook options.Hook) {
+				if h, ok := hook.(HookSubHandler); ok {
+					fn = h(fn)
+				}
+			})
+
+			if n.wg != nil {
+				n.wg.Add(1)
+			}
+			go func() {
+				if n.wg != nil {
+					defer n.wg.Done()
+				}
+				cerr := fn(ctx, &rpcMessage{
+					topic:       sb.topic,
+					contentType: ct,
+					payload:     req.Interface(),
+					header:      msg.Header,
+				})
+				results <- cerr
+			}()
+		}
+		var errors []string
+		for i := 0; i < len(sb.handlers); i++ {
+			if rerr := <-results; rerr != nil {
+				errors = append(errors, rerr.Error())
+			}
+		}
+		if len(errors) > 0 {
+			err = fmt.Errorf("subscriber error: %s", strings.Join(errors, "\n"))
+		}
+		return err
+	}
+}
+
+func (s *subscriber) Topic() string {
+	return s.topic
+}
+
+func (s *subscriber) Subscriber() interface{} {
+	return s.subscriber
+}
+
+func (s *subscriber) Endpoints() []*register.Endpoint {
+	return s.endpoints
+}
+
+func (s *subscriber) Options() SubscriberOptions {
+	return s.opts
+}
+
+type subscriber struct {
+	typ        reflect.Type
+	subscriber interface{}
+	topic      string
+	endpoints  []*register.Endpoint
+	handlers   []*handler
+	opts       SubscriberOptions
+	rcvr       reflect.Value
+}
+
+type handler struct {
+	reqType reflect.Type
+	ctxType reflect.Type
+	method  reflect.Value
+}

server/noop_test.go

@@ -9,7 +9,6 @@ import (
 	"go.unistack.org/micro/v3/client"
 	"go.unistack.org/micro/v3/codec"
 	"go.unistack.org/micro/v3/logger"
-	"go.unistack.org/micro/v3/metadata"
 	"go.unistack.org/micro/v3/server"
 )
 
@@ -26,18 +25,6 @@ func (h *TestHandler) SingleSubHandler(ctx context.Context, msg *codec.Frame) er
 	return nil
 }
 
-func (h *TestHandler) BatchSubHandler(ctxs []context.Context, msgs []*codec.Frame) error {
-	if len(msgs) != 8 {
-		h.t.Fatal("invalid number of messages received")
-	}
-	for idx := 0; idx < len(msgs); idx++ {
-		md, _ := metadata.FromIncomingContext(ctxs[idx])
-		_ = md
-		//	fmt.Printf("msg md %v\n", md)
-	}
-	return nil
-}
-
 func TestNoopSub(t *testing.T) {
 	ctx := context.Background()
 
@@ -76,13 +63,6 @@ func TestNoopSub(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if err := s.Subscribe(s.NewSubscriber("batch_topic", h.BatchSubHandler,
-		server.SubscriberQueue("queue"),
-		server.SubscriberBatch(true),
-	)); err != nil {
-		t.Fatal(err)
-	}
-
 	if err := s.Start(); err != nil {
 		t.Fatal(err)
 	}

server/options.go

@@ -69,12 +69,6 @@ type Options struct {
 	Advertise string
 	// Version holds the server version
 	Version string
-	// SubWrappers holds the server subscribe wrappers
-	SubWrappers []SubscriberWrapper
-	// BatchSubWrappers holds the server batch subscribe wrappers
-	BatchSubWrappers []BatchSubscriberWrapper
-	// HdlrWrappers holds the handler wrappers
-	HdlrWrappers []HandlerWrapper
 	// RegisterAttempts holds the number of register attempts before error
 	RegisterAttempts int
 	// RegisterInterval holds he interval for re-register
@@ -85,7 +79,8 @@ type Options struct {
 	MaxConn int
 	// DeregisterAttempts holds the number of deregister attempts before error
 	DeregisterAttempts int
-	// Hooks may contains SubscriberWrapper, HandlerWrapper or Server func wrapper
+	// Hooks may contains hook actions that performs before/after server handler
+	// or server subscriber handler
 	Hooks options.Hooks
 	// GracefulTimeout timeout for graceful stop server
 	GracefulTimeout time.Duration
@@ -287,27 +282,6 @@ func Wait(wg *sync.WaitGroup) Option {
 	}
 }
 
-// WrapHandler adds a handler Wrapper to a list of options passed into the server
-func WrapHandler(w HandlerWrapper) Option {
-	return func(o *Options) {
-		o.HdlrWrappers = append(o.HdlrWrappers, w)
-	}
-}
-
-// WrapSubscriber adds a subscriber Wrapper to a list of options passed into the server
-func WrapSubscriber(w SubscriberWrapper) Option {
-	return func(o *Options) {
-		o.SubWrappers = append(o.SubWrappers, w)
-	}
-}
-
-// WrapBatchSubscriber adds a batch subscriber Wrapper to a list of options passed into the server
-func WrapBatchSubscriber(w BatchSubscriberWrapper) Option {
-	return func(o *Options) {
-		o.BatchSubWrappers = append(o.BatchSubWrappers, w)
-	}
-}
-
 // MaxConn specifies maximum number of max simultaneous connections to server
 func MaxConn(n int) Option {
 	return func(o *Options) {
@@ -367,8 +341,6 @@ type SubscriberOptions struct {
 	AutoAck bool
 	// BodyOnly flag specifies that message without headers
 	BodyOnly bool
-	// Batch flag specifies that message processed in batches
-	Batch bool
 	// BatchSize flag specifies max size of batch
 	BatchSize int
 	// BatchWait flag specifies max wait time for batch filling
@@ -440,13 +412,6 @@ func SubscriberAck(b bool) SubscriberOption {
 	}
 }
 
-// SubscriberBatch control batch processing for handler
-func SubscriberBatch(b bool) SubscriberOption {
-	return func(o *SubscriberOptions) {
-		o.Batch = b
-	}
-}
-
 // SubscriberBatchSize control batch filling size for handler
 // Batch filling max waiting time controlled by SubscriberBatchWait
 func SubscriberBatchSize(n int) SubscriberOption {
@@ -461,3 +426,10 @@ func SubscriberBatchWait(td time.Duration) SubscriberOption {
 		o.BatchWait = td
 	}
 }
+
+// Hooks sets hook runs before action
+func Hooks(h ...options.Hook) Option {
+	return func(o *Options) {
+		o.Hooks = append(o.Hooks, h...)
+	}
+}

server/request.go

@@ -1,35 +0,0 @@
-package server
-
-import (
-	"go.unistack.org/micro/v3/codec"
-	"go.unistack.org/micro/v3/metadata"
-)
-
-type rpcMessage struct {
-	payload     interface{}
-	codec       codec.Codec
-	header      metadata.Metadata
-	topic       string
-	contentType string
-	body        []byte
-}
-
-func (r *rpcMessage) ContentType() string {
-	return r.contentType
-}
-
-func (r *rpcMessage) Topic() string {
-	return r.topic
-}
-
-func (r *rpcMessage) Body() interface{} {
-	return r.payload
-}
-
-func (r *rpcMessage) Header() metadata.Metadata {
-	return r.header
-}
-
-func (r *rpcMessage) Codec() codec.Codec {
-	return r.codec
-}

server/server.go

@@ -62,6 +62,13 @@ type Server interface {
 	String() string
 }
 
+type (
+	FuncSubHandler func(ctx context.Context, ms Message) error
+	HookSubHandler func(next FuncSubHandler) FuncSubHandler
+	FuncHandler    func(ctx context.Context, req Request, rsp interface{}) error
+	HookHandler    func(next FuncHandler) FuncHandler
+)
+
 /*
 // Router handle serving messages
 type Router interface {
@@ -147,12 +154,11 @@ type Stream interface {
 //
 // Example:
 //
-//      type Greeter struct {}
-//
-//      func (g *Greeter) Hello(context, request, response) error {
-//              return nil
-//      }
+//	type Greeter struct {}
 //
+//	func (g *Greeter) Hello(context, request, response) error {
+//	        return nil
+//	}
 type Handler interface {
 	Name() string
 	Handler() interface{}

server/subscriber.go

@@ -1,52 +1,24 @@
 package server
 
 import (
-	"bytes"
-	"context"
 	"fmt"
 	"reflect"
-	"runtime/debug"
-	"strings"
 	"unicode"
 	"unicode/utf8"
-
-	"go.unistack.org/micro/v3/broker"
-	"go.unistack.org/micro/v3/codec"
-	"go.unistack.org/micro/v3/errors"
-	"go.unistack.org/micro/v3/logger"
-	"go.unistack.org/micro/v3/metadata"
-	"go.unistack.org/micro/v3/register"
 )
 
 const (
-	subSig      = "func(context.Context, interface{}) error"
-	batchSubSig = "func([]context.Context, []interface{}) error"
+	subSig = "func(context.Context, interface{}) error"
 )
 
 // Precompute the reflect type for error. Can't use error directly
 // because Typeof takes an empty interface value. This is annoying.
 var typeOfError = reflect.TypeOf((*error)(nil)).Elem()
 
-type handler struct {
-	reqType reflect.Type
-	ctxType reflect.Type
-	method  reflect.Value
-}
-
-type subscriber struct {
-	typ        reflect.Type
-	subscriber interface{}
-	topic      string
-	endpoints  []*register.Endpoint
-	handlers   []*handler
-	opts       SubscriberOptions
-	rcvr       reflect.Value
-}
-
 // Is this an exported - upper case - name?
 func isExported(name string) bool {
-	rune, _ := utf8.DecodeRuneInString(name)
-	return unicode.IsUpper(rune)
+	r, _ := utf8.DecodeRuneInString(name)
+	return unicode.IsUpper(r)
 }
 
 // Is this type exported or a builtin?
@@ -69,23 +41,15 @@ func ValidateSubscriber(sub Subscriber) error {
 		switch typ.NumIn() {
 		case 2:
 			argType = typ.In(1)
-			if sub.Options().Batch {
-				if argType.Kind() != reflect.Slice {
-					return fmt.Errorf("subscriber %v dont have required signature %s", name, batchSubSig)
-				}
-				if strings.Compare(fmt.Sprintf("%v", argType), "[]interface{}") == 0 {
-					return fmt.Errorf("subscriber %v dont have required signaure %s", name, batchSubSig)
-				}
-			}
 		default:
-			return fmt.Errorf("subscriber %v takes wrong number of args: %v required signature %s or %s", name, typ.NumIn(), subSig, batchSubSig)
+			return fmt.Errorf("subscriber %v takes wrong number of args: %v required signature %s", name, typ.NumIn(), subSig)
 		}
 		if !isExportedOrBuiltinType(argType) {
 			return fmt.Errorf("subscriber %v argument type not exported: %v", name, argType)
 		}
 		if typ.NumOut() != 1 {
-			return fmt.Errorf("subscriber %v has wrong number of return values: %v require signature %s or %s",
-				name, typ.NumOut(), subSig, batchSubSig)
+			return fmt.Errorf("subscriber %v has wrong number of return values: %v require signature %s",
+				name, typ.NumOut(), subSig)
 		}
 		if returnType := typ.Out(0); returnType != typeOfError {
 			return fmt.Errorf("subscriber %v returns %v not error", name, returnType.String())
@@ -100,8 +64,8 @@ func ValidateSubscriber(sub Subscriber) error {
 			case 3:
 				argType = method.Type.In(2)
 			default:
-				return fmt.Errorf("subscriber %v.%v takes wrong number of args: %v required signature %s or %s",
-					name, method.Name, method.Type.NumIn(), subSig, batchSubSig)
+				return fmt.Errorf("subscriber %v.%v takes wrong number of args: %v required signature %s",
+					name, method.Name, method.Type.NumIn(), subSig)
 			}
 
 			if !isExportedOrBuiltinType(argType) {
@@ -109,8 +73,8 @@ func ValidateSubscriber(sub Subscriber) error {
 			}
 			if method.Type.NumOut() != 1 {
 				return fmt.Errorf(
-					"subscriber %v.%v has wrong number of return values: %v require signature %s or %s",
-					name, method.Name, method.Type.NumOut(), subSig, batchSubSig)
+					"subscriber %v.%v has wrong number of return values: %v require signature %s",
+					name, method.Name, method.Type.NumOut(), subSig)
 			}
 			if returnType := method.Type.Out(0); returnType != typeOfError {
 				return fmt.Errorf("subscriber %v.%v returns %v not error", name, method.Name, returnType.String())
@@ -120,318 +84,3 @@ func ValidateSubscriber(sub Subscriber) error {
 
 	return nil
 }
-
-func newSubscriber(topic string, sub interface{}, opts ...SubscriberOption) Subscriber {
-	var endpoints []*register.Endpoint
-	var handlers []*handler
-
-	options := NewSubscriberOptions(opts...)
-
-	if typ := reflect.TypeOf(sub); typ.Kind() == reflect.Func {
-		h := &handler{
-			method: reflect.ValueOf(sub),
-		}
-
-		switch typ.NumIn() {
-		case 1:
-			h.reqType = typ.In(0)
-		case 2:
-			h.ctxType = typ.In(0)
-			h.reqType = typ.In(1)
-		}
-
-		handlers = append(handlers, h)
-		ep := &register.Endpoint{
-			Name:     "Func",
-			Request:  register.ExtractSubValue(typ),
-			Metadata: metadata.New(2),
-		}
-		ep.Metadata.Set("topic", topic)
-		ep.Metadata.Set("subscriber", "true")
-		endpoints = append(endpoints, ep)
-	} else {
-		hdlr := reflect.ValueOf(sub)
-		name := reflect.Indirect(hdlr).Type().Name()
-
-		for m := 0; m < typ.NumMethod(); m++ {
-			method := typ.Method(m)
-			h := &handler{
-				method: method.Func,
-			}
-
-			switch method.Type.NumIn() {
-			case 2:
-				h.reqType = method.Type.In(1)
-			case 3:
-				h.ctxType = method.Type.In(1)
-				h.reqType = method.Type.In(2)
-			}
-
-			handlers = append(handlers, h)
-			ep := &register.Endpoint{
-				Name:     name + "." + method.Name,
-				Request:  register.ExtractSubValue(method.Type),
-				Metadata: metadata.New(2),
-			}
-			ep.Metadata.Set("topic", topic)
-			ep.Metadata.Set("subscriber", "true")
-			endpoints = append(endpoints, ep)
-		}
-	}
-
-	return &subscriber{
-		rcvr:       reflect.ValueOf(sub),
-		typ:        reflect.TypeOf(sub),
-		topic:      topic,
-		subscriber: sub,
-		handlers:   handlers,
-		endpoints:  endpoints,
-		opts:       options,
-	}
-}
-
-//nolint:gocyclo
-func (n *noopServer) createBatchSubHandler(sb *subscriber, opts Options) broker.BatchHandler {
-	return func(ps broker.Events) (err error) {
-		defer func() {
-			if r := recover(); r != nil {
-				n.RLock()
-				config := n.opts
-				n.RUnlock()
-				if config.Logger.V(logger.ErrorLevel) {
-					config.Logger.Error(n.opts.Context, "panic recovered: ", r)
-					config.Logger.Error(n.opts.Context, string(debug.Stack()))
-				}
-				err = errors.InternalServerError(n.opts.Name+".subscriber", "panic recovered: %v", r)
-			}
-		}()
-
-		msgs := make([]Message, 0, len(ps))
-		ctxs := make([]context.Context, 0, len(ps))
-		for _, p := range ps {
-			msg := p.Message()
-			// if we don't have headers, create empty map
-			if msg.Header == nil {
-				msg.Header = metadata.New(2)
-			}
-
-			ct, _ := msg.Header.Get(metadata.HeaderContentType)
-			if len(ct) == 0 {
-				msg.Header.Set(metadata.HeaderContentType, defaultContentType)
-				ct = defaultContentType
-			}
-			hdr := metadata.Copy(msg.Header)
-			topic, _ := msg.Header.Get(metadata.HeaderTopic)
-			ctxs = append(ctxs, metadata.NewIncomingContext(sb.opts.Context, hdr))
-			msgs = append(msgs, &rpcMessage{
-				topic:       topic,
-				contentType: ct,
-				header:      msg.Header,
-				body:        msg.Body,
-			})
-		}
-		results := make(chan error, len(sb.handlers))
-
-		for i := 0; i < len(sb.handlers); i++ {
-			handler := sb.handlers[i]
-
-			var req reflect.Value
-
-			switch handler.reqType.Kind() {
-			case reflect.Ptr:
-				req = reflect.New(handler.reqType.Elem())
-			default:
-				req = reflect.New(handler.reqType.Elem()).Elem()
-			}
-
-			reqType := handler.reqType
-			var cf codec.Codec
-			for _, msg := range msgs {
-				cf, err = n.newCodec(msg.ContentType())
-				if err != nil {
-					return err
-				}
-				rb := reflect.New(req.Type().Elem())
-				if err = cf.ReadBody(bytes.NewReader(msg.(*rpcMessage).body), rb.Interface()); err != nil {
-					return err
-				}
-				msg.(*rpcMessage).codec = cf
-				msg.(*rpcMessage).payload = rb.Interface()
-			}
-
-			fn := func(ctxs []context.Context, ms []Message) error {
-				var vals []reflect.Value
-				if sb.typ.Kind() != reflect.Func {
-					vals = append(vals, sb.rcvr)
-				}
-				if handler.ctxType != nil {
-					vals = append(vals, reflect.ValueOf(ctxs))
-				}
-				payloads := reflect.MakeSlice(reqType, 0, len(ms))
-				for _, m := range ms {
-					payloads = reflect.Append(payloads, reflect.ValueOf(m.Body()))
-				}
-				vals = append(vals, payloads)
-
-				returnValues := handler.method.Call(vals)
-				if rerr := returnValues[0].Interface(); rerr != nil {
-					return rerr.(error)
-				}
-				return nil
-			}
-
-			for i := len(opts.BatchSubWrappers); i > 0; i-- {
-				fn = opts.BatchSubWrappers[i-1](fn)
-			}
-
-			if n.wg != nil {
-				n.wg.Add(1)
-			}
-			go func() {
-				if n.wg != nil {
-					defer n.wg.Done()
-				}
-				results <- fn(ctxs, msgs)
-			}()
-		}
-
-		var errors []string
-		for i := 0; i < len(sb.handlers); i++ {
-			if rerr := <-results; rerr != nil {
-				errors = append(errors, rerr.Error())
-			}
-		}
-		if len(errors) > 0 {
-			err = fmt.Errorf("subscriber error: %s", strings.Join(errors, "\n"))
-		}
-		return err
-	}
-}
-
-//nolint:gocyclo
-func (n *noopServer) createSubHandler(sb *subscriber, opts Options) broker.Handler {
-	return func(p broker.Event) (err error) {
-		defer func() {
-			if r := recover(); r != nil {
-				n.RLock()
-				config := n.opts
-				n.RUnlock()
-				if config.Logger.V(logger.ErrorLevel) {
-					config.Logger.Error(n.opts.Context, "panic recovered: ", r)
-					config.Logger.Error(n.opts.Context, string(debug.Stack()))
-				}
-				err = errors.InternalServerError(n.opts.Name+".subscriber", "panic recovered: %v", r)
-			}
-		}()
-
-		msg := p.Message()
-		// if we don't have headers, create empty map
-		if msg.Header == nil {
-			msg.Header = metadata.New(2)
-		}
-
-		ct := msg.Header["Content-Type"]
-		if len(ct) == 0 {
-			msg.Header.Set(metadata.HeaderContentType, defaultContentType)
-			ct = defaultContentType
-		}
-		cf, err := n.newCodec(ct)
-		if err != nil {
-			return err
-		}
-
-		hdr := metadata.New(len(msg.Header))
-		for k, v := range msg.Header {
-			hdr.Set(k, v)
-		}
-
-		ctx := metadata.NewIncomingContext(sb.opts.Context, hdr)
-
-		results := make(chan error, len(sb.handlers))
-
-		for i := 0; i < len(sb.handlers); i++ {
-			handler := sb.handlers[i]
-
-			var isVal bool
-			var req reflect.Value
-
-			if handler.reqType.Kind() == reflect.Ptr {
-				req = reflect.New(handler.reqType.Elem())
-			} else {
-				req = reflect.New(handler.reqType)
-				isVal = true
-			}
-			if isVal {
-				req = req.Elem()
-			}
-
-			if err = cf.ReadBody(bytes.NewBuffer(msg.Body), req.Interface()); err != nil {
-				return err
-			}
-
-			fn := func(ctx context.Context, msg Message) error {
-				var vals []reflect.Value
-				if sb.typ.Kind() != reflect.Func {
-					vals = append(vals, sb.rcvr)
-				}
-				if handler.ctxType != nil {
-					vals = append(vals, reflect.ValueOf(ctx))
-				}
-
-				vals = append(vals, reflect.ValueOf(msg.Body()))
-
-				returnValues := handler.method.Call(vals)
-				if rerr := returnValues[0].Interface(); rerr != nil {
-					return rerr.(error)
-				}
-				return nil
-			}
-
-			for i := len(opts.SubWrappers); i > 0; i-- {
-				fn = opts.SubWrappers[i-1](fn)
-			}
-
-			if n.wg != nil {
-				n.wg.Add(1)
-			}
-			go func() {
-				if n.wg != nil {
-					defer n.wg.Done()
-				}
-				cerr := fn(ctx, &rpcMessage{
-					topic:       sb.topic,
-					contentType: ct,
-					payload:     req.Interface(),
-					header:      msg.Header,
-				})
-				results <- cerr
-			}()
-		}
-		var errors []string
-		for i := 0; i < len(sb.handlers); i++ {
-			if rerr := <-results; rerr != nil {
-				errors = append(errors, rerr.Error())
-			}
-		}
-		if len(errors) > 0 {
-			err = fmt.Errorf("subscriber error: %s", strings.Join(errors, "\n"))
-		}
-		return err
-	}
-}
-
-func (s *subscriber) Topic() string {
-	return s.topic
-}
-
-func (s *subscriber) Subscriber() interface{} {
-	return s.subscriber
-}
-
-func (s *subscriber) Endpoints() []*register.Endpoint {
-	return s.endpoints
-}
-
-func (s *subscriber) Options() SubscriberOptions {
-	return s.opts
-}

server/wrapper.go

@@ -14,20 +14,12 @@ type HandlerFunc func(ctx context.Context, req Request, rsp interface{}) error
 // publication message.
 type SubscriberFunc func(ctx context.Context, msg Message) error
 
-// BatchSubscriberFunc represents a single method of a subscriber. It's used primarily
-// for the wrappers. What's handed to the actual method is the concrete
-// publication message. This func used by batch subscribers
-type BatchSubscriberFunc func(ctxs []context.Context, msgs []Message) error
-
 // HandlerWrapper wraps the HandlerFunc and returns the equivalent
 type HandlerWrapper func(HandlerFunc) HandlerFunc
 
 // SubscriberWrapper wraps the SubscriberFunc and returns the equivalent
 type SubscriberWrapper func(SubscriberFunc) SubscriberFunc
 
-// BatchSubscriberWrapper wraps the SubscriberFunc and returns the equivalent
-type BatchSubscriberWrapper func(BatchSubscriberFunc) BatchSubscriberFunc
-
 // StreamWrapper wraps a Stream interface and returns the equivalent.
 // Because streams exist for the lifetime of a method invocation this
 // is a convenient way to wrap a Stream as its in use for trace, monitoring,

store/memory/memory.go

@@ -1,4 +1,4 @@
-package store
+package memory
 
 import (
 	"context"
@@ -6,13 +6,15 @@ import (
 	"strings"
 	"time"
 
-	"github.com/patrickmn/go-cache"
+	cache "github.com/patrickmn/go-cache"
+	"go.unistack.org/micro/v3/options"
+	"go.unistack.org/micro/v3/store"
 )
 
 // NewStore returns a memory store
-func NewStore(opts ...Option) Store {
+func NewStore(opts ...store.Option) store.Store {
 	return &memoryStore{
-		opts:  NewOptions(opts...),
+		opts:  store.NewOptions(opts...),
 		store: cache.New(cache.NoExpiration, 5*time.Minute),
 	}
 }
@@ -27,8 +29,13 @@ func (m *memoryStore) Disconnect(ctx context.Context) error {
 }
 
 type memoryStore struct {
-	store *cache.Cache
-	opts  Options
+	funcRead   store.FuncRead
+	funcWrite  store.FuncWrite
+	funcExists store.FuncExists
+	funcList   store.FuncList
+	funcDelete store.FuncDelete
+	store      *cache.Cache
+	opts       store.Options
 }
 
 func (m *memoryStore) key(prefix, key string) string {
@@ -39,7 +46,7 @@ func (m *memoryStore) exists(prefix, key string) error {
 	key = m.key(prefix, key)
 	_, found := m.store.Get(key)
 	if !found {
-		return ErrNotFound
+		return store.ErrNotFound
 	}
 
 	return nil
@@ -50,12 +57,12 @@ func (m *memoryStore) get(prefix, key string, val interface{}) error {
 
 	r, found := m.store.Get(key)
 	if !found {
-		return ErrNotFound
+		return store.ErrNotFound
 	}
 
 	buf, ok := r.([]byte)
 	if !ok {
-		return ErrNotFound
+		return store.ErrNotFound
 	}
 
 	return m.opts.Codec.Unmarshal(buf, val)
@@ -100,10 +107,32 @@ func (m *memoryStore) list(prefix string, limit, offset uint) []string {
 	return allKeys
 }
 
-func (m *memoryStore) Init(opts ...Option) error {
+func (m *memoryStore) Init(opts ...store.Option) error {
 	for _, o := range opts {
 		o(&m.opts)
 	}
+
+	m.funcRead = m.fnRead
+	m.funcWrite = m.fnWrite
+	m.funcExists = m.fnExists
+	m.funcList = m.fnList
+	m.funcDelete = m.fnDelete
+
+	m.opts.Hooks.EachNext(func(hook options.Hook) {
+		switch h := hook.(type) {
+		case store.HookRead:
+			m.funcRead = h(m.funcRead)
+		case store.HookWrite:
+			m.funcWrite = h(m.funcWrite)
+		case store.HookExists:
+			m.funcExists = h(m.funcExists)
+		case store.HookList:
+			m.funcList = h(m.funcList)
+		case store.HookDelete:
+			m.funcDelete = h(m.funcDelete)
+		}
+	})
+
 	return nil
 }
 
@@ -115,24 +144,36 @@ func (m *memoryStore) Name() string {
 	return m.opts.Name
 }
 
-func (m *memoryStore) Exists(ctx context.Context, key string, opts ...ExistsOption) error {
-	options := NewExistsOptions(opts...)
+func (m *memoryStore) Exists(ctx context.Context, key string, opts ...store.ExistsOption) error {
+	return m.funcExists(ctx, key, opts...)
+}
+
+func (m *memoryStore) fnExists(ctx context.Context, key string, opts ...store.ExistsOption) error {
+	options := store.NewExistsOptions(opts...)
 	if options.Namespace == "" {
 		options.Namespace = m.opts.Namespace
 	}
 	return m.exists(options.Namespace, key)
 }
 
-func (m *memoryStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
-	options := NewReadOptions(opts...)
+func (m *memoryStore) Read(ctx context.Context, key string, val interface{}, opts ...store.ReadOption) error {
+	return m.funcRead(ctx, key, val, opts...)
+}
+
+func (m *memoryStore) fnRead(ctx context.Context, key string, val interface{}, opts ...store.ReadOption) error {
+	options := store.NewReadOptions(opts...)
 	if options.Namespace == "" {
 		options.Namespace = m.opts.Namespace
 	}
 	return m.get(options.Namespace, key, val)
 }
 
-func (m *memoryStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
-	options := NewWriteOptions(opts...)
+func (m *memoryStore) Write(ctx context.Context, key string, val interface{}, opts ...store.WriteOption) error {
+	return m.funcWrite(ctx, key, val, opts...)
+}
+
+func (m *memoryStore) fnWrite(ctx context.Context, key string, val interface{}, opts ...store.WriteOption) error {
+	options := store.NewWriteOptions(opts...)
 	if options.Namespace == "" {
 		options.Namespace = m.opts.Namespace
 	}
@@ -151,8 +192,12 @@ func (m *memoryStore) Write(ctx context.Context, key string, val interface{}, op
 	return nil
 }
 
-func (m *memoryStore) Delete(ctx context.Context, key string, opts ...DeleteOption) error {
-	options := NewDeleteOptions(opts...)
+func (m *memoryStore) Delete(ctx context.Context, key string, opts ...store.DeleteOption) error {
+	return m.funcDelete(ctx, key, opts...)
+}
+
+func (m *memoryStore) fnDelete(ctx context.Context, key string, opts ...store.DeleteOption) error {
+	options := store.NewDeleteOptions(opts...)
 	if options.Namespace == "" {
 		options.Namespace = m.opts.Namespace
 	}
@@ -161,12 +206,16 @@ func (m *memoryStore) Delete(ctx context.Context, key string, opts ...DeleteOpti
 	return nil
 }
 
-func (m *memoryStore) Options() Options {
+func (m *memoryStore) Options() store.Options {
 	return m.opts
 }
 
-func (m *memoryStore) List(ctx context.Context, opts ...ListOption) ([]string, error) {
-	options := NewListOptions(opts...)
+func (m *memoryStore) List(ctx context.Context, opts ...store.ListOption) ([]string, error) {
+	return m.funcList(ctx, opts...)
+}
+
+func (m *memoryStore) fnList(ctx context.Context, opts ...store.ListOption) ([]string, error) {
+	options := store.NewListOptions(opts...)
 	if options.Namespace == "" {
 		options.Namespace = m.opts.Namespace
 	}

store/memory/memory_test.go

@@ -1,4 +1,4 @@
-package store_test
+package memory
 
 import (
 	"context"
@@ -8,8 +8,41 @@ import (
 	"go.unistack.org/micro/v3/store"
 )
 
+type testHook struct {
+	f bool
+}
+
+func (t *testHook) Exists(fn store.FuncExists) store.FuncExists {
+	return func(ctx context.Context, key string, opts ...store.ExistsOption) error {
+		t.f = true
+		return fn(ctx, key, opts...)
+	}
+}
+
+func TestHook(t *testing.T) {
+	h := &testHook{}
+
+	s := NewStore(store.Hooks(store.HookExists(h.Exists)))
+
+	if err := s.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.Write(context.TODO(), "test", nil); err != nil {
+		t.Error(err)
+	}
+
+	if err := s.Exists(context.TODO(), "test"); err != nil {
+		t.Fatal(err)
+	}
+
+	if !h.f {
+		t.Fatal("hook not works")
+	}
+}
+
 func TestMemoryReInit(t *testing.T) {
-	s := store.NewStore(store.Namespace("aaa"))
+	s := NewStore(store.Namespace("aaa"))
 	if err := s.Init(store.Namespace("")); err != nil {
 		t.Fatal(err)
 	}
@@ -19,7 +52,7 @@ func TestMemoryReInit(t *testing.T) {
 }
 
 func TestMemoryBasic(t *testing.T) {
-	s := store.NewStore()
+	s := NewStore()
 	if err := s.Init(); err != nil {
 		t.Fatal(err)
 	}
@@ -27,7 +60,7 @@ func TestMemoryBasic(t *testing.T) {
 }
 
 func TestMemoryPrefix(t *testing.T) {
-	s := store.NewStore()
+	s := NewStore()
 	if err := s.Init(store.Namespace("some-prefix")); err != nil {
 		t.Fatal(err)
 	}
@@ -35,7 +68,7 @@ func TestMemoryPrefix(t *testing.T) {
 }
 
 func TestMemoryNamespace(t *testing.T) {
-	s := store.NewStore()
+	s := NewStore()
 	if err := s.Init(store.Namespace("some-namespace")); err != nil {
 		t.Fatal(err)
 	}
@@ -43,7 +76,7 @@ func TestMemoryNamespace(t *testing.T) {
 }
 
 func TestMemoryNamespacePrefix(t *testing.T) {
-	s := store.NewStore()
+	s := NewStore()
 	if err := s.Init(store.Namespace("some-namespace")); err != nil {
 		t.Fatal(err)
 	}

store/noop.go

@@ -0,0 +1,147 @@
+package store
+
+import (
+	"context"
+
+	"go.unistack.org/micro/v3/options"
+)
+
+var _ Store = (*noopStore)(nil)
+
+type noopStore struct {
+	funcRead   FuncRead
+	funcWrite  FuncWrite
+	funcExists FuncExists
+	funcList   FuncList
+	funcDelete FuncDelete
+	opts       Options
+}
+
+func NewStore(opts ...Option) *noopStore {
+	options := NewOptions(opts...)
+	return &noopStore{opts: options}
+}
+
+func (n *noopStore) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&n.opts)
+	}
+
+	n.funcRead = n.fnRead
+	n.funcWrite = n.fnWrite
+	n.funcExists = n.fnExists
+	n.funcList = n.fnList
+	n.funcDelete = n.fnDelete
+
+	n.opts.Hooks.EachNext(func(hook options.Hook) {
+		switch h := hook.(type) {
+		case HookRead:
+			n.funcRead = h(n.funcRead)
+		case HookWrite:
+			n.funcWrite = h(n.funcWrite)
+		case HookExists:
+			n.funcExists = h(n.funcExists)
+		case HookList:
+			n.funcList = h(n.funcList)
+		case HookDelete:
+			n.funcDelete = h(n.funcDelete)
+		}
+	})
+
+	return nil
+}
+
+func (n *noopStore) Connect(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	return nil
+}
+
+func (n *noopStore) Disconnect(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	return nil
+}
+
+func (n *noopStore) Read(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
+	return n.funcRead(ctx, key, val, opts...)
+}
+
+func (n *noopStore) fnRead(ctx context.Context, key string, val interface{}, opts ...ReadOption) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	return nil
+}
+
+func (n *noopStore) Delete(ctx context.Context, key string, opts ...DeleteOption) error {
+	return n.funcDelete(ctx, key, opts...)
+}
+
+func (n *noopStore) fnDelete(ctx context.Context, key string, opts ...DeleteOption) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	return nil
+}
+
+func (n *noopStore) Exists(ctx context.Context, key string, opts ...ExistsOption) error {
+	return n.funcExists(ctx, key, opts...)
+}
+
+func (n *noopStore) fnExists(ctx context.Context, key string, opts ...ExistsOption) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	return nil
+}
+
+func (n *noopStore) Write(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
+	return n.funcWrite(ctx, key, val, opts...)
+}
+
+func (n *noopStore) fnWrite(ctx context.Context, key string, val interface{}, opts ...WriteOption) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+	return nil
+}
+
+func (n *noopStore) List(ctx context.Context, opts ...ListOption) ([]string, error) {
+	return n.funcList(ctx, opts...)
+}
+
+func (n *noopStore) fnList(ctx context.Context, opts ...ListOption) ([]string, error) {
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	default:
+	}
+	return nil, nil
+}
+
+func (n *noopStore) Name() string {
+	return n.opts.Name
+}
+
+func (n *noopStore) String() string {
+	return "noop"
+}
+
+func (n *noopStore) Options() Options {
+	return n.opts
+}

store/noop_test.go

@@ -0,0 +1,35 @@
+package store
+
+import (
+	"context"
+	"testing"
+)
+
+type testHook struct {
+	f bool
+}
+
+func (t *testHook) Exists(fn FuncExists) FuncExists {
+	return func(ctx context.Context, key string, opts ...ExistsOption) error {
+		t.f = true
+		return fn(ctx, key, opts...)
+	}
+}
+
+func TestHook(t *testing.T) {
+	h := &testHook{}
+
+	s := NewStore(Hooks(HookExists(h.Exists)))
+
+	if err := s.Init(); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := s.Exists(context.TODO(), "test"); err != nil {
+		t.Fatal(err)
+	}
+
+	if !h.f {
+		t.Fatal("hook not works")
+	}
+}

store/options.go

@@ -9,6 +9,7 @@ import (
 	"go.unistack.org/micro/v3/logger"
 	"go.unistack.org/micro/v3/metadata"
 	"go.unistack.org/micro/v3/meter"
+	"go.unistack.org/micro/v3/options"
 	"go.unistack.org/micro/v3/tracer"
 )
 
@@ -38,6 +39,8 @@ type Options struct {
 	// Wrappers []Wrapper
 	// Timeout specifies timeout duration for all operations
 	Timeout time.Duration
+	// Hooks can be run before/after store Read/List/Write/Exists/Delete
+	Hooks options.Hooks
 }
 
 // NewOptions creates options struct
@@ -441,11 +444,9 @@ func ExistsTimeout(td time.Duration) ExistsOption {
 	}
 }
 
-/*
-// WrapStore adds a store Wrapper to a list of options passed into the store
-func WrapStore(w Wrapper) Option {
+// Hooks sets hook runs before action
+func Hooks(h ...options.Hook) Option {
 	return func(o *Options) {
-		o.Wrappers = append(o.Wrappers, w)
+		o.Hooks = append(o.Hooks, h...)
 	}
 }
-*/

store/store.go

@@ -19,6 +19,7 @@ var (
 
 // Store is a data storage interface
 type Store interface {
+	// Name returns store name
 	Name() string
 	// Init initialises the store
 	Init(opts ...Option) error
@@ -41,3 +42,16 @@ type Store interface {
 	// String returns the name of the implementation.
 	String() string
 }
+
+type (
+	FuncExists func(ctx context.Context, key string, opts ...ExistsOption) error
+	HookExists func(next FuncExists) FuncExists
+	FuncRead   func(ctx context.Context, key string, val interface{}, opts ...ReadOption) error
+	HookRead   func(next FuncRead) FuncRead
+	FuncWrite  func(ctx context.Context, key string, val interface{}, opts ...WriteOption) error
+	HookWrite  func(next FuncWrite) FuncWrite
+	FuncDelete func(ctx context.Context, key string, opts ...DeleteOption) error
+	HookDelete func(next FuncDelete) FuncDelete
+	FuncList   func(ctx context.Context, opts ...ListOption) ([]string, error)
+	HookList   func(next FuncList) FuncList
+)

tracer/tracer_test.go

@@ -1,13 +0,0 @@
-package tracer
-
-import (
-	"testing"
-)
-
-func TestUniqLabels(t *testing.T) {
-	labels := []interface{}{"key1", "val1", "key1", "val2"}
-	labels = UniqLabels(labels)
-	if labels[1] != "val2" {
-		t.Fatalf("UniqLabels not works")
-	}
-}

util/reflect/struct.go

@@ -25,6 +25,48 @@ type StructField struct {
 	Field reflect.StructField
 }
 
+// StructFieldNameByTag get struct field name by tag key and its value
+func StructFieldNameByTag(src interface{}, tkey string, tval string) (string, error) {
+	sv := reflect.ValueOf(src)
+	if sv.Kind() == reflect.Ptr {
+		sv = sv.Elem()
+	}
+	if sv.Kind() != reflect.Struct {
+		return "", ErrInvalidStruct
+	}
+
+	typ := sv.Type()
+	for idx := 0; idx < typ.NumField(); idx++ {
+		fld := typ.Field(idx)
+		val := sv.Field(idx)
+		if len(fld.PkgPath) != 0 {
+			continue
+		}
+
+		if ts, ok := fld.Tag.Lookup(tkey); ok {
+			for _, p := range strings.Split(ts, ",") {
+				if p == tval {
+					return fld.Name, nil
+				}
+			}
+		}
+
+		switch val.Kind() {
+		case reflect.Ptr:
+			if val = val.Elem(); val.Kind() == reflect.Struct {
+				if name, err := StructFieldNameByTag(val.Interface(), tkey, tval); err == nil {
+					return name, nil
+				}
+			}
+		case reflect.Struct:
+			if name, err := StructFieldNameByTag(val.Interface(), tkey, tval); err == nil {
+				return name, nil
+			}
+		}
+	}
+	return "", ErrNotFound
+}
+
 // StructFieldByTag get struct field by tag key and its value
 func StructFieldByTag(src interface{}, tkey string, tval string) (interface{}, error) {
 	sv := reflect.ValueOf(src)
@@ -46,9 +88,6 @@ func StructFieldByTag(src interface{}, tkey string, tval string) (interface{}, e
 		if ts, ok := fld.Tag.Lookup(tkey); ok {
 			for _, p := range strings.Split(ts, ",") {
 				if p == tval {
-					if val.Kind() != reflect.Ptr && val.CanAddr() {
-						val = val.Addr()
-					}
 					return val.Interface(), nil
 				}
 			}
@@ -493,13 +532,14 @@ func btSplitter(str string) []string {
 }
 
 // queryToMap turns something like a[b][c]=4 into
-//   map[string]interface{}{
-//     "a": map[string]interface{}{
-// 		  "b": map[string]interface{}{
-// 			  "c": 4,
-// 		  },
-// 	  },
-//   }
+//
+//	  map[string]interface{}{
+//	    "a": map[string]interface{}{
+//			  "b": map[string]interface{}{
+//				  "c": 4,
+//			  },
+//		  },
+//	  }
 func queryToMap(param string) (map[string]interface{}, error) {
 	rawKey, rawValue, err := splitKeyAndValue(param)
 	if err != nil {

util/reflect/struct_test.go

@@ -190,9 +190,9 @@ func TestStructFieldByTag(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if v, ok := iface.(*[]string); !ok {
+	if v, ok := iface.([]string); !ok {
 		t.Fatalf("not *[]string %v", iface)
-	} else if len(*v) != 2 {
+	} else if len(v) != 2 {
 		t.Fatalf("invalid number %v", iface)
 	}
 }

util/xpool/pool.go

@@ -0,0 +1,25 @@
+package pool
+
+import "sync"
+
+type Pool[T any] struct {
+	p *sync.Pool
+}
+
+func NewPool[T any](fn func() T) Pool[T] {
+	return Pool[T]{
+		p: &sync.Pool{
+			New: func() interface{} {
+				return fn()
+			},
+		},
+	}
+}
+
+func (p Pool[T]) Get() T {
+	return p.p.Get().(T)
+}
+
+func (p Pool[T]) Put(t T) {
+	p.p.Put(t)
+}

util/xpool/pool_test.go

@@ -0,0 +1,27 @@
+package pool
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+)
+
+func TestBytes(t *testing.T) {
+	p := NewPool(func() *bytes.Buffer { return bytes.NewBuffer(nil) })
+	b := p.Get()
+	b.Write([]byte(`test`))
+	if b.String() != "test" {
+		t.Fatal("pool not works")
+	}
+	p.Put(b)
+}
+
+func TestStrings(t *testing.T) {
+	p := NewPool(func() *strings.Builder { return &strings.Builder{} })
+	b := p.Get()
+	b.Write([]byte(`test`))
+	if b.String() != "test" {
+		t.Fatal("pool not works")
+	}
+	p.Put(b)
+}