cleanup interfaces for v4

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

client/client.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"go.unistack.org/micro/v4/codec"
-	"go.unistack.org/micro/v4/metadata"
 )
 
 var (
@@ -35,23 +34,12 @@ type Client interface {
 	Name() string
 	Init(opts ...Option) error
 	Options() Options
-	NewMessage(topic string, msg interface{}, opts ...MessageOption) Message
 	NewRequest(service string, endpoint string, req interface{}, opts ...RequestOption) Request
 	Call(ctx context.Context, req Request, rsp interface{}, opts ...CallOption) error
 	Stream(ctx context.Context, req Request, opts ...CallOption) (Stream, error)
-	Publish(ctx context.Context, msg Message, opts ...PublishOption) error
-	BatchPublish(ctx context.Context, msg []Message, opts ...PublishOption) error
 	String() string
 }
 
-// Message is the interface for publishing asynchronously
-type Message interface {
-	Topic() string
-	Payload() interface{}
-	ContentType() string
-	Metadata() metadata.Metadata
-}
-
 // Request is the interface for a synchronous request used by Call or Stream
 type Request interface {
 	// The service to call
@@ -68,16 +56,22 @@ type Request interface {
 	Codec() codec.Codec
 	// indicates whether the request will be a streaming one rather than unary
 	Stream() bool
+	// Header data
+	// Header() metadata.Metadata
 }
 
 // Response is the response received from a service
 type Response interface {
 	// Read the response
 	Codec() codec.Codec
+	// The content type
+	// ContentType() string
 	// Header data
-	Header() metadata.Metadata
+	// Header() metadata.Metadata
 	// Read the undecoded response
 	Read() ([]byte, error)
+	// The unencoded request body
+	// Body() interface{}
 }
 
 // Stream is the interface for a bidirectional synchronous stream

client/noop.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"time"
 
-	"go.unistack.org/micro/v4/broker"
 	"go.unistack.org/micro/v4/codec"
 	"go.unistack.org/micro/v4/errors"
 	"go.unistack.org/micro/v4/metadata"
@@ -285,6 +284,9 @@ func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opt
 	ch := make(chan error, callOpts.Retries)
 	var gerr error
 
+	ts := time.Now()
+	endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
+	n.opts.Meter.Counter(ClientRequestInflight, "endpoint", endpoint).Inc()
 	for i := 0; i <= callOpts.Retries; i++ {
 		go func() {
 			ch <- call(i)
@@ -312,6 +314,16 @@ func (n *noopClient) Call(ctx context.Context, req Request, rsp interface{}, opt
 		}
 	}
 
+	if gerr != nil {
+		n.opts.Meter.Counter(ClientRequestTotal, "endpoint", endpoint, "status", "failure").Inc()
+	} else {
+		n.opts.Meter.Counter(ClientRequestTotal, "endpoint", endpoint, "status", "success").Inc()
+	}
+	n.opts.Meter.Counter(ClientRequestInflight, "endpoint", endpoint).Dec()
+	te := time.Since(ts)
+	n.opts.Meter.Summary(ClientRequestLatencyMicroseconds, "endpoint", endpoint).Update(te.Seconds())
+	n.opts.Meter.Histogram(ClientRequestDurationSeconds, "endpoint", endpoint).Update(te.Seconds())
+
 	return gerr
 }
 
@@ -323,11 +335,6 @@ func (n *noopClient) NewRequest(service, endpoint string, req interface{}, opts
 	return &noopRequest{service: service, endpoint: endpoint}
 }
 
-func (n *noopClient) NewMessage(topic string, msg interface{}, opts ...MessageOption) Message {
-	options := NewMessageOptions(append([]MessageOption{MessageContentType(n.opts.ContentType)}, opts...)...)
-	return &noopMessage{topic: topic, payload: msg, opts: options}
-}
-
 func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption) (Stream, error) {
 	var err error
 
@@ -414,7 +421,15 @@ func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption
 
 		node := next()
 
+		// ts := time.Now()
+		endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
+		n.opts.Meter.Counter(ClientRequestInflight, "endpoint", endpoint).Inc()
 		stream, cerr := n.stream(ctx, node, req, callOpts)
+		if cerr != nil {
+			n.opts.Meter.Counter(ClientRequestTotal, "endpoint", endpoint, "status", "failure").Inc()
+		} else {
+			n.opts.Meter.Counter(ClientRequestTotal, "endpoint", endpoint, "status", "success").Inc()
+		}
 
 		// record the result of the call to inform future routing decisions
 		if verr := n.opts.Selector.Record(node, cerr); verr != nil {
@@ -468,64 +483,6 @@ func (n *noopClient) Stream(ctx context.Context, req Request, opts ...CallOption
 	return nil, grr
 }
 
-func (n *noopClient) stream(ctx context.Context, addr string, req Request, opts CallOptions) (Stream, error) {
+func (n *noopClient) stream(ctx context.Context, addr string, req Request, opts CallOptions) (*noopStream, error) {
 	return &noopStream{}, nil
 }
-
-func (n *noopClient) BatchPublish(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.publish(ctx, []Message{p}, opts...)
-}
-
-func (n *noopClient) publish(ctx context.Context, ps []Message, opts ...PublishOption) error {
-	options := NewPublishOptions(opts...)
-
-	msgs := make([]*broker.Message, 0, len(ps))
-
-	for _, p := range ps {
-		md, ok := metadata.FromOutgoingContext(ctx)
-		if !ok {
-			md = metadata.New(0)
-		}
-		md[metadata.HeaderContentType] = p.ContentType()
-
-		topic := p.Topic()
-
-		// get the exchange
-		if len(options.Exchange) > 0 {
-			topic = options.Exchange
-		}
-
-		md[metadata.HeaderTopic] = topic
-
-		var body []byte
-
-		// passed in raw data
-		if d, ok := p.Payload().(*codec.Frame); ok {
-			body = d.Data
-		} else {
-			// use codec for payload
-			cf, err := n.newCodec(p.ContentType())
-			if err != nil {
-				return errors.InternalServerError("go.micro.client", err.Error())
-			}
-
-			// set the body
-			b, err := cf.Marshal(p.Payload())
-			if err != nil {
-				return errors.InternalServerError("go.micro.client", err.Error())
-			}
-			body = b
-		}
-
-		msgs = append(msgs, &broker.Message{Header: md, Body: body})
-	}
-
-	return n.opts.Broker.BatchPublish(ctx, msgs,
-		broker.PublishContext(options.Context),
-		broker.PublishBodyOnly(options.BodyOnly),
-	)
-}

client/options.go

@@ -6,7 +6,6 @@ import (
 	"net"
 	"time"
 
-	"go.unistack.org/micro/v4/broker"
 	"go.unistack.org/micro/v4/codec"
 	"go.unistack.org/micro/v4/logger"
 	"go.unistack.org/micro/v4/metadata"
@@ -19,6 +18,17 @@ import (
 	"go.unistack.org/micro/v4/tracer"
 )
 
+var (
+	// ClientRequestDurationSeconds specifies meter metric name
+	ClientRequestDurationSeconds = "client_request_duration_seconds"
+	// ClientRequestLatencyMicroseconds specifies meter metric name
+	ClientRequestLatencyMicroseconds = "client_request_latency_microseconds"
+	// ClientRequestTotal specifies meter metric name
+	ClientRequestTotal = "client_request_total"
+	// ClientRequestInflight specifies meter metric name
+	ClientRequestInflight = "client_request_inflight"
+)
+
 // Options holds client options
 type Options struct {
 	// Transport used for transfer messages
@@ -29,8 +39,6 @@ type Options struct {
 	Logger logger.Logger
 	// Tracer used for tracing
 	Tracer tracer.Tracer
-	// Broker used to publish messages
-	Broker broker.Broker
 	// Meter used for metrics
 	Meter meter.Meter
 	// Context is used for external options
@@ -199,7 +207,6 @@ func NewOptions(opts ...Option) Options {
 		PoolTTL:   DefaultPoolTTL,
 		Selector:  random.NewSelector(),
 		Logger:    logger.DefaultLogger,
-		Broker:    broker.DefaultBroker,
 		Meter:     meter.DefaultMeter,
 		Tracer:    tracer.DefaultTracer,
 		Router:    router.DefaultRouter,
@@ -213,13 +220,6 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
-// Broker to be used for pub/sub
-func Broker(b broker.Broker) Option {
-	return func(o *Options) {
-		o.Broker = b
-	}
-}
-
 // Tracer to be used for tracing
 func Tracer(t tracer.Tracer) Option {
 	return func(o *Options) {