many fixes for lint and context.Context usage

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

api/handler/rpc/rpc.go

@@ -10,13 +10,13 @@ import (
 
 	jsonpatch "github.com/evanphx/json-patch/v5"
 	"github.com/oxtoacart/bpool"
+	jsonrpc "github.com/unistack-org/micro-codec-jsonrpc"
+	protorpc "github.com/unistack-org/micro-codec-protorpc"
 	"github.com/unistack-org/micro/v3/api"
 	"github.com/unistack-org/micro/v3/api/handler"
 	"github.com/unistack-org/micro/v3/api/internal/proto"
 	"github.com/unistack-org/micro/v3/client"
 	"github.com/unistack-org/micro/v3/codec"
-	"github.com/unistack-org/micro/v3/codec/jsonrpc"
-	"github.com/unistack-org/micro/v3/codec/protorpc"
 	"github.com/unistack-org/micro/v3/errors"
 	"github.com/unistack-org/micro/v3/logger"
 	"github.com/unistack-org/micro/v3/metadata"

api/router/options.go

@@ -1,6 +1,8 @@
 package router
 
 import (
+	"context"
+
 	"github.com/unistack-org/micro/v3/api/resolver"
 	"github.com/unistack-org/micro/v3/api/resolver/vpath"
 	"github.com/unistack-org/micro/v3/registry"
@@ -10,12 +12,14 @@ type Options struct {
 	Handler  string
 	Registry registry.Registry
 	Resolver resolver.Resolver
+	Context  context.Context
 }
 
 type Option func(o *Options)
 
 func NewOptions(opts ...Option) Options {
 	options := Options{
+		Context: context.Background(),
 		Handler: "meta",
 	}
 
@@ -32,18 +36,28 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
+// WithContext sets the context
+func WithContext(ctx context.Context) Option {
+	return func(o *Options) {
+		o.Context = ctx
+	}
+}
+
+// WithHandler sets the handler
 func WithHandler(h string) Option {
 	return func(o *Options) {
 		o.Handler = h
 	}
 }
 
+// WithRegistry sets the registry
 func WithRegistry(r registry.Registry) Option {
 	return func(o *Options) {
 		o.Registry = r
 	}
 }
 
+// WithResolver sets the resolver
 func WithResolver(r resolver.Resolver) Option {
 	return func(o *Options) {
 		o.Resolver = r

api/router/registry/registry.go

@@ -50,7 +50,7 @@ func (r *registryRouter) refresh() {
 	var attempts int
 
 	for {
-		services, err := r.opts.Registry.ListServices()
+		services, err := r.opts.Registry.ListServices(r.opts.Context)
 		if err != nil {
 			attempts++
 			if logger.V(logger.ErrorLevel) {
@@ -64,7 +64,7 @@ func (r *registryRouter) refresh() {
 
 		// for each service, get service and store endpoints
 		for _, s := range services {
-			service, err := r.opts.Registry.GetService(s.Name)
+			service, err := r.opts.Registry.GetService(r.opts.Context, s.Name)
 			if err != nil {
 				if logger.V(logger.ErrorLevel) {
 					logger.Errorf("unable to get service: %v", err)
@@ -92,7 +92,7 @@ func (r *registryRouter) process(res *registry.Result) {
 	}
 
 	// get entry from cache
-	service, err := r.opts.Registry.GetService(res.Service.Name)
+	service, err := r.opts.Registry.GetService(r.opts.Context, res.Service.Name)
 	if err != nil {
 		if logger.V(logger.ErrorLevel) {
 			logger.Errorf("unable to get %v service: %v", res.Service.Name, err)
@@ -230,7 +230,7 @@ func (r *registryRouter) watch() {
 		}
 
 		// watch for changes
-		w, err := r.opts.Registry.Watch()
+		w, err := r.opts.Registry.Watch(r.opts.Context)
 		if err != nil {
 			attempts++
 			if logger.V(logger.ErrorLevel) {
@@ -432,7 +432,7 @@ func (r *registryRouter) Route(req *http.Request) (*api.Service, error) {
 	name := rp.Name
 
 	// get service
-	services, err := r.opts.Registry.GetService(name, registry.GetDomain(rp.Domain))
+	services, err := r.opts.Registry.GetService(r.opts.Context, name, registry.GetDomain(rp.Domain))
 	if err != nil {
 		return nil, err
 	}

api/router/static/static.go

@@ -177,7 +177,7 @@ func (r *staticRouter) Endpoint(req *http.Request) (*api.Service, error) {
 	}
 
 	epf := strings.Split(ep.apiep.Name, ".")
-	services, err := r.opts.Registry.GetService(epf[0])
+	services, err := r.opts.Registry.GetService(r.opts.Context, epf[0])
 	if err != nil {
 		return nil, err
 	}

broker/options.go

@@ -190,7 +190,7 @@ func Secure(b bool) Option {
 	}
 }
 
-// Specify TLS Config
+// TLSConfig sets the TLS Config
 func TLSConfig(t *tls.Config) Option {
 	return func(o *Options) {
 		o.TLSConfig = t

cache/cache.go

@@ -1,24 +1,29 @@
 // Package cache is a caching interface
 package cache
 
+import "context"
+
 // Cache is an interface for caching
 type Cache interface {
 	// Initialise options
 	Init(...Option) error
 	// Get a value
-	Get(key string) (interface{}, error)
+	Get(ctx context.Context, key string) (interface{}, error)
 	// Set a value
-	Set(key string, val interface{}) error
+	Set(ctx context.Context, key string, val interface{}) error
 	// Delete a value
-	Delete(key string) error
+	Delete(ctx context.Context, key string) error
 	// Name of the implementation
 	String() string
 }
 
+// Options struct
 type Options struct {
-	Nodes []string
+	Nodes   []string
+	Context context.Context
 }
 
+// Option func
 type Option func(o *Options)
 
 // Nodes sets the nodes for the cache

client/client.go

@@ -9,6 +9,7 @@ import (
 )
 
 var (
+	// DefaultClient is the global default client
 	DefaultClient Client = NewClient()
 )
 

client/noop.go

@@ -4,9 +4,9 @@ import (
 	"context"
 
 	raw "github.com/unistack-org/micro-codec-bytes"
+	json "github.com/unistack-org/micro-codec-json"
 	"github.com/unistack-org/micro/v3/broker"
 	"github.com/unistack-org/micro/v3/codec"
-	"github.com/unistack-org/micro/v3/codec/json"
 	"github.com/unistack-org/micro/v3/errors"
 	"github.com/unistack-org/micro/v3/metadata"
 )

codec/codec.go

@@ -14,12 +14,14 @@ const (
 )
 
 var (
+	// ErrInvalidMessage returned when invalid messge passed to codec
 	ErrInvalidMessage = errors.New("invalid message")
 )
 
+// MessageType
 type MessageType int
 
-// Takes in a connection/buffer and returns a new Codec
+// NewCodec takes in a connection/buffer and returns a new Codec
 type NewCodec func(io.ReadWriteCloser) Codec
 
 // Codec encodes/decodes various types of messages used within go-micro.
@@ -34,11 +36,13 @@ type Codec interface {
 	String() string
 }
 
+// Reader interface
 type Reader interface {
 	ReadHeader(*Message, MessageType) error
 	ReadBody(interface{}) error
 }
 
+// Writer interface
 type Writer interface {
 	Write(*Message, interface{}) error
 }

codec/json/json.go

@@ -1,60 +0,0 @@
-// Package json provides a json codec
-package json
-
-import (
-	"encoding/json"
-	"io"
-	"io/ioutil"
-
-	"github.com/unistack-org/micro/v3/codec"
-	jsonpb "google.golang.org/protobuf/encoding/protojson"
-	"google.golang.org/protobuf/proto"
-)
-
-type Codec struct {
-	Conn    io.ReadWriteCloser
-	Encoder *json.Encoder
-	Decoder *json.Decoder
-}
-
-func (c *Codec) ReadHeader(m *codec.Message, t codec.MessageType) error {
-	return nil
-}
-
-func (c *Codec) ReadBody(b interface{}) error {
-	if b == nil {
-		return nil
-	}
-	switch m := b.(type) {
-	case proto.Message:
-		buf, err := ioutil.ReadAll(c.Conn)
-		if err != nil {
-			return err
-		}
-		return jsonpb.Unmarshal(buf, m)
-	}
-	return c.Decoder.Decode(b)
-}
-
-func (c *Codec) Write(m *codec.Message, b interface{}) error {
-	if b == nil {
-		return nil
-	}
-	return c.Encoder.Encode(b)
-}
-
-func (c *Codec) Close() error {
-	return c.Conn.Close()
-}
-
-func (c *Codec) String() string {
-	return "json"
-}
-
-func NewCodec(c io.ReadWriteCloser) codec.Codec {
-	return &Codec{
-		Conn:    c,
-		Decoder: json.NewDecoder(c),
-		Encoder: json.NewEncoder(c),
-	}
-}

codec/json/marshaler.go

@@ -1,45 +0,0 @@
-package json
-
-import (
-	"bytes"
-	"encoding/json"
-
-	oldjsonpb "github.com/golang/protobuf/jsonpb"
-	oldproto "github.com/golang/protobuf/proto"
-	"github.com/oxtoacart/bpool"
-	jsonpb "google.golang.org/protobuf/encoding/protojson"
-	"google.golang.org/protobuf/proto"
-)
-
-var jsonpbMarshaler = &jsonpb.MarshalOptions{}
-var oldjsonpbMarshaler = &oldjsonpb.Marshaler{}
-
-// create buffer pool with 16 instances each preallocated with 256 bytes
-var bufferPool = bpool.NewSizedBufferPool(16, 256)
-
-type Marshaler struct{}
-
-func (j Marshaler) Marshal(v interface{}) ([]byte, error) {
-	switch m := v.(type) {
-	case proto.Message:
-		return jsonpbMarshaler.Marshal(m)
-	case oldproto.Message:
-		buf, err := oldjsonpbMarshaler.MarshalToString(m)
-		return []byte(buf), err
-	}
-	return json.Marshal(v)
-}
-
-func (j Marshaler) Unmarshal(d []byte, v interface{}) error {
-	switch m := v.(type) {
-	case proto.Message:
-		return jsonpb.Unmarshal(d, m)
-	case oldproto.Message:
-		return oldjsonpb.Unmarshal(bytes.NewReader(d), m)
-	}
-	return json.Unmarshal(d, v)
-}
-
-func (j Marshaler) String() string {
-	return "json"
-}

codec/jsonrpc/client.go

@@ -1,97 +0,0 @@
-package jsonrpc
-
-import (
-	"encoding/json"
-	"fmt"
-	"io"
-	"sync"
-
-	"github.com/unistack-org/micro/v3/codec"
-)
-
-type clientCodec struct {
-	dec *json.Decoder // for reading JSON values
-	enc *json.Encoder // for writing JSON values
-	c   io.Closer
-
-	// temporary work space
-	req  clientRequest
-	resp clientResponse
-
-	sync.Mutex
-	pending map[interface{}]string
-}
-
-type clientRequest struct {
-	Method string         `json:"method"`
-	Params [1]interface{} `json:"params"`
-	ID     interface{}    `json:"id"`
-}
-
-type clientResponse struct {
-	ID     interface{}      `json:"id"`
-	Result *json.RawMessage `json:"result"`
-	Error  interface{}      `json:"error"`
-}
-
-func newClientCodec(conn io.ReadWriteCloser) *clientCodec {
-	return &clientCodec{
-		dec:     json.NewDecoder(conn),
-		enc:     json.NewEncoder(conn),
-		c:       conn,
-		pending: make(map[interface{}]string),
-	}
-}
-
-func (c *clientCodec) Write(m *codec.Message, b interface{}) error {
-	c.Lock()
-	c.pending[m.Id] = m.Method
-	c.Unlock()
-	c.req.Method = m.Method
-	c.req.Params[0] = b
-	c.req.ID = m.Id
-	return c.enc.Encode(&c.req)
-}
-
-func (r *clientResponse) reset() {
-	r.ID = 0
-	r.Result = nil
-	r.Error = nil
-}
-
-func (c *clientCodec) ReadHeader(m *codec.Message) error {
-	c.resp.reset()
-	if err := c.dec.Decode(&c.resp); err != nil {
-		return err
-	}
-
-	c.Lock()
-	m.Method = c.pending[c.resp.ID]
-	delete(c.pending, c.resp.ID)
-	c.Unlock()
-
-	m.Error = ""
-	m.Id = fmt.Sprintf("%v", c.resp.ID)
-	if c.resp.Error != nil {
-		x, ok := c.resp.Error.(string)
-		if !ok {
-			return fmt.Errorf("invalid error %v", c.resp.Error)
-		}
-		if x == "" {
-			x = "unspecified error"
-		}
-		m.Error = x
-	}
-	return nil
-}
-
-func (c *clientCodec) ReadBody(x interface{}) error {
-	if x == nil || c.resp.Result == nil {
-		return nil
-	}
-	return json.Unmarshal(*c.resp.Result, x)
-}
-
-func (c *clientCodec) Close() error {
-	return c.c.Close()
-}

codec/jsonrpc/jsonrpc.go

@@ -1,88 +0,0 @@
-// Package jsonrpc provides a json-rpc 1.0 codec
-package jsonrpc
-
-import (
-	"bytes"
-	"encoding/json"
-	"fmt"
-	"io"
-
-	"github.com/unistack-org/micro/v3/codec"
-)
-
-type jsonCodec struct {
-	buf *bytes.Buffer
-	mt  codec.MessageType
-	rwc io.ReadWriteCloser
-	c   *clientCodec
-	s   *serverCodec
-}
-
-func (j *jsonCodec) Close() error {
-	j.buf.Reset()
-	return j.rwc.Close()
-}
-
-func (j *jsonCodec) String() string {
-	return "json-rpc"
-}
-
-func (j *jsonCodec) Write(m *codec.Message, b interface{}) error {
-	switch m.Type {
-	case codec.Request:
-		return j.c.Write(m, b)
-	case codec.Response, codec.Error:
-		return j.s.Write(m, b)
-	case codec.Event:
-		data, err := json.Marshal(b)
-		if err != nil {
-			return err
-		}
-		_, err = j.rwc.Write(data)
-		return err
-	default:
-		return fmt.Errorf("Unrecognised message type: %v", m.Type)
-	}
-}
-
-func (j *jsonCodec) ReadHeader(m *codec.Message, mt codec.MessageType) error {
-	j.buf.Reset()
-	j.mt = mt
-
-	switch mt {
-	case codec.Request:
-		return j.s.ReadHeader(m)
-	case codec.Response:
-		return j.c.ReadHeader(m)
-	case codec.Event:
-		_, err := io.Copy(j.buf, j.rwc)
-		return err
-	default:
-		return fmt.Errorf("Unrecognised message type: %v", mt)
-	}
-}
-
-func (j *jsonCodec) ReadBody(b interface{}) error {
-	switch j.mt {
-	case codec.Request:
-		return j.s.ReadBody(b)
-	case codec.Response:
-		return j.c.ReadBody(b)
-	case codec.Event:
-		if b != nil {
-			return json.Unmarshal(j.buf.Bytes(), b)
-		}
-	default:
-		return fmt.Errorf("Unrecognised message type: %v", j.mt)
-	}
-	return nil
-}
-
-func NewCodec(rwc io.ReadWriteCloser) codec.Codec {
-	return &jsonCodec{
-		buf: bytes.NewBuffer(nil),
-		rwc: rwc,
-		c:   newClientCodec(rwc),
-		s:   newServerCodec(rwc),
-	}
-}

codec/jsonrpc/server.go

@@ -1,84 +0,0 @@
-package jsonrpc
-
-import (
-	"encoding/json"
-	"fmt"
-	"io"
-
-	"github.com/unistack-org/micro/v3/codec"
-)
-
-type serverCodec struct {
-	dec *json.Decoder // for reading JSON values
-	enc *json.Encoder // for writing JSON values
-	c   io.Closer
-
-	// temporary work space
-	req serverRequest
-}
-
-type serverRequest struct {
-	Method string           `json:"method"`
-	Params *json.RawMessage `json:"params"`
-	ID     interface{}      `json:"id"`
-}
-
-type serverResponse struct {
-	ID     interface{} `json:"id"`
-	Result interface{} `json:"result"`
-	Error  interface{} `json:"error"`
-}
-
-func newServerCodec(conn io.ReadWriteCloser) *serverCodec {
-	return &serverCodec{
-		dec: json.NewDecoder(conn),
-		enc: json.NewEncoder(conn),
-		c:   conn,
-	}
-}
-
-func (r *serverRequest) reset() {
-	r.Method = ""
-	if r.Params != nil {
-		*r.Params = (*r.Params)[0:0]
-	}
-	if r.ID != nil {
-		r.ID = nil
-	}
-}
-
-func (c *serverCodec) ReadHeader(m *codec.Message) error {
-	c.req.reset()
-	if err := c.dec.Decode(&c.req); err != nil {
-		return err
-	}
-	m.Method = c.req.Method
-	m.Id = fmt.Sprintf("%v", c.req.ID)
-	c.req.ID = nil
-	return nil
-}
-
-func (c *serverCodec) ReadBody(x interface{}) error {
-	if x == nil {
-		return nil
-	}
-	var params [1]interface{}
-	params[0] = x
-	return json.Unmarshal(*c.req.Params, &params)
-}
-
-func (c *serverCodec) Write(m *codec.Message, x interface{}) error {
-	var resp serverResponse
-	resp.ID = m.Id
-	resp.Result = x
-	if m.Error == "" {
-		resp.Error = nil
-	} else {
-		resp.Error = m.Error
-	}
-	return c.enc.Encode(resp)
-}
-
-func (c *serverCodec) Close() error {
-	return c.c.Close()
-}

codec/proto/marshaler.go

@@ -1,47 +0,0 @@
-package proto
-
-import (
-	"bytes"
-
-	"github.com/golang/protobuf/proto"
-	"github.com/oxtoacart/bpool"
-	"github.com/unistack-org/micro/v3/codec"
-)
-
-// create buffer pool with 16 instances each preallocated with 256 bytes
-var bufferPool = bpool.NewSizedBufferPool(16, 256)
-
-type Marshaler struct{}
-
-func (Marshaler) Marshal(v interface{}) ([]byte, error) {
-	pb, ok := v.(proto.Message)
-	if !ok {
-		return nil, codec.ErrInvalidMessage
-	}
-
-	// looks not good, but allows to reuse underlining bytes
-	buf := bufferPool.Get()
-	pbuf := proto.NewBuffer(buf.Bytes())
-	defer func() {
-		bufferPool.Put(bytes.NewBuffer(pbuf.Bytes()))
-	}()
-
-	if err := pbuf.Marshal(pb); err != nil {
-		return nil, err
-	}
-
-	return pbuf.Bytes(), nil
-}
-
-func (Marshaler) Unmarshal(data []byte, v interface{}) error {
-	pb, ok := v.(proto.Message)
-	if !ok {
-		return codec.ErrInvalidMessage
-	}
-
-	return proto.Unmarshal(data, pb)
-}
-
-func (Marshaler) String() string {
-	return "proto"
-}

codec/proto/message.go

@@ -1,37 +0,0 @@
-package proto
-
-type Message struct {
-	Data []byte
-}
-
-func (m *Message) MarshalJSON() ([]byte, error) {
-	return m.Data, nil
-}
-
-func (m *Message) UnmarshalJSON(data []byte) error {
-	m.Data = data
-	return nil
-}
-
-func (m *Message) ProtoMessage() {}
-
-func (m *Message) Reset() {
-	*m = Message{}
-}
-
-func (m *Message) String() string {
-	return string(m.Data)
-}
-
-func (m *Message) Marshal() ([]byte, error) {
-	return m.Data, nil
-}
-
-func (m *Message) Unmarshal(data []byte) error {
-	m.Data = data
-	return nil
-}
-
-func NewMessage(data []byte) *Message {
-	return &Message{data}
-}

codec/proto/proto.go

@@ -1,64 +0,0 @@
-// Package proto provides a proto codec
-package proto
-
-import (
-	"io"
-	"io/ioutil"
-
-	"github.com/unistack-org/micro/v3/codec"
-	"google.golang.org/protobuf/proto"
-)
-
-type Codec struct {
-	Conn io.ReadWriteCloser
-}
-
-func (c *Codec) ReadHeader(m *codec.Message, t codec.MessageType) error {
-	return nil
-}
-
-func (c *Codec) ReadBody(b interface{}) error {
-	if b == nil {
-		return nil
-	}
-	buf, err := ioutil.ReadAll(c.Conn)
-	if err != nil {
-		return err
-	}
-	m, ok := b.(proto.Message)
-	if !ok {
-		return codec.ErrInvalidMessage
-	}
-	return proto.Unmarshal(buf, m)
-}
-
-func (c *Codec) Write(m *codec.Message, b interface{}) error {
-	if b == nil {
-		// Nothing to write
-		return nil
-	}
-	p, ok := b.(proto.Message)
-	if !ok {
-		return codec.ErrInvalidMessage
-	}
-	buf, err := proto.Marshal(p)
-	if err != nil {
-		return err
-	}
-	_, err = c.Conn.Write(buf)
-	return err
-}
-
-func (c *Codec) Close() error {
-	return c.Conn.Close()
-}
-
-func (c *Codec) String() string {
-	return "proto"
-}
-
-func NewCodec(c io.ReadWriteCloser) codec.Codec {
-	return &Codec{
-		Conn: c,
-	}
-}

codec/protorpc/envelope.pb.go

@@ -1,238 +0,0 @@
-// Code generated by protoc-gen-go. DO NOT EDIT.
-// versions:
-// 	protoc-gen-go v1.25.0
-// 	protoc        v3.6.1
-// source: codec/protorpc/envelope.proto
-
-package protorpc
-
-import (
-	proto "github.com/golang/protobuf/proto"
-	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
-	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
-	reflect "reflect"
-	sync "sync"
-)
-
-const (
-	// Verify that this generated code is sufficiently up-to-date.
-	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
-	// Verify that runtime/protoimpl is sufficiently up-to-date.
-	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
-)
-
-// This is a compile-time assertion that a sufficiently up-to-date version
-// of the legacy proto package is being used.
-const _ = proto.ProtoPackageIsVersion4
-
-type Request struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	ServiceMethod string `protobuf:"bytes,1,opt,name=service_method,json=serviceMethod,proto3" json:"service_method,omitempty"`
-	Seq           uint64 `protobuf:"fixed64,2,opt,name=seq,proto3" json:"seq,omitempty"`
-}
-
-func (x *Request) Reset() {
-	*x = Request{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_codec_protorpc_envelope_proto_msgTypes[0]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *Request) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*Request) ProtoMessage() {}
-
-func (x *Request) ProtoReflect() protoreflect.Message {
-	mi := &file_codec_protorpc_envelope_proto_msgTypes[0]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use Request.ProtoReflect.Descriptor instead.
-func (*Request) Descriptor() ([]byte, []int) {
-	return file_codec_protorpc_envelope_proto_rawDescGZIP(), []int{0}
-}
-
-func (x *Request) GetServiceMethod() string {
-	if x != nil {
-		return x.ServiceMethod
-	}
-	return ""
-}
-
-func (x *Request) GetSeq() uint64 {
-	if x != nil {
-		return x.Seq
-	}
-	return 0
-}
-
-type Response struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	ServiceMethod string `protobuf:"bytes,1,opt,name=service_method,json=serviceMethod,proto3" json:"service_method,omitempty"`
-	Seq           uint64 `protobuf:"fixed64,2,opt,name=seq,proto3" json:"seq,omitempty"`
-	Error         string `protobuf:"bytes,3,opt,name=error,proto3" json:"error,omitempty"`
-}
-
-func (x *Response) Reset() {
-	*x = Response{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_codec_protorpc_envelope_proto_msgTypes[1]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *Response) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*Response) ProtoMessage() {}
-
-func (x *Response) ProtoReflect() protoreflect.Message {
-	mi := &file_codec_protorpc_envelope_proto_msgTypes[1]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use Response.ProtoReflect.Descriptor instead.
-func (*Response) Descriptor() ([]byte, []int) {
-	return file_codec_protorpc_envelope_proto_rawDescGZIP(), []int{1}
-}
-
-func (x *Response) GetServiceMethod() string {
-	if x != nil {
-		return x.ServiceMethod
-	}
-	return ""
-}
-
-func (x *Response) GetSeq() uint64 {
-	if x != nil {
-		return x.Seq
-	}
-	return 0
-}
-
-func (x *Response) GetError() string {
-	if x != nil {
-		return x.Error
-	}
-	return ""
-}
-
-var File_codec_protorpc_envelope_proto protoreflect.FileDescriptor
-
-var file_codec_protorpc_envelope_proto_rawDesc = []byte{
-	0x0a, 0x1d, 0x63, 0x6f, 0x64, 0x65, 0x63, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x72, 0x70, 0x63,
-	0x2f, 0x65, 0x6e, 0x76, 0x65, 0x6c, 0x6f, 0x70, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12,
-	0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x72, 0x70, 0x63, 0x22, 0x42, 0x0a, 0x07, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x12, 0x25, 0x0a, 0x0e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x5f,
-	0x6d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x73, 0x65,
-	0x72, 0x76, 0x69, 0x63, 0x65, 0x4d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x73,
-	0x65, 0x71, 0x18, 0x02, 0x20, 0x01, 0x28, 0x06, 0x52, 0x03, 0x73, 0x65, 0x71, 0x22, 0x59, 0x0a,
-	0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x25, 0x0a, 0x0e, 0x73, 0x65, 0x72,
-	0x76, 0x69, 0x63, 0x65, 0x5f, 0x6d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x0d, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x4d, 0x65, 0x74, 0x68, 0x6f, 0x64,
-	0x12, 0x10, 0x0a, 0x03, 0x73, 0x65, 0x71, 0x18, 0x02, 0x20, 0x01, 0x28, 0x06, 0x52, 0x03, 0x73,
-	0x65, 0x71, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
-}
-
-var (
-	file_codec_protorpc_envelope_proto_rawDescOnce sync.Once
-	file_codec_protorpc_envelope_proto_rawDescData = file_codec_protorpc_envelope_proto_rawDesc
-)
-
-func file_codec_protorpc_envelope_proto_rawDescGZIP() []byte {
-	file_codec_protorpc_envelope_proto_rawDescOnce.Do(func() {
-		file_codec_protorpc_envelope_proto_rawDescData = protoimpl.X.CompressGZIP(file_codec_protorpc_envelope_proto_rawDescData)
-	})
-	return file_codec_protorpc_envelope_proto_rawDescData
-}
-
-var file_codec_protorpc_envelope_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
-var file_codec_protorpc_envelope_proto_goTypes = []interface{}{
-	(*Request)(nil),  // 0: protorpc.Request
-	(*Response)(nil), // 1: protorpc.Response
-}
-var file_codec_protorpc_envelope_proto_depIdxs = []int32{
-	0, // [0:0] is the sub-list for method output_type
-	0, // [0:0] is the sub-list for method input_type
-	0, // [0:0] is the sub-list for extension type_name
-	0, // [0:0] is the sub-list for extension extendee
-	0, // [0:0] is the sub-list for field type_name
-}
-
-func init() { file_codec_protorpc_envelope_proto_init() }
-func file_codec_protorpc_envelope_proto_init() {
-	if File_codec_protorpc_envelope_proto != nil {
-		return
-	}
-	if !protoimpl.UnsafeEnabled {
-		file_codec_protorpc_envelope_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Request); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_codec_protorpc_envelope_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Response); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-	}
-	type x struct{}
-	out := protoimpl.TypeBuilder{
-		File: protoimpl.DescBuilder{
-			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
-			RawDescriptor: file_codec_protorpc_envelope_proto_rawDesc,
-			NumEnums:      0,
-			NumMessages:   2,
-			NumExtensions: 0,
-			NumServices:   0,
-		},
-		GoTypes:           file_codec_protorpc_envelope_proto_goTypes,
-		DependencyIndexes: file_codec_protorpc_envelope_proto_depIdxs,
-		MessageInfos:      file_codec_protorpc_envelope_proto_msgTypes,
-	}.Build()
-	File_codec_protorpc_envelope_proto = out.File
-	file_codec_protorpc_envelope_proto_rawDesc = nil
-	file_codec_protorpc_envelope_proto_goTypes = nil
-	file_codec_protorpc_envelope_proto_depIdxs = nil
-}

codec/protorpc/envelope.pb.micro.go

@@ -1,21 +0,0 @@
-// Code generated by protoc-gen-micro. DO NOT EDIT.
-// source: codec/protorpc/envelope.proto
-
-package protorpc
-
-import (
-	fmt "fmt"
-	proto "github.com/golang/protobuf/proto"
-	math "math"
-)
-
-// Reference imports to suppress errors if they are not otherwise used.
-var _ = proto.Marshal
-var _ = fmt.Errorf
-var _ = math.Inf
-
-// This is a compile-time assertion to ensure that this generated file
-// is compatible with the proto package it is being compiled against.
-// A compilation error at this line likely means your copy of the
-// proto package needs to be updated.
-const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package

codec/protorpc/envelope.proto

@@ -1,14 +0,0 @@
-syntax = "proto3";
-
-package protorpc;
-
-message Request {
-	string service_method = 1;
-	fixed64 seq = 2;
-}
-
-message Response {
-	string service_method = 1;
-	fixed64 seq = 2;
-	string error = 3;
-}

codec/protorpc/netstring.go

@@ -1,36 +0,0 @@
-package protorpc
-
-import (
-	"encoding/binary"
-	"io"
-)
-
-// WriteNetString writes data to a big-endian netstring on a Writer.
-// Size is always a 32-bit unsigned int.
-func WriteNetString(w io.Writer, data []byte) (written int, err error) {
-	size := make([]byte, 4)
-	binary.BigEndian.PutUint32(size, uint32(len(data)))
-	if written, err = w.Write(size); err != nil {
-		return
-	}
-	return w.Write(data)
-}
-
-// ReadNetString reads data from a big-endian netstring.
-func ReadNetString(r io.Reader) (data []byte, err error) {
-	sizeBuf := make([]byte, 4)
-	_, err = r.Read(sizeBuf)
-	if err != nil {
-		return nil, err
-	}
-	size := binary.BigEndian.Uint32(sizeBuf)
-	if size == 0 {
-		return nil, nil
-	}
-	data = make([]byte, size)
-	_, err = r.Read(data)
-	if err != nil {
-		return nil, err
-	}
-	return
-}

codec/protorpc/protorpc.go

@@ -1,186 +0,0 @@
-// Protorpc provides a net/rpc proto-rpc codec. See envelope.proto for the format.
-package protorpc
-
-import (
-	"bytes"
-	"fmt"
-	"io"
-	"strconv"
-	"sync"
-
-	"github.com/golang/protobuf/proto"
-	"github.com/unistack-org/micro/v3/codec"
-)
-
-type flusher interface {
-	Flush() error
-}
-
-type protoCodec struct {
-	sync.Mutex
-	rwc io.ReadWriteCloser
-	mt  codec.MessageType
-	buf *bytes.Buffer
-}
-
-func (c *protoCodec) Close() error {
-	c.buf.Reset()
-	return c.rwc.Close()
-}
-
-func (c *protoCodec) String() string {
-	return "proto-rpc"
-}
-
-func id(id string) uint64 {
-	p, err := strconv.ParseInt(id, 10, 64)
-	if err != nil {
-		p = 0
-	}
-	i := uint64(p)
-	return i
-}
-
-func (c *protoCodec) Write(m *codec.Message, b interface{}) error {
-	switch m.Type {
-	case codec.Request:
-		c.Lock()
-		defer c.Unlock()
-		// This is protobuf, of course we copy it.
-		pbr := &Request{ServiceMethod: m.Method, Seq: id(m.Id)}
-		data, err := proto.Marshal(pbr)
-		if err != nil {
-			return err
-		}
-		_, err = WriteNetString(c.rwc, data)
-		if err != nil {
-			return err
-		}
-		// dont trust or incoming message
-		m, ok := b.(proto.Message)
-		if !ok {
-			return codec.ErrInvalidMessage
-		}
-		data, err = proto.Marshal(m)
-		if err != nil {
-			return err
-		}
-		_, err = WriteNetString(c.rwc, data)
-		if err != nil {
-			return err
-		}
-		if flusher, ok := c.rwc.(flusher); ok {
-			if err = flusher.Flush(); err != nil {
-				return err
-			}
-		}
-	case codec.Response, codec.Error:
-		c.Lock()
-		defer c.Unlock()
-		rtmp := &Response{ServiceMethod: m.Method, Seq: id(m.Id), Error: m.Error}
-		data, err := proto.Marshal(rtmp)
-		if err != nil {
-			return err
-		}
-		_, err = WriteNetString(c.rwc, data)
-		if err != nil {
-			return err
-		}
-		if pb, ok := b.(proto.Message); ok {
-			data, err = proto.Marshal(pb)
-			if err != nil {
-				return err
-			}
-		} else {
-			data = nil
-		}
-		_, err = WriteNetString(c.rwc, data)
-		if err != nil {
-			return err
-		}
-		if flusher, ok := c.rwc.(flusher); ok {
-			if err = flusher.Flush(); err != nil {
-				return err
-			}
-		}
-	case codec.Event:
-		m, ok := b.(proto.Message)
-		if !ok {
-			return codec.ErrInvalidMessage
-		}
-		data, err := proto.Marshal(m)
-		if err != nil {
-			return err
-		}
-		c.rwc.Write(data)
-	default:
-		return fmt.Errorf("Unrecognised message type: %v", m.Type)
-	}
-	return nil
-}
-
-func (c *protoCodec) ReadHeader(m *codec.Message, mt codec.MessageType) error {
-	c.buf.Reset()
-	c.mt = mt
-
-	switch mt {
-	case codec.Request:
-		data, err := ReadNetString(c.rwc)
-		if err != nil {
-			return err
-		}
-		rtmp := new(Request)
-		err = proto.Unmarshal(data, rtmp)
-		if err != nil {
-			return err
-		}
-		m.Method = rtmp.GetServiceMethod()
-		m.Id = fmt.Sprintf("%d", rtmp.GetSeq())
-	case codec.Response:
-		data, err := ReadNetString(c.rwc)
-		if err != nil {
-			return err
-		}
-		rtmp := new(Response)
-		err = proto.Unmarshal(data, rtmp)
-		if err != nil {
-			return err
-		}
-		m.Method = rtmp.GetServiceMethod()
-		m.Id = fmt.Sprintf("%d", rtmp.GetSeq())
-		m.Error = rtmp.GetError()
-	case codec.Event:
-		_, err := io.Copy(c.buf, c.rwc)
-		return err
-	default:
-		return fmt.Errorf("Unrecognised message type: %v", mt)
-	}
-	return nil
-}
-
-func (c *protoCodec) ReadBody(b interface{}) error {
-	var data []byte
-	switch c.mt {
-	case codec.Request, codec.Response:
-		var err error
-		data, err = ReadNetString(c.rwc)
-		if err != nil {
-			return err
-		}
-	case codec.Event:
-		data = c.buf.Bytes()
-	default:
-		return fmt.Errorf("Unrecognised message type: %v", c.mt)
-	}
-	if b != nil {
-		return proto.Unmarshal(data, b.(proto.Message))
-	}
-	return nil
-}
-
-func NewCodec(rwc io.ReadWriteCloser) codec.Codec {
-	return &protoCodec{
-		buf: bytes.NewBuffer(nil),
-		rwc: rwc,
-	}
-}

codec/text/text.go

@@ -1,80 +0,0 @@
-// Package text reads any text/* content-type
-package text
-
-import (
-	"fmt"
-	"io"
-	"io/ioutil"
-
-	"github.com/unistack-org/micro/v3/codec"
-)
-
-type Codec struct {
-	Conn io.ReadWriteCloser
-}
-
-// Frame gives us the ability to define raw data to send over the pipes
-type Frame struct {
-	Data []byte
-}
-
-func (c *Codec) ReadHeader(m *codec.Message, t codec.MessageType) error {
-	return nil
-}
-
-func (c *Codec) ReadBody(b interface{}) error {
-	// read bytes
-	buf, err := ioutil.ReadAll(c.Conn)
-	if err != nil {
-		return err
-	}
-
-	switch v := b.(type) {
-	case *string:
-		*v = string(buf)
-	case *[]byte:
-		*v = buf
-	case *Frame:
-		v.Data = buf
-	default:
-		return fmt.Errorf("failed to read body: %v is not type of *[]byte", b)
-	}
-
-	return nil
-}
-
-func (c *Codec) Write(m *codec.Message, b interface{}) error {
-	var v []byte
-	switch ve := b.(type) {
-	case nil:
-		return nil
-	case *Frame:
-		v = ve.Data
-	case *[]byte:
-		v = *ve
-	case *string:
-		v = []byte(*ve)
-	case string:
-		v = []byte(ve)
-	case []byte:
-		v = ve
-	default:
-		return fmt.Errorf("failed to write: %v is not type of *[]byte or []byte", b)
-	}
-	_, err := c.Conn.Write(v)
-	return err
-}
-
-func (c *Codec) Close() error {
-	return c.Conn.Close()
-}
-
-func (c *Codec) String() string {
-	return "text"
-}
-
-func NewCodec(c io.ReadWriteCloser) codec.Codec {
-	return &Codec{
-		Conn: c,
-	}
-}

events/events.go

@@ -2,6 +2,7 @@
 package events
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"time"
@@ -16,14 +17,14 @@ var (
 
 // Stream of events
 type Stream interface {
-	Publish(topic string, msg interface{}, opts ...PublishOption) error
+	Publish(ctx context.Context, topic string, msg interface{}, opts ...PublishOption) error
-	Subscribe(topic string, opts ...SubscribeOption) (<-chan Event, error)
+	Subscribe(ctx context.Context, topic string, opts ...SubscribeOption) (<-chan Event, error)
 }
 
 // Store of events
 type Store interface {
-	Read(opts ...ReadOption) ([]*Event, error)
+	Read(ctx context.Context, opts ...ReadOption) ([]*Event, error)
-	Write(event *Event, opts ...WriteOption) error
+	Write(ctx context.Context, event *Event, opts ...WriteOption) error
 }
 
 // Event is the object returned by the broker when you subscribe to a topic

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/go-acme/lego/v3 v3.4.0
 	github.com/gobwas/httphead v0.0.0-20180130184737-2c6c146eadee
 	github.com/gobwas/ws v1.0.3
-	github.com/golang/protobuf v1.4.2
+	github.com/golang/protobuf v1.4.3
 	github.com/google/uuid v1.1.2
 	github.com/hashicorp/hcl v1.0.0
 	github.com/micro/cli/v2 v2.1.2
@@ -21,6 +21,10 @@ require (
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/stretchr/testify v1.5.1
 	github.com/unistack-org/micro-codec-bytes v0.0.0-20200828083432-4e49e953d844
+	github.com/unistack-org/micro-codec-json v0.0.0-20201102222734-a29c895ec05c
+	github.com/unistack-org/micro-codec-jsonrpc v0.0.0-20201102222451-ff6a69988bcd
+	github.com/unistack-org/micro-codec-proto v0.0.0-20201102222202-769c2d6a4b92
+	github.com/unistack-org/micro-codec-protorpc v0.0.0-20201102222610-3a343898c077
 	github.com/unistack-org/micro-config-cmd v0.0.0-20201028144621-5a55f1aad70a
 	golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a
 	golang.org/x/net v0.0.0-20200904194848-62affa334b73

go.sum

@@ -121,6 +121,8 @@ github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvq
 github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
 github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/golang/protobuf v1.4.3 h1:JjCZWpVbqXDqFVmTfYWEVTMIYrL/NPdPSCHPJ0T/raM=
+github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0 h1:0udJVsspx3VBr5FwtLhQQtuAsVc79tTq0ocGIPAU6qo=
@@ -281,8 +283,18 @@ github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5
 github.com/timewasted/linode v0.0.0-20160829202747-37e84520dcf7/go.mod h1:imsgLplxEC/etjIhdr3dNzV3JeT27LbVu5pYWm0JCBY=
 github.com/transip/gotransip v0.0.0-20190812104329-6d8d9179b66f/go.mod h1:i0f4R4o2HM0m3DZYQWsj6/MEowD57VzoH0v3d7igeFY=
 github.com/uber-go/atomic v1.3.2/go.mod h1:/Ct5t2lcmbJ4OSe/waGBoaVvVqtO0bmtfVNex1PFV8g=
+github.com/unistack-org/micro-codec-bytes v0.0.0-20200827104921-3616a69473a6/go.mod h1:g5sOI8TWgGZiVHe8zoUPdtz7+0oLnqTnfBoai6Qb7jE=
 github.com/unistack-org/micro-codec-bytes v0.0.0-20200828083432-4e49e953d844 h1:5b1yuSllbsMm/9fUIlIXSr8DbsKT/sAKSCgOx6+SAfI=
 github.com/unistack-org/micro-codec-bytes v0.0.0-20200828083432-4e49e953d844/go.mod h1:g5sOI8TWgGZiVHe8zoUPdtz7+0oLnqTnfBoai6Qb7jE=
+github.com/unistack-org/micro-codec-json v0.0.0-20201102222734-a29c895ec05c h1:RtcNaK8rQSl7xAoy1W437dvZLCVjSC6e4JcolepSQs0=
+github.com/unistack-org/micro-codec-json v0.0.0-20201102222734-a29c895ec05c/go.mod h1:dG5aUyhBv+ebOl/UFW2Aj2GTfVxxXWi6AcynpePOAhQ=
+github.com/unistack-org/micro-codec-jsonrpc v0.0.0-20201102222451-ff6a69988bcd h1:qXSiEfVnCgrwTHYvAnEPSHEai3+5EUH9ZYovLpxGDwg=
+github.com/unistack-org/micro-codec-jsonrpc v0.0.0-20201102222451-ff6a69988bcd/go.mod h1:PFyvkGhavl+3tEPgOaLAhoJJX4/webVGW59BSOXDfNM=
+github.com/unistack-org/micro-codec-proto v0.0.0-20201102222202-769c2d6a4b92 h1:1rPDBu7Nwo3ZL6r6H5rj7qNchHSdBF4zcewAeTUEMC4=
+github.com/unistack-org/micro-codec-proto v0.0.0-20201102222202-769c2d6a4b92/go.mod h1:31JMo683bBQ+uN9YufpUU6ESHphyx3DFmTXEnjpJV9Y=
+github.com/unistack-org/micro-codec-protorpc v0.0.0-20201102222610-3a343898c077 h1:uK7owL8TPSwoQiDM1V/0swmgCEepSQKXoi8GEnGxtlU=
+github.com/unistack-org/micro-codec-protorpc v0.0.0-20201102222610-3a343898c077/go.mod h1:Ct4uAVZaDEyBZj9Q0poDkbzu6zKXUCcSqJkv/MWPpeI=
+github.com/unistack-org/micro-config-cmd v0.0.0-20200828075439-d859b9d7265b/go.mod h1:6pm1cadbwsFcEW1ZbV5Fp0i3goR3TNfROMNSPih3I8k=
 github.com/unistack-org/micro-config-cmd v0.0.0-20200909210346-ec89783dc46c h1:GbcjxyOyA9tnNoe4FcnzzLDa8JwEBnQKN/7Bhd8t47I=
 github.com/unistack-org/micro-config-cmd v0.0.0-20200909210346-ec89783dc46c/go.mod h1:6pm1cadbwsFcEW1ZbV5Fp0i3goR3TNfROMNSPih3I8k=
 github.com/unistack-org/micro-config-cmd v0.0.0-20200909210755-6e7e85eeab34 h1:VHc98t4SoiCF/jbkFu2e/j+IyJ/+MFQ1T+INNL7LubU=
@@ -292,6 +304,7 @@ github.com/unistack-org/micro-config-cmd v0.0.0-20200920140133-0853deb2e5dc/go.m
 github.com/unistack-org/micro-config-cmd v0.0.0-20201028144621-5a55f1aad70a h1:VjlqP1qZkjC0Chmx5MKFPIbtSCigeICFDf8vaLZGh9o=
 github.com/unistack-org/micro-config-cmd v0.0.0-20201028144621-5a55f1aad70a/go.mod h1:MzMg+qh1wORZwYtg5AVgFkNFrXVVbdPKW7s/Is+A994=
 github.com/unistack-org/micro/v3 v3.0.0-20200827083227-aa99378adc6e/go.mod h1:rPQbnry3nboAnMczj8B1Gzlcyv/HYoMZLgd3/3nttJ4=
+github.com/unistack-org/micro/v3 v3.0.0-gamma/go.mod h1:iEtpu3wTYCRs3pQ3VsFEO7JBO4lOMpkOwMyrpZyIDPo=
 github.com/unistack-org/micro/v3 v3.0.0-gamma.0.20200909210629-caec730248b1/go.mod h1:mmqHR9WelHUXqg2mELjsQ+FJHcWs6mNmXg+wEYO2T3c=
 github.com/unistack-org/micro/v3 v3.0.0-gamma.0.20200920135754-1cbd1d2bad83/go.mod h1:HUzMG4Mcy97958VxWTg8zuazZgwQ/aoLZ8wtBVONwRE=
 github.com/unistack-org/micro/v3 v3.0.0-gamma.0.20200922103357-4c4fa00a5d94/go.mod h1:aL+8VhSXpx0SuEeXPOWUo5BgS7kyvWYobeXFay90UUM=

network/transport/noop.go

@@ -1,5 +1,7 @@
 package transport
 
+import "context"
+
 type noopTransport struct {
 	opts Options
 }
@@ -19,12 +21,12 @@ func (t *noopTransport) Options() Options {
 	return t.opts
 }
 
-func (t *noopTransport) Dial(addr string, opts ...DialOption) (Client, error) {
+func (t *noopTransport) Dial(ctx context.Context, addr string, opts ...DialOption) (Client, error) {
 	options := NewDialOptions(opts...)
 	return &noopClient{opts: options}, nil
 }
 
-func (t *noopTransport) Listen(addr string, opts ...ListenOption) (Listener, error) {
+func (t *noopTransport) Listen(ctx context.Context, addr string, opts ...ListenOption) (Listener, error) {
 	options := NewListenOptions(opts...)
 	return &noopListener{opts: options}, nil
 }

network/transport/options.go

@@ -31,6 +31,7 @@ type Options struct {
 	Context context.Context
 }
 
+// NewOptions returns new options
 func NewOptions(opts ...Option) Options {
 	options := Options{
 		Logger:  logger.DefaultLogger,
@@ -44,6 +45,7 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
+// DialOptions struct
 type DialOptions struct {
 	// Tells the transport this is a streaming connection with
 	// multiple calls to send/recv and that send may not even be called
@@ -59,6 +61,7 @@ type DialOptions struct {
 	Context context.Context
 }
 
+// NewDialOptions returns new DialOptions
 func NewDialOptions(opts ...DialOption) DialOptions {
 	options := DialOptions{
 		Context: context.Background(),
@@ -71,6 +74,7 @@ func NewDialOptions(opts ...DialOption) DialOptions {
 	return options
 }
 
+// ListenOptions struct
 type ListenOptions struct {
 	// TODO: add tls options when listening
 	// Currently set in global options
@@ -80,6 +84,7 @@ type ListenOptions struct {
 	Context context.Context
 }
 
+// NewListenOptions returns new ListenOptions
 func NewListenOptions(opts ...ListenOption) ListenOptions {
 	options := ListenOptions{
 		Context: context.Background(),
@@ -106,6 +111,7 @@ func Logger(l logger.Logger) Option {
 	}
 }
 
+// Context sets the context
 func Context(ctx context.Context) Option {
 	return func(o *Options) {
 		o.Context = ctx
@@ -142,14 +148,14 @@ func TLSConfig(t *tls.Config) Option {
 	}
 }
 
-// Indicates whether this is a streaming connection
+// WithStream indicates whether this is a streaming connection
 func WithStream() DialOption {
 	return func(o *DialOptions) {
 		o.Stream = true
 	}
 }
 
-// Timeout used when dialling the remote side
+// WithTimeout used when dialling the remote side
 func WithTimeout(d time.Duration) DialOption {
 	return func(o *DialOptions) {
 		o.Timeout = d

network/transport/transport.go

@@ -2,6 +2,7 @@
 package transport
 
 import (
+	"context"
 	"time"
 )
 
@@ -16,8 +17,8 @@ var (
 type Transport interface {
 	Init(...Option) error
 	Options() Options
-	Dial(addr string, opts ...DialOption) (Client, error)
+	Dial(ctx context.Context, addr string, opts ...DialOption) (Client, error)
-	Listen(addr string, opts ...ListenOption) (Listener, error)
+	Listen(ctx context.Context, addr string, opts ...ListenOption) (Listener, error)
 	String() string
 }
 

network/tunnel/broker/broker.go

@@ -50,17 +50,17 @@ func (t *tunBroker) Address() string {
 }
 
 func (t *tunBroker) Connect(ctx context.Context) error {
-	return t.tunnel.Connect()
+	return t.tunnel.Connect(ctx)
 }
 
 func (t *tunBroker) Disconnect(ctx context.Context) error {
-	return t.tunnel.Close()
+	return t.tunnel.Close(ctx)
 }
 
 func (t *tunBroker) Publish(ctx context.Context, topic string, m *broker.Message, opts ...broker.PublishOption) error {
 	// TODO: this is probably inefficient, we might want to just maintain an open connection
 	// it may be easier to add broadcast to the tunnel
-	c, err := t.tunnel.Dial(topic, tunnel.DialMode(tunnel.Multicast))
+	c, err := t.tunnel.Dial(ctx, topic, tunnel.DialMode(tunnel.Multicast))
 	if err != nil {
 		return err
 	}
@@ -73,7 +73,7 @@ func (t *tunBroker) Publish(ctx context.Context, topic string, m *broker.Message
 }
 
 func (t *tunBroker) Subscribe(ctx context.Context, topic string, h broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
-	l, err := t.tunnel.Listen(topic, tunnel.ListenMode(tunnel.Multicast))
+	l, err := t.tunnel.Listen(ctx, topic, tunnel.ListenMode(tunnel.Multicast))
 	if err != nil {
 		return nil, err
 	}

network/tunnel/transport/transport.go

@@ -26,7 +26,7 @@ func (t *tunTransport) Init(opts ...transport.Option) error {
 
 	// close the existing tunnel
 	if t.tunnel != nil {
-		t.tunnel.Close()
+		t.tunnel.Close(context.TODO())
 	}
 
 	// get the tunnel
@@ -47,12 +47,12 @@ func (t *tunTransport) Init(opts ...transport.Option) error {
 	return nil
 }
 
-func (t *tunTransport) Dial(addr string, opts ...transport.DialOption) (transport.Client, error) {
+func (t *tunTransport) Dial(ctx context.Context, addr string, opts ...transport.DialOption) (transport.Client, error) {
-	if err := t.tunnel.Connect(); err != nil {
+	if err := t.tunnel.Connect(ctx); err != nil {
 		return nil, err
 	}
 
-	c, err := t.tunnel.Dial(addr)
+	c, err := t.tunnel.Dial(ctx, addr)
 	if err != nil {
 		return nil, err
 	}
@@ -60,12 +60,12 @@ func (t *tunTransport) Dial(addr string, opts ...transport.DialOption) (transpor
 	return c, nil
 }
 
-func (t *tunTransport) Listen(addr string, opts ...transport.ListenOption) (transport.Listener, error) {
+func (t *tunTransport) Listen(ctx context.Context, addr string, opts ...transport.ListenOption) (transport.Listener, error) {
-	if err := t.tunnel.Connect(); err != nil {
+	if err := t.tunnel.Connect(ctx); err != nil {
 		return nil, err
 	}
 
-	l, err := t.tunnel.Listen(addr)
+	l, err := t.tunnel.Listen(ctx, addr)
 	if err != nil {
 		return nil, err
 	}

network/tunnel/tunnel.go

@@ -2,6 +2,7 @@
 package tunnel
 
 import (
+	"context"
 	"errors"
 	"time"
 
@@ -55,15 +56,15 @@ type Tunnel interface {
 	// Address returns the address the tunnel is listening on
 	Address() string
 	// Connect connects the tunnel
-	Connect() error
+	Connect(ctx context.Context) error
 	// Close closes the tunnel
-	Close() error
+	Close(ctx context.Context) error
 	// Links returns all the links the tunnel is connected to
 	Links() []Link
 	// Dial allows a client to connect to a channel
-	Dial(channel string, opts ...DialOption) (Session, error)
+	Dial(ctx context.Context, channel string, opts ...DialOption) (Session, error)
 	// Listen allows to accept connections on a channel
-	Listen(channel string, opts ...ListenOption) (Listener, error)
+	Listen(ctx context.Context, channel string, opts ...ListenOption) (Listener, error)
 	// String returns the name of the tunnel implementation
 	String() string
 }

options.go

@@ -91,6 +91,7 @@ func Broker(b broker.Broker) Option {
 	}
 }
 
+// Cmd to be used for service
 func Cmd(c cmd.Cmd) Option {
 	return func(o *Options) {
 		o.Cmd = c

registry/noop.go

@@ -33,27 +33,27 @@ func (n *noopRegistry) Disconnect(ctx context.Context) error {
 }
 
 // Register registers service
-func (n *noopRegistry) Register(*Service, ...RegisterOption) error {
+func (n *noopRegistry) Register(ctx context.Context, svc *Service, opts ...RegisterOption) error {
 	return nil
 }
 
 // Deregister deregisters service
-func (n *noopRegistry) Deregister(*Service, ...DeregisterOption) error {
+func (n *noopRegistry) Deregister(ctx context.Context, svc *Service, opts ...DeregisterOption) error {
 	return nil
 }
 
 // GetService returns servive info
-func (n *noopRegistry) GetService(string, ...GetOption) ([]*Service, error) {
+func (n *noopRegistry) GetService(ctx context.Context, name string, opts ...GetOption) ([]*Service, error) {
 	return []*Service{}, nil
 }
 
 // ListServices listing services
-func (n *noopRegistry) ListServices(...ListOption) ([]*Service, error) {
+func (n *noopRegistry) ListServices(ctx context.Context, opts ...ListOption) ([]*Service, error) {
 	return []*Service{}, nil
 }
 
 // Watch is used to watch for service changes
-func (n *noopRegistry) Watch(...WatchOption) (Watcher, error) {
+func (n *noopRegistry) Watch(ctx context.Context, opts ...WatchOption) (Watcher, error) {
 	return nil, fmt.Errorf("not implemented")
 }
 

registry/registry.go

@@ -30,11 +30,11 @@ type Registry interface {
 	Options() Options
 	Connect(context.Context) error
 	Disconnect(context.Context) error
-	Register(*Service, ...RegisterOption) error
+	Register(context.Context, *Service, ...RegisterOption) error
-	Deregister(*Service, ...DeregisterOption) error
+	Deregister(context.Context, *Service, ...DeregisterOption) error
-	GetService(string, ...GetOption) ([]*Service, error)
+	GetService(context.Context, string, ...GetOption) ([]*Service, error)
-	ListServices(...ListOption) ([]*Service, error)
+	ListServices(context.Context, ...ListOption) ([]*Service, error)
-	Watch(...WatchOption) (Watcher, error)
+	Watch(context.Context, ...WatchOption) (Watcher, error)
 	String() string
 }
 

resolver/registry/registry.go

@@ -2,6 +2,8 @@
 package registry
 
 import (
+	"context"
+
 	"github.com/unistack-org/micro/v3/registry"
 	"github.com/unistack-org/micro/v3/resolver"
 )
@@ -14,7 +16,7 @@ type Resolver struct {
 
 // Resolve assumes ID is a domain name e.g micro.mu
 func (r *Resolver) Resolve(name string) ([]*resolver.Record, error) {
-	services, err := r.Registry.GetService(name)
+	services, err := r.Registry.GetService(context.TODO(), name)
 	if err != nil {
 		return nil, err
 	}

server/noop.go

@@ -8,12 +8,12 @@ import (
 	"time"
 
 	craw "github.com/unistack-org/micro-codec-bytes"
+	cjson "github.com/unistack-org/micro-codec-json"
+	cjsonrpc "github.com/unistack-org/micro-codec-jsonrpc"
+	cproto "github.com/unistack-org/micro-codec-proto"
+	cprotorpc "github.com/unistack-org/micro-codec-protorpc"
 	"github.com/unistack-org/micro/v3/broker"
 	"github.com/unistack-org/micro/v3/codec"
-	cjson "github.com/unistack-org/micro/v3/codec/json"
-	cjsonrpc "github.com/unistack-org/micro/v3/codec/jsonrpc"
-	cproto "github.com/unistack-org/micro/v3/codec/proto"
-	cprotorpc "github.com/unistack-org/micro/v3/codec/protorpc"
 	"github.com/unistack-org/micro/v3/logger"
 	"github.com/unistack-org/micro/v3/registry"
 )

server/registry.go

@@ -12,7 +12,7 @@ import (
 
 var (
 	// DefaultRegisterFunc uses backoff to register service
-	DefaultRegisterFunc = func(service *registry.Service, config Options) error {
+	DefaultRegisterFunc = func(svc *registry.Service, config Options) error {
 		var err error
 
 		opts := []registry.RegisterOption{
@@ -21,7 +21,7 @@ var (
 		}
 
 		for i := 0; i <= config.RegisterAttempts; i++ {
-			err = config.Registry.Register(service, opts...)
+			err = config.Registry.Register(config.Context, svc, opts...)
 			if err == nil {
 				break
 			}
@@ -32,7 +32,7 @@ var (
 		return err
 	}
 	// DefaultDeregisterFunc uses backoff to deregister service
-	DefaultDeregisterFunc = func(service *registry.Service, config Options) error {
+	DefaultDeregisterFunc = func(svc *registry.Service, config Options) error {
 		var err error
 
 		opts := []registry.DeregisterOption{
@@ -40,7 +40,7 @@ var (
 		}
 
 		for i := 0; i <= config.DeregisterAttempts; i++ {
-			err = config.Registry.Deregister(service, opts...)
+			err = config.Registry.Deregister(config.Context, svc, opts...)
 			if err == nil {
 				break
 			}

sync/sync.go

@@ -6,6 +6,7 @@ import (
 )
 
 var (
+	// ErrLockTimeout error
 	ErrLockTimeout = errors.New("lock timeout")
 )
 

tracer/noop.go

@@ -2,24 +2,34 @@ package tracer
 
 import "context"
 
-type NoopTracer struct{}
+type noopTracer struct {
+	opts Options
+}
 
-func (n *NoopTracer) Init(...Option) error {
+// Init initilize tracer
+func (n *noopTracer) Init(opts ...Option) error {
+	for _, o := range opts {
+		o(&n.opts)
+	}
 	return nil
 }
 
-func (n *NoopTracer) Start(ctx context.Context, name string) (context.Context, *Span) {
+// Start starts new span
+func (n *noopTracer) Start(ctx context.Context, name string) (context.Context, *Span) {
 	return nil, nil
 }
 
-func (n *NoopTracer) Finish(*Span) error {
+// Finish finishes span
+func (n *noopTracer) Finish(*Span) error {
 	return nil
 }
 
-func (n *NoopTracer) Read(...ReadOption) ([]*Span, error) {
+// Read reads span
+func (n *noopTracer) Read(...ReadOption) ([]*Span, error) {
 	return nil, nil
 }
 
+// NewTracer returns new noop tracer
 func NewTracer(opts ...Option) Tracer {
-	return &NoopTracer{}
+	return &noopTracer{opts: NewOptions(opts...)}
 }

tracer/options.go

@@ -1,5 +1,10 @@
 package tracer
 
+var (
+	// DefaultSize of the buffer
+	DefaultSize = 64
+)
+
 // Options struct
 type Options struct {
 	// Size is the size of ring buffer
@@ -9,6 +14,7 @@ type Options struct {
 // Option func
 type Option func(o *Options)
 
+// ReadOptions struct
 type ReadOptions struct {
 	// Trace id
 	Trace string
@@ -24,14 +30,13 @@ func ReadTrace(t string) ReadOption {
 	}
 }
 
-const (
+// NewOptions returns default options
-	// DefaultSize of the buffer
+func NewOptions(opts ...Option) Options {
-	DefaultSize = 64
+	options := Options{
-)
-
-// DefaultOptions returns default options
-func DefaultOptions() Options {
-	return Options{
 		Size: DefaultSize,
 	}
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
 }

tracer/trace.go

@@ -1,4 +1,4 @@
-// Package trace provides an interface for distributed tracing
+// Package tracer provides an interface for distributed tracing
 package tracer
 
 import (
@@ -7,6 +7,7 @@ import (
 )
 
 var (
+	// DefaultTracer is the global default tracer
 	DefaultTracer Tracer = NewTracer()
 )
 

util/pool/default.go

@@ -1,6 +1,7 @@
 package pool
 
 import (
+	"context"
 	"sync"
 	"time"
 
@@ -57,7 +58,7 @@ func (p *poolConn) Created() time.Time {
 	return p.created
 }
 
-func (p *pool) Get(addr string, opts ...transport.DialOption) (Conn, error) {
+func (p *pool) Get(ctx context.Context, addr string, opts ...transport.DialOption) (Conn, error) {
 	p.Lock()
 	conns := p.conns[addr]
 
@@ -83,7 +84,7 @@ func (p *pool) Get(addr string, opts ...transport.DialOption) (Conn, error) {
 	p.Unlock()
 
 	// create new conn
-	c, err := p.tr.Dial(addr, opts...)
+	c, err := p.tr.Dial(ctx, addr, opts...)
 	if err != nil {
 		return nil, err
 	}

util/pool/pool.go

@@ -2,6 +2,7 @@
 package pool
 
 import (
+	"context"
 	"time"
 
 	"github.com/unistack-org/micro/v3/network/transport"
@@ -12,7 +13,7 @@ type Pool interface {
 	// Close the pool
 	Close() error
 	// Get a connection
-	Get(addr string, opts ...transport.DialOption) (Conn, error)
+	Get(ctx context.Context, addr string, opts ...transport.DialOption) (Conn, error)
 	// Release the connection
 	Release(c Conn, status error) error
 }