Add grpc plugin (#2758)

* Add grpc plugin * remove stale readmes
2025-08-04 21:42:57 +02:00 · 2025-05-14 21:04:42 +01:00
parent 01b8394c81
commit 230505bf5a
40 changed files with 5527 additions and 0 deletions
--- a/client/grpc/codec.go
+++ b/client/grpc/codec.go
@ -0,0 +1,209 @@
+package grpc
+
+import (
+	b "bytes"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"go-micro.dev/v5/codec"
+	"go-micro.dev/v5/codec/bytes"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/encoding"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/runtime/protoiface"
+	"google.golang.org/protobuf/runtime/protoimpl"
+)
+
+type jsonCodec struct{}
+type protoCodec struct{}
+type bytesCodec struct{}
+type wrapCodec struct{ encoding.Codec }
+
+var useNumber bool
+
+var (
+	defaultGRPCCodecs = map[string]encoding.Codec{
+		"application/json":         jsonCodec{},
+		"application/proto":        protoCodec{},
+		"application/protobuf":     protoCodec{},
+		"application/octet-stream": protoCodec{},
+		"application/grpc":         protoCodec{},
+		"application/grpc+json":    jsonCodec{},
+		"application/grpc+proto":   protoCodec{},
+		"application/grpc+bytes":   bytesCodec{},
+	}
+)
+
+// UseNumber fix unmarshal Number(8234567890123456789) to interface(8.234567890123457e+18).
+func UseNumber() {
+	useNumber = true
+}
+
+func (w wrapCodec) String() string {
+	return w.Codec.Name()
+}
+
+func (w wrapCodec) Marshal(v interface{}) ([]byte, error) {
+	b, ok := v.(*bytes.Frame)
+	if ok {
+		return b.Data, nil
+	}
+	return w.Codec.Marshal(v)
+}
+
+func (w wrapCodec) Unmarshal(data []byte, v interface{}) error {
+	b, ok := v.(*bytes.Frame)
+	if ok {
+		b.Data = data
+		return nil
+	}
+	return w.Codec.Unmarshal(data, v)
+}
+
+func (protoCodec) Marshal(v interface{}) ([]byte, error) {
+	switch m := v.(type) {
+	case *bytes.Frame:
+		return m.Data, nil
+	case proto.Message:
+		return proto.Marshal(m)
+	case protoiface.MessageV1:
+		// #2333 compatible with etcd legacy proto.Message
+		m2 := protoimpl.X.ProtoMessageV2Of(m)
+		return proto.Marshal(m2)
+	}
+	return nil, fmt.Errorf("failed to marshal: %v is not type of *bytes.Frame or proto.Message", v)
+}
+
+func (protoCodec) Unmarshal(data []byte, v interface{}) error {
+	switch m := v.(type) {
+	case proto.Message:
+		return proto.Unmarshal(data, m)
+	case protoiface.MessageV1:
+		// #2333 compatible with etcd legacy proto.Message
+		m2 := protoimpl.X.ProtoMessageV2Of(m)
+		return proto.Unmarshal(data, m2)
+	}
+	return fmt.Errorf("failed to unmarshal: %v is not type of proto.Message", v)
+}
+
+func (protoCodec) Name() string {
+	return "proto"
+}
+
+func (bytesCodec) Marshal(v interface{}) ([]byte, error) {
+	b, ok := v.(*[]byte)
+	if !ok {
+		return nil, fmt.Errorf("failed to marshal: %v is not type of *[]byte", v)
+	}
+	return *b, nil
+}
+
+func (bytesCodec) Unmarshal(data []byte, v interface{}) error {
+	b, ok := v.(*[]byte)
+	if !ok {
+		return fmt.Errorf("failed to unmarshal: %v is not type of *[]byte", v)
+	}
+	*b = data
+	return nil
+}
+
+func (bytesCodec) Name() string {
+	return "bytes"
+}
+
+func (jsonCodec) Marshal(v interface{}) ([]byte, error) {
+	if b, ok := v.(*bytes.Frame); ok {
+		return b.Data, nil
+	}
+
+	if pb, ok := v.(proto.Message); ok {
+		bytes, err := protojson.Marshal(pb)
+		if err != nil {
+			return nil, err
+		}
+		return bytes, nil
+	}
+
+	return json.Marshal(v)
+}
+
+func (jsonCodec) Unmarshal(data []byte, v interface{}) error {
+	if len(data) == 0 {
+		return nil
+	}
+	if b, ok := v.(*bytes.Frame); ok {
+		b.Data = data
+		return nil
+	}
+	if pb, ok := v.(proto.Message); ok {
+		return protojson.Unmarshal(data, pb)
+	}
+
+	dec := json.NewDecoder(b.NewReader(data))
+	if useNumber {
+		dec.UseNumber()
+	}
+	return dec.Decode(v)
+}
+
+func (jsonCodec) Name() string {
+	return "json"
+}
+
+type grpcCodec struct {
+	// headers
+	id       string
+	target   string
+	method   string
+	endpoint string
+
+	s grpc.ClientStream
+	c encoding.Codec
+}
+
+func (g *grpcCodec) ReadHeader(m *codec.Message, mt codec.MessageType) error {
+	md, err := g.s.Header()
+	if err != nil {
+		return err
+	}
+	if m == nil {
+		m = new(codec.Message)
+	}
+	if m.Header == nil {
+		m.Header = make(map[string]string, len(md))
+	}
+	for k, v := range md {
+		m.Header[k] = strings.Join(v, ",")
+	}
+	m.Id = g.id
+	m.Target = g.target
+	m.Method = g.method
+	m.Endpoint = g.endpoint
+	return nil
+}
+
+func (g *grpcCodec) ReadBody(v interface{}) error {
+	if f, ok := v.(*bytes.Frame); ok {
+		return g.s.RecvMsg(f)
+	}
+	return g.s.RecvMsg(v)
+}
+
+func (g *grpcCodec) Write(m *codec.Message, v interface{}) error {
+	// if we don't have a body
+	if v != nil {
+		return g.s.SendMsg(v)
+	}
+	// write the body using the framing codec
+	return g.s.SendMsg(&bytes.Frame{Data: m.Body})
+}
+
+func (g *grpcCodec) Close() error {
+	return g.s.CloseSend()
+}
+
+func (g *grpcCodec) String() string {
+	return g.c.Name()
+}
--- a/client/grpc/error.go
+++ b/client/grpc/error.go
@ -0,0 +1,69 @@
+package grpc
+
+import (
+	"net/http"
+
+	"go-micro.dev/v5/errors"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func microError(err error) error {
+	// no error
+	switch err {
+	case nil:
+		return nil
+	}
+
+	if verr, ok := err.(*errors.Error); ok {
+		return verr
+	}
+
+	// grpc error
+	s, ok := status.FromError(err)
+	if !ok {
+		return err
+	}
+
+	// return first error from details
+	if details := s.Details(); len(details) > 0 {
+		return microError(details[0].(error))
+	}
+
+	// try to decode micro *errors.Error
+	if e := errors.Parse(s.Message()); e.Code > 0 {
+		return e // actually a micro error
+	}
+
+	// fallback
+	return errors.New("go.micro.client", s.Message(), microStatusFromGrpcCode(s.Code()))
+}
+
+func microStatusFromGrpcCode(code codes.Code) int32 {
+	switch code {
+	case codes.OK:
+		return http.StatusOK
+	case codes.InvalidArgument:
+		return http.StatusBadRequest
+	case codes.DeadlineExceeded:
+		return http.StatusRequestTimeout
+	case codes.NotFound:
+		return http.StatusNotFound
+	case codes.AlreadyExists:
+		return http.StatusConflict
+	case codes.PermissionDenied:
+		return http.StatusForbidden
+	case codes.Unauthenticated:
+		return http.StatusUnauthorized
+	case codes.FailedPrecondition:
+		return http.StatusPreconditionFailed
+	case codes.Unimplemented:
+		return http.StatusNotImplemented
+	case codes.Internal:
+		return http.StatusInternalServerError
+	case codes.Unavailable:
+		return http.StatusServiceUnavailable
+	}
+
+	return http.StatusInternalServerError
+}
--- a/client/grpc/go.mod
+++ b/client/grpc/go.mod
@ -0,0 +1,35 @@
+module go-micro.dev/v5/client/grpc
+
+go 1.24
+
+require (
+	go-micro.dev/v5 v5.5.1-0.20250507183911-01b8394c8119
+	google.golang.org/grpc v1.53.0
+	google.golang.org/grpc/examples v0.0.0-20211102180624-670c133e568e
+	google.golang.org/protobuf v1.33.0
+)
+
+require (
+	github.com/bitly/go-simplejson v0.5.0 // indirect
+	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
+	github.com/fsnotify/fsnotify v1.6.0 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
+	github.com/google/uuid v1.3.0 // indirect
+	github.com/imdario/mergo v0.3.13 // indirect
+	github.com/miekg/dns v1.1.50 // indirect
+	github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c // indirect
+	github.com/patrickmn/go-cache v2.1.0+incompatible // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/rogpeppe/go-internal v1.12.0 // indirect
+	github.com/russross/blackfriday/v2 v2.1.0 // indirect
+	github.com/urfave/cli/v2 v2.25.7 // indirect
+	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
+	go.etcd.io/bbolt v1.4.0 // indirect
+	golang.org/x/mod v0.9.0 // indirect
+	golang.org/x/net v0.23.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.29.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/tools v0.7.0 // indirect
+	google.golang.org/genproto v0.0.0-20230306155012-7f2fa6fef1f4 // indirect
+)
--- a/client/grpc/go.sum
+++ b/client/grpc/go.sum
@ -0,0 +1,175 @@
+cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/bitly/go-simplejson v0.5.0 h1:6IH+V8/tVMab511d5bn4M7EwGXZf9Hj6i2xSwkNEM+Y=
+github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA=
+github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
+github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
+github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
+github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
+github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
+github.com/cpuguy83/go-md2man/v2 v2.0.2 h1:p1EgwI/C7NhT0JmVkwCD2ZBK8j4aeHQX2pMHHBfMQ6w=
+github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
+github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
+github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
+github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
+github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
+github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
+github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
+github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
+github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
+github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
+github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
+github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
+github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
+github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/imdario/mergo v0.3.13 h1:lFzP57bqS/wsqKssCGmtLAb8A0wKjLGrve2q3PPVcBk=
+github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg=
+github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/miekg/dns v1.1.50 h1:DQUfb9uc6smULcREF09Uc+/Gd46YWqJd5DbpPE9xkcA=
+github.com/miekg/dns v1.1.50/go.mod h1:e3IlAVfNqAllflbibAZEWOXOQ+Ynzk/dDozDxY7XnME=
+github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c h1:rp5dCmg/yLR3mgFuSOe4oEnDDmGLROTvMragMUXpTQw=
+github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c/go.mod h1:X07ZCGwUbLaax7L0S3Tw4hpejzu63ZrrQiUe6W0hcy0=
+github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
+github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
+github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
+github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
+github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/urfave/cli/v2 v2.25.7 h1:VAzn5oq403l5pHjc4OhD54+XGO9cdKVL/7lDjF+iKUs=
+github.com/urfave/cli/v2 v2.25.7/go.mod h1:8qnjx1vcq5s2/wpsqoZFndg2CE5tNFyrTvS6SinrnYQ=
+github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 h1:bAn7/zixMGCfxrRTfdpNzjtPYqr8smhKouy9mxVdGPU=
+github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673/go.mod h1:N3UwUGtsrSj3ccvlPHLoLsHnpR27oXr4ZE984MbSER8=
+github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
+go-micro.dev/v5 v5.5.1-0.20250507183911-01b8394c8119 h1:hKzFHTSwaYvPS0sTrWpUcS2HFFbJte//n91NMWFFk4k=
+go-micro.dev/v5 v5.5.1-0.20250507183911-01b8394c8119/go.mod h1:pplSGFQk5So43OrzGAg1gH434XydWybOTyFhv/nR0pA=
+go.etcd.io/bbolt v1.4.0 h1:TU77id3TnN/zKr7CO/uk+fBCwF2jGcMuw2B/FMAzYIk=
+go.etcd.io/bbolt v1.4.0/go.mod h1:AsD+OCi/qPN1giOX1aiLAha3o1U8rAz65bvN4j0sRuk=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
+golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
+golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
+golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.9.0 h1:KENHtAZL2y3NLMYZeHY9DW8HW8V+kQyJsY/V9JlKvCs=
+golang.org/x/mod v0.9.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
+golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.23.0 h1:7EYJ93RZ9vYSZAIb2x3lnuvqO5zneoD6IvWjuhfxjTs=
+golang.org/x/net v0.23.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
+golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
+golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
+golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
+golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
+golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.1.6-0.20210726203631-07bc1bf47fb2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
+golang.org/x/tools v0.7.0 h1:W4OVu8VVOaIO0yzWMNdepAulS7YfoS3Zabrm8DOXXU4=
+golang.org/x/tools v0.7.0/go.mod h1:4pg6aUX35JBAogB10C9AtvVL+qowtN4pT3CGSQex14s=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
+google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
+google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
+google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
+google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
+google.golang.org/genproto v0.0.0-20200806141610-86f49bd18e98/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
+google.golang.org/genproto v0.0.0-20230306155012-7f2fa6fef1f4 h1:DdoeryqhaXp1LtT/emMP1BRJPHHKFi5akj/nbx/zNTA=
+google.golang.org/genproto v0.0.0-20230306155012-7f2fa6fef1f4/go.mod h1:NWraEVixdDnqcqQ30jipen1STv2r/n24Wb7twVTGR4s=
+google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
+google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
+google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
+google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
+google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
+google.golang.org/grpc v1.53.0 h1:LAv2ds7cmFV/XTS3XG1NneeENYrXGmorPxsBbptIjNc=
+google.golang.org/grpc v1.53.0/go.mod h1:OnIrk0ipVdj4N5d9IUoFUx72/VlD7+jUsHwZgwSMQpw=
+google.golang.org/grpc/examples v0.0.0-20211102180624-670c133e568e h1:m7aQHHqd0q89mRwhwS9Bx2rjyl/hsFAeta+uGrHsQaU=
+google.golang.org/grpc/examples v0.0.0-20211102180624-670c133e568e/go.mod h1:gID3PKrg7pWKntu9Ss6zTLJ0ttC0X9IHgREOCZwbCVU=
+google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
+google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
+google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
+google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
+google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
+google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
+google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
+google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
+google.golang.org/protobuf v1.24.0/go.mod h1:r/3tXBNzIEhYS9I1OUVjXDlt8tc493IdKGjtUeSXeh4=
+google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
+honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
--- a/client/grpc/grpc.go
+++ b/client/grpc/grpc.go
@ -0,0 +1,709 @@
+// Package grpc provides a gRPC client
+package grpc
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net"
+	"reflect"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"go-micro.dev/v5/broker"
+	"go-micro.dev/v5/client"
+	"go-micro.dev/v5/cmd"
+	raw "go-micro.dev/v5/codec/bytes"
+	"go-micro.dev/v5/errors"
+	"go-micro.dev/v5/metadata"
+	"go-micro.dev/v5/registry"
+	"go-micro.dev/v5/selector"
+	pnet "go-micro.dev/v5/util/net"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/encoding"
+	gmetadata "google.golang.org/grpc/metadata"
+)
+
+type grpcClient struct {
+	opts client.Options
+	pool *pool
+	once atomic.Value
+}
+
+func init() {
+	cmd.DefaultClients["grpc"] = NewClient
+
+	encoding.RegisterCodec(wrapCodec{jsonCodec{}})
+	encoding.RegisterCodec(wrapCodec{protoCodec{}})
+	encoding.RegisterCodec(wrapCodec{bytesCodec{}})
+}
+
+// secure returns the dial option for whether its a secure or insecure connection.
+func (g *grpcClient) secure(addr string) grpc.DialOption {
+	// first we check if theres'a  tls config
+	if g.opts.Context != nil {
+		if v := g.opts.Context.Value(tlsAuth{}); v != nil {
+			tls := v.(*tls.Config)
+			creds := credentials.NewTLS(tls)
+			// return tls config if it exists
+			return grpc.WithTransportCredentials(creds)
+		}
+	}
+
+	// default config
+	tlsConfig := &tls.Config{}
+	defaultCreds := grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig))
+
+	// check if the address is prepended with https
+	if strings.HasPrefix(addr, "https://") {
+		return defaultCreds
+	}
+
+	// if no port is specified or port is 443 default to tls
+	_, port, err := net.SplitHostPort(addr)
+	// assuming with no port its going to be secured
+	if port == "443" {
+		return defaultCreds
+	} else if err != nil && strings.Contains(err.Error(), "missing port in address") {
+		return defaultCreds
+	}
+
+	// other fallback to insecure
+	return grpc.WithInsecure()
+}
+
+func (g *grpcClient) next(request client.Request, opts client.CallOptions) (selector.Next, error) {
+	service, address, _ := pnet.Proxy(request.Service(), opts.Address)
+
+	// return remote address
+	if len(address) > 0 {
+		return func() (*registry.Node, error) {
+			return &registry.Node{
+				Address: address[0],
+			}, nil
+		}, nil
+	}
+
+	// get next nodes from the selector
+	next, err := g.opts.Selector.Select(service, opts.SelectOptions...)
+	if err != nil {
+		if err == selector.ErrNotFound {
+			return nil, errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
+		}
+		return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
+	}
+
+	return next, nil
+}
+
+func (g *grpcClient) call(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
+	var header map[string]string
+
+	address := node.Address
+
+	if md, ok := metadata.FromContext(ctx); ok {
+		header = make(map[string]string, len(md))
+		for k, v := range md {
+			header[strings.ToLower(k)] = v
+		}
+	} else {
+		header = make(map[string]string)
+	}
+
+	// set timeout in nanoseconds
+	header["timeout"] = fmt.Sprintf("%d", opts.RequestTimeout)
+	// set the content type for the request
+	header["x-content-type"] = req.ContentType()
+
+	md := gmetadata.New(header)
+	ctx = gmetadata.NewOutgoingContext(ctx, md)
+
+	cf, err := g.newGRPCCodec(req.ContentType())
+	if err != nil {
+		return errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	maxRecvMsgSize := g.maxRecvMsgSizeValue()
+	maxSendMsgSize := g.maxSendMsgSizeValue()
+
+	var grr error
+
+	var dialCtx context.Context
+	var cancel context.CancelFunc
+	if opts.DialTimeout > 0 {
+		dialCtx, cancel = context.WithTimeout(ctx, opts.DialTimeout)
+	} else {
+		dialCtx, cancel = context.WithCancel(ctx)
+	}
+	defer cancel()
+
+	grpcDialOptions := []grpc.DialOption{
+		g.secure(address),
+		grpc.WithDefaultCallOptions(
+			grpc.MaxCallRecvMsgSize(maxRecvMsgSize),
+			grpc.MaxCallSendMsgSize(maxSendMsgSize),
+		),
+	}
+
+	if opts := g.getGrpcDialOptions(); opts != nil {
+		grpcDialOptions = append(grpcDialOptions, opts...)
+	}
+
+	cc, err := g.pool.getConn(dialCtx, address, grpcDialOptions...)
+	if err != nil {
+		return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
+	}
+	defer func() {
+		// defer execution of release
+		g.pool.release(address, cc, grr)
+	}()
+
+	ch := make(chan error, 1)
+
+	go func() {
+		grpcCallOptions := []grpc.CallOption{
+			grpc.ForceCodec(cf),
+			grpc.CallContentSubtype(cf.Name())}
+		if opts := callOpts(opts); opts != nil {
+			grpcCallOptions = append(grpcCallOptions, opts...)
+		}
+		err := cc.Invoke(ctx, methodToGRPC(req.Service(), req.Endpoint()), req.Body(), rsp, grpcCallOptions...)
+		ch <- microError(err)
+	}()
+
+	select {
+	case err := <-ch:
+		grr = err
+	case <-ctx.Done():
+		grr = errors.Timeout("go.micro.client", "%v", ctx.Err())
+	}
+
+	return grr
+}
+
+func (g *grpcClient) stream(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
+	var header map[string]string
+
+	address := node.Address
+
+	if md, ok := metadata.FromContext(ctx); ok {
+		header = make(map[string]string, len(md))
+		for k, v := range md {
+			header[k] = v
+		}
+	} else {
+		header = make(map[string]string)
+	}
+
+	// set timeout in nanoseconds
+	if opts.StreamTimeout > time.Duration(0) {
+		header["timeout"] = fmt.Sprintf("%d", opts.StreamTimeout)
+	}
+	// set the content type for the request
+	header["x-content-type"] = req.ContentType()
+
+	md := gmetadata.New(header)
+
+	// WebSocket connection adds the `Connection: Upgrade` header.
+	// But as per the HTTP/2 spec, the `Connection` header makes the request malformed
+	delete(md, "connection")
+
+	ctx = gmetadata.NewOutgoingContext(ctx, md)
+
+	cf, err := g.newGRPCCodec(req.ContentType())
+	if err != nil {
+		return errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	var dialCtx context.Context
+	var cancel context.CancelFunc
+	if opts.DialTimeout > 0 {
+		dialCtx, cancel = context.WithTimeout(ctx, opts.DialTimeout)
+	} else {
+		dialCtx, cancel = context.WithCancel(ctx)
+	}
+	defer cancel()
+
+	wc := wrapCodec{cf}
+
+	grpcDialOptions := []grpc.DialOption{
+		g.secure(address),
+	}
+
+	if opts := g.getGrpcDialOptions(); opts != nil {
+		grpcDialOptions = append(grpcDialOptions, opts...)
+	}
+
+	cc, err := g.pool.getConn(dialCtx, address, grpcDialOptions...)
+	if err != nil {
+		return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
+	}
+
+	desc := &grpc.StreamDesc{
+		StreamName:    req.Service() + req.Endpoint(),
+		ClientStreams: true,
+		ServerStreams: true,
+	}
+
+	grpcCallOptions := []grpc.CallOption{
+		grpc.ForceCodec(wc),
+		grpc.CallContentSubtype(cf.Name()),
+	}
+	if opts := callOpts(opts); opts != nil {
+		grpcCallOptions = append(grpcCallOptions, opts...)
+	}
+
+	// create a new canceling context
+	newCtx, cancel := context.WithCancel(ctx)
+
+	st, err := cc.NewStream(newCtx, desc, methodToGRPC(req.Service(), req.Endpoint()), grpcCallOptions...)
+	if err != nil {
+		// we need to cleanup as we dialed and created a context
+		// cancel the context
+		cancel()
+		// close the connection
+		cc.Close()
+		// now return the error
+		return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err))
+	}
+
+	codec := &grpcCodec{
+		s: st,
+		c: wc,
+	}
+
+	// set request codec
+	if r, ok := req.(*grpcRequest); ok {
+		r.codec = codec
+	}
+
+	// setup the stream response
+	stream := &grpcStream{
+		context: ctx,
+		request: req,
+		response: &response{
+			conn:   cc.ClientConn,
+			stream: st,
+			codec:  cf,
+			gcodec: codec,
+		},
+		stream: st,
+		cancel: cancel,
+		release: func(err error) {
+			g.pool.release(address, cc, err)
+		},
+	}
+
+	// set the stream as the response
+	val := reflect.ValueOf(rsp).Elem()
+	val.Set(reflect.ValueOf(stream).Elem())
+	return nil
+}
+
+func (g *grpcClient) poolMaxStreams() int {
+	if g.opts.Context == nil {
+		return DefaultPoolMaxStreams
+	}
+	v := g.opts.Context.Value(poolMaxStreams{})
+	if v == nil {
+		return DefaultPoolMaxStreams
+	}
+	return v.(int)
+}
+
+func (g *grpcClient) poolMaxIdle() int {
+	if g.opts.Context == nil {
+		return DefaultPoolMaxIdle
+	}
+	v := g.opts.Context.Value(poolMaxIdle{})
+	if v == nil {
+		return DefaultPoolMaxIdle
+	}
+	return v.(int)
+}
+
+func (g *grpcClient) maxRecvMsgSizeValue() int {
+	if g.opts.Context == nil {
+		return DefaultMaxRecvMsgSize
+	}
+	v := g.opts.Context.Value(maxRecvMsgSizeKey{})
+	if v == nil {
+		return DefaultMaxRecvMsgSize
+	}
+	return v.(int)
+}
+
+func (g *grpcClient) maxSendMsgSizeValue() int {
+	if g.opts.Context == nil {
+		return DefaultMaxSendMsgSize
+	}
+	v := g.opts.Context.Value(maxSendMsgSizeKey{})
+	if v == nil {
+		return DefaultMaxSendMsgSize
+	}
+	return v.(int)
+}
+
+func (g *grpcClient) newGRPCCodec(contentType string) (encoding.Codec, error) {
+	codecs := make(map[string]encoding.Codec)
+	if g.opts.Context != nil {
+		if v := g.opts.Context.Value(codecsKey{}); v != nil {
+			codecs = v.(map[string]encoding.Codec)
+		}
+	}
+	if c, ok := codecs[contentType]; ok {
+		return wrapCodec{c}, nil
+	}
+	if c, ok := defaultGRPCCodecs[contentType]; ok {
+		return wrapCodec{c}, nil
+	}
+	return nil, fmt.Errorf("unsupported Content-Type: %s", contentType)
+}
+
+func (g *grpcClient) Init(opts ...client.Option) error {
+	size := g.opts.PoolSize
+	ttl := g.opts.PoolTTL
+
+	for _, o := range opts {
+		o(&g.opts)
+	}
+
+	// update pool configuration if the options changed
+	if size != g.opts.PoolSize || ttl != g.opts.PoolTTL {
+		g.pool.Lock()
+		g.pool.size = g.opts.PoolSize
+		g.pool.ttl = int64(g.opts.PoolTTL.Seconds())
+		g.pool.Unlock()
+	}
+
+	return nil
+}
+
+func (g *grpcClient) Options() client.Options {
+	return g.opts
+}
+
+func (g *grpcClient) NewMessage(topic string, msg interface{}, opts ...client.MessageOption) client.Message {
+	return newGRPCEvent(topic, msg, g.opts.ContentType, opts...)
+}
+
+func (g *grpcClient) NewRequest(service, method string, req interface{}, reqOpts ...client.RequestOption) client.Request {
+	return newGRPCRequest(service, method, req, g.opts.ContentType, reqOpts...)
+}
+
+func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
+	if req == nil {
+		return errors.InternalServerError("go.micro.client", "req is nil")
+	} else if rsp == nil {
+		return errors.InternalServerError("go.micro.client", "rsp is nil")
+	}
+	// make a copy of call opts
+	callOpts := g.opts.CallOptions
+	for _, opt := range opts {
+		opt(&callOpts)
+	}
+
+	next, err := g.next(req, callOpts)
+	if err != nil {
+		return err
+	}
+
+	// check if we already have a deadline
+	d, ok := ctx.Deadline()
+	if !ok {
+		// no deadline so we create a new one
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, callOpts.RequestTimeout)
+		defer cancel()
+	} else {
+		// got a deadline so no need to setup context
+		// but we need to set the timeout we pass along
+		opt := client.WithRequestTimeout(time.Until(d))
+		opt(&callOpts)
+	}
+
+	// should we noop right here?
+	select {
+	case <-ctx.Done():
+		return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
+	default:
+	}
+
+	// make copy of call method
+	gcall := g.call
+
+	// wrap the call in reverse
+	for i := len(callOpts.CallWrappers); i > 0; i-- {
+		gcall = callOpts.CallWrappers[i-1](gcall)
+	}
+
+	// return errors.New("go.micro.client", "request timeout", 408)
+	call := func(i int) error {
+		// call backoff first. Someone may want an initial start delay
+		t, err := callOpts.Backoff(ctx, req, i)
+		if err != nil {
+			return errors.InternalServerError("go.micro.client", err.Error())
+		}
+
+		// only sleep if greater than 0
+		if t.Seconds() > 0 {
+			time.Sleep(t)
+		}
+
+		// select next node
+		node, err := next()
+		service := req.Service()
+		if err != nil {
+			if err == selector.ErrNotFound {
+				return errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
+			}
+			return errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
+		}
+
+		// make the call
+		err = gcall(ctx, node, req, rsp, callOpts)
+		g.opts.Selector.Mark(service, node, err)
+		if verr, ok := err.(*errors.Error); ok {
+			return verr
+		}
+
+		return err
+	}
+
+	ch := make(chan error, callOpts.Retries+1)
+	var gerr error
+
+	for i := 0; i <= callOpts.Retries; i++ {
+		go func(i int) {
+			ch <- call(i)
+		}(i)
+
+		select {
+		case <-ctx.Done():
+			return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
+		case err := <-ch:
+			// if the call succeeded lets bail early
+			if err == nil {
+				return nil
+			}
+
+			retry, rerr := callOpts.Retry(ctx, req, i, err)
+			if rerr != nil {
+				return rerr
+			}
+
+			if !retry {
+				return err
+			}
+
+			gerr = err
+		}
+	}
+
+	return gerr
+}
+
+func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
+	// make a copy of call opts
+	callOpts := g.opts.CallOptions
+	for _, opt := range opts {
+		opt(&callOpts)
+	}
+
+	next, err := g.next(req, callOpts)
+	if err != nil {
+		return nil, err
+	}
+
+	// #200 - streams shouldn't have a request timeout set on the context
+
+	// should we noop right here?
+	select {
+	case <-ctx.Done():
+		return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
+	default:
+	}
+
+	// make a copy of stream
+	gstream := g.stream
+
+	// wrap the call in reverse
+	for i := len(callOpts.CallWrappers); i > 0; i-- {
+		gstream = callOpts.CallWrappers[i-1](gstream)
+	}
+
+	call := func(i int) (client.Stream, error) {
+		// call backoff first. Someone may want an initial start delay
+		t, err := callOpts.Backoff(ctx, req, i)
+		if err != nil {
+			return nil, errors.InternalServerError("go.micro.client", err.Error())
+		}
+
+		// only sleep if greater than 0
+		if t.Seconds() > 0 {
+			time.Sleep(t)
+		}
+
+		node, err := next()
+		service := req.Service()
+		if err != nil {
+			if err == selector.ErrNotFound {
+				return nil, errors.InternalServerError("go.micro.client", "service %s: %s", service, err.Error())
+			}
+			return nil, errors.InternalServerError("go.micro.client", "error selecting %s node: %s", service, err.Error())
+		}
+
+		// make the call
+		stream := &grpcStream{}
+		err = g.stream(ctx, node, req, stream, callOpts)
+
+		g.opts.Selector.Mark(service, node, err)
+		return stream, err
+	}
+
+	type response struct {
+		stream client.Stream
+		err    error
+	}
+
+	ch := make(chan response, callOpts.Retries+1)
+	var grr error
+
+	for i := 0; i <= callOpts.Retries; i++ {
+		go func(i int) {
+			s, err := call(i)
+			ch <- response{s, err}
+		}(i)
+
+		select {
+		case <-ctx.Done():
+			return nil, errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
+		case rsp := <-ch:
+			// if the call succeeded lets bail early
+			if rsp.err == nil {
+				return rsp.stream, nil
+			}
+
+			retry, rerr := callOpts.Retry(ctx, req, i, err)
+			if rerr != nil {
+				return nil, rerr
+			}
+
+			if !retry {
+				return nil, rsp.err
+			}
+
+			grr = rsp.err
+		}
+	}
+
+	return nil, grr
+}
+
+func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
+	var options client.PublishOptions
+	for _, o := range opts {
+		o(&options)
+	}
+
+	md, ok := metadata.FromContext(ctx)
+	if !ok {
+		md = make(map[string]string)
+	}
+	md["Content-Type"] = p.ContentType()
+	md["Micro-Topic"] = p.Topic()
+
+	cf, err := g.newGRPCCodec(p.ContentType())
+	if err != nil {
+		return errors.InternalServerError("go.micro.client", err.Error())
+	}
+
+	var body []byte
+
+	// passed in raw data
+	if d, ok := p.Payload().(*raw.Frame); ok {
+		body = d.Data
+	} else {
+		// set the body
+		b, err := cf.Marshal(p.Payload())
+		if err != nil {
+			return errors.InternalServerError("go.micro.client", err.Error())
+		}
+		body = b
+	}
+
+	if !g.once.Load().(bool) {
+		if err = g.opts.Broker.Connect(); err != nil {
+			return errors.InternalServerError("go.micro.client", err.Error())
+		}
+		g.once.Store(true)
+	}
+
+	topic := p.Topic()
+
+	// get the exchange
+	if len(options.Exchange) > 0 {
+		topic = options.Exchange
+	}
+
+	return g.opts.Broker.Publish(topic, &broker.Message{
+		Header: md,
+		Body:   body,
+	}, broker.PublishContext(options.Context))
+}
+
+func (g *grpcClient) String() string {
+	return "grpc"
+}
+
+func (g *grpcClient) getGrpcDialOptions() []grpc.DialOption {
+	if g.opts.CallOptions.Context == nil {
+		return nil
+	}
+
+	v := g.opts.CallOptions.Context.Value(grpcDialOptions{})
+
+	if v == nil {
+		return nil
+	}
+
+	opts, ok := v.([]grpc.DialOption)
+
+	if !ok {
+		return nil
+	}
+
+	return opts
+}
+
+func newClient(opts ...client.Option) client.Client {
+	options := client.NewOptions()
+	// default content type for grpc
+	options.ContentType = "application/grpc+proto"
+
+	for _, o := range opts {
+		o(&options)
+	}
+
+	rc := &grpcClient{
+		opts: options,
+	}
+	rc.once.Store(false)
+
+	rc.pool = newPool(options.PoolSize, options.PoolTTL, rc.poolMaxIdle(), rc.poolMaxStreams())
+
+	c := client.Client(rc)
+
+	// wrap in reverse
+	for i := len(options.Wrappers); i > 0; i-- {
+		c = options.Wrappers[i-1](c)
+	}
+
+	return c
+}
+
+func NewClient(opts ...client.Option) client.Client {
+	return newClient(opts...)
+}
--- a/client/grpc/grpc_pool.go
+++ b/client/grpc/grpc_pool.go
@ -0,0 +1,218 @@
+package grpc
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+)
+
+type pool struct {
+	size int
+	ttl  int64
+
+	//  max streams on a *poolConn
+	maxStreams int
+	//  max idle conns
+	maxIdle int
+
+	sync.Mutex
+	conns map[string]*streamsPool
+}
+
+type streamsPool struct {
+	//  head of list
+	head *poolConn
+	//  busy conns list
+	busy *poolConn
+	//  the size of list
+	count int
+	//  idle conn
+	idle int
+}
+
+type poolConn struct {
+	//  grpc conn
+	*grpc.ClientConn
+	err  error
+	addr string
+
+	//  pool and streams pool
+	pool    *pool
+	sp      *streamsPool
+	streams int
+	created int64
+
+	//  list
+	pre  *poolConn
+	next *poolConn
+	in   bool
+}
+
+func newPool(size int, ttl time.Duration, idle int, ms int) *pool {
+	if ms <= 0 {
+		ms = 1
+	}
+	if idle < 0 {
+		idle = 0
+	}
+	return &pool{
+		size:       size,
+		ttl:        int64(ttl.Seconds()),
+		maxStreams: ms,
+		maxIdle:    idle,
+		conns:      make(map[string]*streamsPool),
+	}
+}
+
+func (p *pool) getConn(dialCtx context.Context, addr string, opts ...grpc.DialOption) (*poolConn, error) {
+	now := time.Now().Unix()
+	p.Lock()
+	sp, ok := p.conns[addr]
+	if !ok {
+		sp = &streamsPool{head: &poolConn{}, busy: &poolConn{}, count: 0, idle: 0}
+		p.conns[addr] = sp
+	}
+	//  while we have conns check streams and then return one
+	//  otherwise we'll create a new conn
+	conn := sp.head.next
+	for conn != nil {
+		//  check conn state
+		// https://github.com/grpc/grpc/blob/master/doc/connectivity-semantics-and-api.md
+		switch conn.GetState() {
+		case connectivity.Connecting:
+			conn = conn.next
+			continue
+		case connectivity.Shutdown:
+			next := conn.next
+			if conn.streams == 0 {
+				removeConn(conn)
+				sp.idle--
+			}
+			conn = next
+			continue
+		case connectivity.TransientFailure:
+			next := conn.next
+			if conn.streams == 0 {
+				removeConn(conn)
+				conn.ClientConn.Close()
+				sp.idle--
+			}
+			conn = next
+			continue
+		case connectivity.Ready:
+		case connectivity.Idle:
+		}
+		//  a old conn
+		if now-conn.created > p.ttl {
+			next := conn.next
+			if conn.streams == 0 {
+				removeConn(conn)
+				conn.ClientConn.Close()
+				sp.idle--
+			}
+			conn = next
+			continue
+		}
+		//  a busy conn
+		if conn.streams >= p.maxStreams {
+			next := conn.next
+			removeConn(conn)
+			addConnAfter(conn, sp.busy)
+			conn = next
+			continue
+		}
+		//  a idle conn
+		if conn.streams == 0 {
+			sp.idle--
+		}
+		//  a good conn
+		conn.streams++
+		p.Unlock()
+		return conn, nil
+	}
+	p.Unlock()
+
+	//  create new conn
+	cc, err := grpc.DialContext(dialCtx, addr, opts...)
+	if err != nil {
+		return nil, err
+	}
+	conn = &poolConn{cc, nil, addr, p, sp, 1, time.Now().Unix(), nil, nil, false}
+
+	//  add conn to streams pool
+	p.Lock()
+	if sp.count < p.size {
+		addConnAfter(conn, sp.head)
+	}
+	p.Unlock()
+
+	return conn, nil
+}
+
+func (p *pool) release(addr string, conn *poolConn, err error) {
+	p.Lock()
+	p, sp, created := conn.pool, conn.sp, conn.created
+	//  try to add conn
+	if !conn.in && sp.count < p.size {
+		addConnAfter(conn, sp.head)
+	}
+	if !conn.in {
+		p.Unlock()
+		conn.ClientConn.Close()
+		return
+	}
+	//  a busy conn
+	if conn.streams >= p.maxStreams {
+		removeConn(conn)
+		addConnAfter(conn, sp.head)
+	}
+	conn.streams--
+	//  if streams == 0, we can do something
+	if conn.streams == 0 {
+		//  1. it has errored
+		//  2. too many idle conn or
+		//  3. conn is too old
+		now := time.Now().Unix()
+		if err != nil || sp.idle >= p.maxIdle || now-created > p.ttl {
+			removeConn(conn)
+			p.Unlock()
+			conn.ClientConn.Close()
+			return
+		}
+		sp.idle++
+	}
+	p.Unlock()
+}
+
+func (conn *poolConn) Close() {
+	conn.pool.release(conn.addr, conn, conn.err)
+}
+
+func removeConn(conn *poolConn) {
+	if conn.pre != nil {
+		conn.pre.next = conn.next
+	}
+	if conn.next != nil {
+		conn.next.pre = conn.pre
+	}
+	conn.pre = nil
+	conn.next = nil
+	conn.in = false
+	conn.sp.count--
+	return
+}
+
+func addConnAfter(conn *poolConn, after *poolConn) {
+	conn.next = after.next
+	conn.pre = after
+	if after.next != nil {
+		after.next.pre = conn
+	}
+	after.next = conn
+	conn.in = true
+	conn.sp.count++
+	return
+}
--- a/client/grpc/grpc_pool_test.go
+++ b/client/grpc/grpc_pool_test.go
@ -0,0 +1,63 @@
+package grpc
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	pb "google.golang.org/grpc/examples/helloworld/helloworld"
+)
+
+func testPool(t *testing.T, size int, ttl time.Duration, idle int, ms int) {
+	// setup server
+	l, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Errorf("failed to listen: %v", err)
+	}
+	defer l.Close()
+
+	s := grpc.NewServer()
+	pb.RegisterGreeterServer(s, &greeterServer{})
+
+	go s.Serve(l)
+	defer s.Stop()
+
+	// zero pool
+	p := newPool(size, ttl, idle, ms)
+
+	for i := 0; i < 10; i++ {
+		// get a conn
+		cc, err := p.getConn(context.TODO(), l.Addr().String(), grpc.WithInsecure())
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		rsp := pb.HelloReply{}
+
+		err = cc.Invoke(context.TODO(), "/helloworld.Greeter/SayHello", &pb.HelloRequest{Name: "John"}, &rsp)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if rsp.Message != "Hello John" {
+			t.Errorf("Got unexpected response %v", rsp.Message)
+		}
+
+		// release the conn
+		p.release(l.Addr().String(), cc, nil)
+
+		p.Lock()
+		if i := p.conns[l.Addr().String()].count; i > size {
+			p.Unlock()
+			t.Errorf("pool size %d is greater than expected %d", i, size)
+		}
+		p.Unlock()
+	}
+}
+
+func TestGRPCPool(t *testing.T) {
+	testPool(t, 0, time.Minute, 10, 2)
+	testPool(t, 2, time.Minute, 10, 1)
+}
--- a/client/grpc/grpc_test.go
+++ b/client/grpc/grpc_test.go
@ -0,0 +1,112 @@
+package grpc
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"go-micro.dev/v5/client"
+	"go-micro.dev/v5/errors"
+	"go-micro.dev/v5/registry"
+	"go-micro.dev/v5/selector"
+	pgrpc "google.golang.org/grpc"
+	pb "google.golang.org/grpc/examples/helloworld/helloworld"
+)
+
+// server is used to implement helloworld.GreeterServer.
+type greeterServer struct {
+	pb.UnimplementedGreeterServer
+}
+
+// SayHello implements helloworld.GreeterServer.
+func (g *greeterServer) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
+	if in.Name == "Error" {
+		return nil, &errors.Error{Id: "1", Code: 99, Detail: "detail"}
+	}
+	return &pb.HelloReply{Message: "Hello " + in.Name}, nil
+}
+
+func TestGRPCClient(t *testing.T) {
+	l, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	defer l.Close()
+
+	s := pgrpc.NewServer()
+	pb.RegisterGreeterServer(s, &greeterServer{})
+
+	go s.Serve(l)
+	defer s.Stop()
+
+	// create mock registry
+	r := registry.NewMemoryRegistry()
+
+	// register service
+	r.Register(&registry.Service{
+		Name:    "helloworld",
+		Version: "test",
+		Nodes: []*registry.Node{
+			{
+				Id:      "test-1",
+				Address: l.Addr().String(),
+				Metadata: map[string]string{
+					"protocol": "grpc",
+				},
+			},
+		},
+	})
+
+	// create selector
+	se := selector.NewSelector(
+		selector.Registry(r),
+	)
+
+	// create client
+	c := NewClient(
+		client.Registry(r),
+		client.Selector(se),
+	)
+
+	testMethods := []string{
+		"/helloworld.Greeter/SayHello",
+		"Greeter.SayHello",
+	}
+
+	for _, method := range testMethods {
+		req := c.NewRequest("helloworld", method, &pb.HelloRequest{
+			Name: "John",
+		})
+
+		rsp := pb.HelloReply{}
+
+		err = c.Call(context.TODO(), req, &rsp)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if rsp.Message != "Hello John" {
+			t.Fatalf("Got unexpected response %v", rsp.Message)
+		}
+	}
+
+	req := c.NewRequest("helloworld", "/helloworld.Greeter/SayHello", &pb.HelloRequest{
+		Name: "Error",
+	})
+
+	rsp := pb.HelloReply{}
+
+	err = c.Call(context.TODO(), req, &rsp)
+	if err == nil {
+		t.Fatal("nil error received")
+	}
+
+	verr, ok := err.(*errors.Error)
+	if !ok {
+		t.Fatalf("invalid error received %#+v\n", err)
+	}
+
+	if verr.Code != 99 && verr.Id != "1" && verr.Detail != "detail" {
+		t.Fatalf("invalid error received %#+v\n", verr)
+	}
+}
--- a/client/grpc/message.go
+++ b/client/grpc/message.go
@ -0,0 +1,40 @@
+package grpc
+
+import (
+	"go-micro.dev/v5/client"
+)
+
+type grpcEvent struct {
+	topic       string
+	contentType string
+	payload     interface{}
+}
+
+func newGRPCEvent(topic string, payload interface{}, contentType string, opts ...client.MessageOption) client.Message {
+	var options client.MessageOptions
+	for _, o := range opts {
+		o(&options)
+	}
+
+	if len(options.ContentType) > 0 {
+		contentType = options.ContentType
+	}
+
+	return &grpcEvent{
+		payload:     payload,
+		topic:       topic,
+		contentType: contentType,
+	}
+}
+
+func (g *grpcEvent) ContentType() string {
+	return g.contentType
+}
+
+func (g *grpcEvent) Topic() string {
+	return g.topic
+}
+
+func (g *grpcEvent) Payload() interface{} {
+	return g.payload
+}
--- a/client/grpc/options.go
+++ b/client/grpc/options.go
@ -0,0 +1,143 @@
+// Package grpc provides a gRPC options
+package grpc
+
+import (
+	"context"
+	"crypto/tls"
+
+	"go-micro.dev/v5/client"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/encoding"
+)
+
+var (
+	// DefaultPoolMaxStreams maximum streams on a connectioin
+	// (20).
+	DefaultPoolMaxStreams = 20
+
+	// DefaultPoolMaxIdle maximum idle conns of a pool
+	// (50).
+	DefaultPoolMaxIdle = 50
+
+	// DefaultMaxRecvMsgSize maximum message that client can receive
+	// (4 MB).
+	DefaultMaxRecvMsgSize = 1024 * 1024 * 4
+
+	// DefaultMaxSendMsgSize maximum message that client can send
+	// (4 MB).
+	DefaultMaxSendMsgSize = 1024 * 1024 * 4
+)
+
+type poolMaxStreams struct{}
+type poolMaxIdle struct{}
+type codecsKey struct{}
+type tlsAuth struct{}
+type maxRecvMsgSizeKey struct{}
+type maxSendMsgSizeKey struct{}
+type grpcDialOptions struct{}
+type grpcCallOptions struct{}
+
+// maximum streams on a connectioin.
+func PoolMaxStreams(n int) client.Option {
+	return func(o *client.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, poolMaxStreams{}, n)
+	}
+}
+
+// maximum idle conns of a pool.
+func PoolMaxIdle(d int) client.Option {
+	return func(o *client.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, poolMaxIdle{}, d)
+	}
+}
+
+// gRPC Codec to be used to encode/decode requests for a given content type.
+func Codec(contentType string, c encoding.Codec) client.Option {
+	return func(o *client.Options) {
+		codecs := make(map[string]encoding.Codec)
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		if v := o.Context.Value(codecsKey{}); v != nil {
+			codecs = v.(map[string]encoding.Codec)
+		}
+		codecs[contentType] = c
+		o.Context = context.WithValue(o.Context, codecsKey{}, codecs)
+	}
+}
+
+// AuthTLS should be used to setup a secure authentication using TLS.
+func AuthTLS(t *tls.Config) client.Option {
+	return func(o *client.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, tlsAuth{}, t)
+	}
+}
+
+// MaxRecvMsgSize set the maximum size of message that client can receive.
+func MaxRecvMsgSize(s int) client.Option {
+	return func(o *client.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, maxRecvMsgSizeKey{}, s)
+	}
+}
+
+// MaxSendMsgSize set the maximum size of message that client can send.
+func MaxSendMsgSize(s int) client.Option {
+	return func(o *client.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, maxSendMsgSizeKey{}, s)
+	}
+}
+
+// DialOptions to be used to configure gRPC dial options.
+func DialOptions(opts ...grpc.DialOption) client.CallOption {
+	return func(o *client.CallOptions) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, grpcDialOptions{}, opts)
+	}
+}
+
+// CallOptions to be used to configure gRPC call options.
+func CallOptions(opts ...grpc.CallOption) client.CallOption {
+	return func(o *client.CallOptions) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, grpcCallOptions{}, opts)
+	}
+}
+
+func callOpts(opts client.CallOptions) []grpc.CallOption {
+	if opts.Context == nil {
+		return nil
+	}
+
+	v := opts.Context.Value(grpcCallOptions{})
+
+	if v == nil {
+		return nil
+	}
+
+	options, ok := v.([]grpc.CallOption)
+
+	if !ok {
+		return nil
+	}
+
+	return options
+}
--- a/client/grpc/request.go
+++ b/client/grpc/request.go
@ -0,0 +1,87 @@
+package grpc
+
+import (
+	"fmt"
+	"strings"
+
+	"go-micro.dev/v5/client"
+	"go-micro.dev/v5/codec"
+)
+
+type grpcRequest struct {
+	service     string
+	method      string
+	contentType string
+	request     interface{}
+	opts        client.RequestOptions
+	codec       codec.Codec
+}
+
+// service Struct.Method /service.Struct/Method.
+func methodToGRPC(service, method string) string {
+	// no method or already grpc method
+	if len(method) == 0 || method[0] == '/' {
+		return method
+	}
+
+	// assume method is Foo.Bar
+	mParts := strings.Split(method, ".")
+	if len(mParts) != 2 {
+		return method
+	}
+
+	if len(service) == 0 {
+		return fmt.Sprintf("/%s/%s", mParts[0], mParts[1])
+	}
+
+	// return /pkg.Foo/Bar
+	return fmt.Sprintf("/%s.%s/%s", service, mParts[0], mParts[1])
+}
+
+func newGRPCRequest(service, method string, request interface{}, contentType string, reqOpts ...client.RequestOption) client.Request {
+	var opts client.RequestOptions
+	for _, o := range reqOpts {
+		o(&opts)
+	}
+
+	// set the content-type specified
+	if len(opts.ContentType) > 0 {
+		contentType = opts.ContentType
+	}
+
+	return &grpcRequest{
+		service:     service,
+		method:      method,
+		request:     request,
+		contentType: contentType,
+		opts:        opts,
+	}
+}
+
+func (g *grpcRequest) ContentType() string {
+	return g.contentType
+}
+
+func (g *grpcRequest) Service() string {
+	return g.service
+}
+
+func (g *grpcRequest) Method() string {
+	return g.method
+}
+
+func (g *grpcRequest) Endpoint() string {
+	return g.method
+}
+
+func (g *grpcRequest) Codec() codec.Writer {
+	return g.codec
+}
+
+func (g *grpcRequest) Body() interface{} {
+	return g.request
+}
+
+func (g *grpcRequest) Stream() bool {
+	return g.opts.Stream
+}
--- a/client/grpc/request_test.go
+++ b/client/grpc/request_test.go
@ -0,0 +1,41 @@
+package grpc
+
+import (
+	"testing"
+)
+
+func TestMethodToGRPC(t *testing.T) {
+	testData := []struct {
+		service string
+		method  string
+		expect  string
+	}{
+		{
+			"helloworld",
+			"Greeter.SayHello",
+			"/helloworld.Greeter/SayHello",
+		},
+		{
+			"helloworld",
+			"/helloworld.Greeter/SayHello",
+			"/helloworld.Greeter/SayHello",
+		},
+		{
+			"",
+			"/helloworld.Greeter/SayHello",
+			"/helloworld.Greeter/SayHello",
+		},
+		{
+			"",
+			"Greeter.SayHello",
+			"/Greeter/SayHello",
+		},
+	}
+
+	for _, d := range testData {
+		method := methodToGRPC(d.service, d.method)
+		if method != d.expect {
+			t.Fatalf("expected %s got %s", d.expect, method)
+		}
+	}
+}
--- a/client/grpc/response.go
+++ b/client/grpc/response.go
@ -0,0 +1,44 @@
+package grpc
+
+import (
+	"strings"
+
+	"go-micro.dev/v5/codec"
+	"go-micro.dev/v5/codec/bytes"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/encoding"
+)
+
+type response struct {
+	conn   *grpc.ClientConn
+	stream grpc.ClientStream
+	codec  encoding.Codec
+	gcodec codec.Codec
+}
+
+// Read the response.
+func (r *response) Codec() codec.Reader {
+	return r.gcodec
+}
+
+// read the header.
+func (r *response) Header() map[string]string {
+	md, err := r.stream.Header()
+	if err != nil {
+		return map[string]string{}
+	}
+	hdr := make(map[string]string, len(md))
+	for k, v := range md {
+		hdr[k] = strings.Join(v, ",")
+	}
+	return hdr
+}
+
+// Read the undecoded response.
+func (r *response) Read() ([]byte, error) {
+	f := &bytes.Frame{}
+	if err := r.gcodec.ReadBody(f); err != nil {
+		return nil, err
+	}
+	return f.Data, nil
+}
--- a/client/grpc/stream.go
+++ b/client/grpc/stream.go
@ -0,0 +1,84 @@
+package grpc
+
+import (
+	"context"
+	"io"
+	"sync"
+
+	"go-micro.dev/v5/client"
+	"google.golang.org/grpc"
+)
+
+// Implements the streamer interface.
+type grpcStream struct {
+	sync.RWMutex
+	closed   bool
+	err      error
+	stream   grpc.ClientStream
+	request  client.Request
+	response client.Response
+	context  context.Context
+	cancel   func()
+	release  func(error)
+}
+
+func (g *grpcStream) Context() context.Context {
+	return g.context
+}
+
+func (g *grpcStream) Request() client.Request {
+	return g.request
+}
+
+func (g *grpcStream) Response() client.Response {
+	return g.response
+}
+
+func (g *grpcStream) Send(msg interface{}) error {
+	if err := g.stream.SendMsg(msg); err != nil {
+		g.setError(err)
+		return err
+	}
+	return nil
+}
+
+func (g *grpcStream) Recv(msg interface{}) (err error) {
+	if err = g.stream.RecvMsg(msg); err != nil {
+		if err != io.EOF {
+			g.setError(err)
+		}
+		return err
+	}
+	return
+}
+
+func (g *grpcStream) Error() error {
+	g.RLock()
+	defer g.RUnlock()
+	return g.err
+}
+
+func (g *grpcStream) setError(e error) {
+	g.Lock()
+	g.err = e
+	g.Unlock()
+}
+
+func (g *grpcStream) CloseSend() error {
+	return g.stream.CloseSend()
+}
+
+func (g *grpcStream) Close() error {
+	g.Lock()
+	defer g.Unlock()
+
+	if g.closed {
+		return nil
+	}
+	// cancel the context
+	g.cancel()
+	g.closed = true
+	// release back to pool
+	g.release(g.err)
+	return nil
+}