export grpc pool conn

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
@@ -21,36 +21,36 @@ type pool struct {
 
 type streamsPool struct {
 	//  head of list
-	head *poolConn
+	head *PoolConn
 	//  busy conns list
-	busy *poolConn
+	busy *PoolConn
 	//  the siza of list
 	count int
 	//  idle conn
 	idle int
 }
 
-type poolConn struct {
+type PoolConn struct {
 	err error
 	*grpc.ClientConn
-	next    *poolConn
+	next    *PoolConn
-	pool    *pool
+	pool    *ConnPool
 	sp      *streamsPool
-	pre     *poolConn
+	pre     *PoolConn
 	addr    string
 	streams int
 	created int64
 	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:]
 	}
@@ -67,7 +67,7 @@ func (p *pool) getConn(ctx context.Context, addr string, opts ...grpc.DialOption
 	p.Lock()
 	sp, ok := p.conns[addr]
 	if !ok {
-		sp = &streamsPool{head: &poolConn{}, busy: &poolConn{}, count: 0, idle: 0}
+		sp = &streamsPool{head: &PoolConn{}, busy: &PoolConn{}, count: 0, idle: 0}
 		p.conns[addr] = sp
 	}
 	//  while we have conns check streams and then return one
@@ -135,7 +135,7 @@ func (p *pool) getConn(ctx context.Context, addr string, opts ...grpc.DialOption
 	if err != nil {
 		return nil, err
 	}
-	conn = &poolConn{ClientConn: cc, err: nil, addr: addr, pool: p, sp: sp, streams: 1, created: time.Now().Unix(), pre: nil, next: nil, in: false}
+	conn = &PoolConn{ClientConn: cc, err: nil, addr: addr, pool: p, sp: sp, streams: 1, created: time.Now().Unix(), pre: nil, next: nil, in: false}
 
 	//  add conn to streams pool
 	p.Lock()
@@ -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
@@ -182,11 +182,11 @@ func (p *pool) release(conn *poolConn, err error) {
 	p.Unlock()
 }
 
-func (conn *poolConn) Close() {
+func (conn *PoolConn) Close() {
-	conn.pool.release(conn, conn.err)
+	conn.pool.Put(conn, conn.err)
 }
 
-func removeConn(conn *poolConn) {
+func removeConn(conn *PoolConn) {
 	if conn.pre != nil {
 		conn.pre.next = conn.next
 	}
@@ -199,7 +199,7 @@ func removeConn(conn *poolConn) {
 	conn.sp.count--
 }
 
-func addConnAfter(conn *poolConn, after *poolConn) {
+func addConnAfter(conn *PoolConn, after *PoolConn) {
 	conn.next = after.next
 	conn.pre = after
 	if after.next != nil {

options.go

@@ -10,7 +10,7 @@ import (
 )
 
 var (
-	// DefaultPoolMaxStreams maximum streams on a connectioin
+	// DefaultPoolMaxStreams maximum streams on a connection
 	// (20)
 	DefaultPoolMaxStreams = 20
 
@@ -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()
 		}

response.go

@@ -9,7 +9,7 @@ import (
 )
 
 type response struct {
-	conn   *poolConn
+	conn   *PoolConn
 	stream grpc.ClientStream
 	codec  codec.Codec
 }

stream.go

@@ -17,7 +17,7 @@ type grpcStream struct {
 	request  client.Request
 	response client.Response
 	close    func(err error)
-	conn     *poolConn
+	conn     *PoolConn
 	sync.RWMutex
 	closed bool
 }