export grpc conn pool

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

go.mod

@@ -1,8 +1,17 @@
 module go.unistack.org/micro-client-grpc/v3
 
-go 1.16
+go 1.20
 
 require (
-	go.unistack.org/micro/v3 v3.10.22
+	go.unistack.org/micro/v3 v3.10.28
-	google.golang.org/grpc v1.52.3
+	google.golang.org/grpc v1.59.0
+)
+
+require (
+	github.com/golang/protobuf v1.5.3 // indirect
+	golang.org/x/net v0.17.0 // indirect
+	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/text v0.13.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // indirect
+	google.golang.org/protobuf v1.31.0 // indirect
 )

grpc.go

@@ -30,7 +30,7 @@ const (
 )
 
 type grpcClient struct {
-	pool *pool
+	pool *ConnPool
 	opts client.Options
 	sync.RWMutex
 	init bool
@@ -118,6 +118,9 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
 		),
 	}
 
+	if opts := g.getGrpcDialOptions(g.opts.Context); opts != nil {
+		grpcDialOptions = append(grpcDialOptions, opts...)
+	}
 	if opts := g.getGrpcDialOptions(opts.Context); opts != nil {
 		grpcDialOptions = append(grpcDialOptions, opts...)
 	}
@@ -130,13 +133,13 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
 		grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer))
 	}
 
-	cc, err := g.pool.getConn(dialCtx, addr, grpcDialOptions...)
+	cc, err := g.pool.Get(dialCtx, addr, 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(cc, grr)
+		g.pool.Put(cc, grr)
 	}()
 
 	ch := make(chan error, 1)
@@ -239,7 +242,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
 		grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer))
 	}
 
-	cc, err := g.pool.getConn(dialCtx, addr, grpcDialOptions...)
+	cc, err := g.pool.Get(dialCtx, addr, grpcDialOptions...)
 	if err != nil {
 		return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
 	}
@@ -272,7 +275,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
 		// cancel the context
 		cancel()
 		// release the connection
-		g.pool.release(cc, err)
+		g.pool.Put(cc, err)
 		// now return the error
 		return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err))
 	}
@@ -300,7 +303,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
 			}
 
 			// defer execution of release
-			g.pool.release(cc, err)
+			g.pool.Put(cc, err)
 		},
 	}
 
@@ -715,6 +718,10 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
 	if v, ok := os.LookupEnv("MICRO_PROXY"); ok {
 		exchange = v
 	}
+	// get the exchange
+	if len(options.Exchange) > 0 {
+		exchange = options.Exchange
+	}
 
 	msgs := make([]*broker.Message, 0, len(ps))
 
@@ -726,6 +733,16 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
 	for _, p := range ps {
 		md := metadata.Copy(omd)
 		md[metadata.HeaderContentType] = p.ContentType()
+		topic := p.Topic()
+		if len(exchange) > 0 {
+			topic = exchange
+		}
+		md.Set(metadata.HeaderTopic, topic)
+		iter := p.Metadata().Iterator()
+		var k, v string
+		for iter.Next(&k, &v) {
+			md.Set(k, v)
+		}
 
 		// passed in raw data
 		if d, ok := p.Payload().(*codec.Frame); ok {
@@ -744,15 +761,6 @@ func (g *grpcClient) publish(ctx context.Context, ps []client.Message, opts ...c
 			body = b
 		}
 
-		topic := p.Topic()
-		if len(exchange) > 0 {
-			topic = exchange
-		}
-
-		for k, v := range p.Metadata() {
-			md.Set(k, v)
-		}
-		md.Set(metadata.HeaderTopic, topic)
 		msgs = append(msgs, &broker.Message{Header: md, Body: body})
 	}
 
@@ -825,7 +833,7 @@ func NewClient(opts ...client.Option) client.Client {
 		opts: options,
 	}
 
-	rc.pool = newPool(options.PoolSize, options.PoolTTL, rc.poolMaxIdle(), rc.poolMaxStreams())
+	rc.pool = NewConnPool(options.PoolSize, options.PoolTTL, rc.poolMaxIdle(), rc.poolMaxStreams())
 
 	c := client.Client(rc)
 

grpc_pool.go

@@ -10,7 +10,7 @@ import (
 	"google.golang.org/grpc/connectivity"
 )
 
-type pool struct {
+type ConnPool struct {
 	conns      map[string]*streamsPool
 	size       int
 	ttl        int64
@@ -34,7 +34,7 @@ type poolConn struct {
 	err error
 	*grpc.ClientConn
 	next    *poolConn
-	pool    *pool
+	pool    *ConnPool
 	sp      *streamsPool
 	pre     *poolConn
 	addr    string
@@ -43,14 +43,14 @@ type poolConn struct {
 	in      bool
 }
 
-func newPool(size int, ttl time.Duration, idle int, ms int) *pool {
+func NewConnPool(size int, ttl time.Duration, idle int, ms int) *ConnPool {
 	if ms <= 0 {
 		ms = 1
 	}
 	if idle < 0 {
 		idle = 0
 	}
-	return &pool{
+	return &ConnPool{
 		size:       size,
 		ttl:        int64(ttl.Seconds()),
 		maxStreams: ms,
@@ -59,7 +59,7 @@ func newPool(size int, ttl time.Duration, idle int, ms int) *pool {
 	}
 }
 
-func (p *pool) getConn(ctx context.Context, addr string, opts ...grpc.DialOption) (*poolConn, error) {
+func (p *ConnPool) Get(ctx context.Context, addr string, opts ...grpc.DialOption) (*poolConn, error) {
 	if strings.HasPrefix(addr, "http") {
 		addr = addr[strings.Index(addr, ":")+3:]
 	}
@@ -147,7 +147,7 @@ func (p *pool) getConn(ctx context.Context, addr string, opts ...grpc.DialOption
 	return conn, nil
 }
 
-func (p *pool) release(conn *poolConn, err error) {
+func (p *ConnPool) Put(conn *poolConn, err error) {
 	p.Lock()
 	p, sp, created := conn.pool, conn.sp, conn.created
 	//  try to add conn
@@ -183,7 +183,7 @@ func (p *pool) release(conn *poolConn, err error) {
 }
 
 func (conn *poolConn) Close() {
-	conn.pool.release(conn, conn.err)
+	conn.pool.Put(conn, conn.err)
 }
 
 func removeConn(conn *poolConn) {

options.go

@@ -95,8 +95,8 @@ func MaxSendMsgSize(s int) client.Option {
 type grpcDialOptions struct{}
 
 // DialOptions to be used to configure gRPC dial options
-func DialOptions(opts ...grpc.DialOption) client.CallOption {
+func DialOptions(opts ...grpc.DialOption) client.Option {
-	return func(o *client.CallOptions) {
+	return func(o *client.Options) {
 		if o.Context == nil {
 			o.Context = context.Background()
 		}