unistack-org/micro-client-grpc
4
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 35 Wiki Activity

issue_132 #134

Merged
vtolstov merged 4 commits from devstigneev/micro-client-grpc:issue_132 into v3 2024-02-29 23:07:34 +03:00
Conversation 0 Commits 4 Files Changed 3 +14 -14
2 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 2a6a93a792 - Show all commits

14
grpc.go
View File

@ -30,7 +30,7 @@ const (
) )
type grpcClient struct { type grpcClient struct {
pool *pool pool *ConnPool
opts client.Options opts client.Options
sync.RWMutex sync.RWMutex
init bool init bool
@ -130,13 +130,13 @@ func (g *grpcClient) call(ctx context.Context, addr string, req client.Request,
grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer)) 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 { if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
} }
defer func() { defer func() {
// defer execution of release // defer execution of release
g.pool.release(cc, grr) g.pool.Put(cc, grr)
}() }()
ch := make(chan error, 1) ch := make(chan error, 1)
@ -239,7 +239,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
grpcDialOptions = append(grpcDialOptions, grpc.WithContextDialer(contextDialer)) 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 { if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
} }
@ -272,7 +272,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
// cancel the context // cancel the context
cancel() cancel()
// release the connection // release the connection
g.pool.release(cc, err) g.pool.Put(cc, err)
// now return the error // now return the error
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error creating stream: %v", err))
} }
@ -300,7 +300,7 @@ func (g *grpcClient) stream(ctx context.Context, addr string, req client.Request
} }
// defer execution of release // defer execution of release
g.pool.release(cc, err) g.pool.Put(cc, err)
}, },
} }
@ -825,7 +825,7 @@ func NewClient(opts ...client.Option) client.Client {
opts: options, 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) c := client.Client(rc)

14
grpc_pool.go
View File

@ -10,7 +10,7 @@ import (
"google.golang.org/grpc/connectivity" "google.golang.org/grpc/connectivity"
) )
type pool struct { type ConnPool struct {
conns map[string]*streamsPool conns map[string]*streamsPool
size int size int
ttl int64 ttl int64
@ -34,7 +34,7 @@ type poolConn struct {
err error err error
*grpc.ClientConn *grpc.ClientConn
next *poolConn next *poolConn
pool *pool pool *ConnPool
sp *streamsPool sp *streamsPool
pre *poolConn pre *poolConn
addr string addr string
@ -43,14 +43,14 @@ type poolConn struct {
in bool 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 { if ms <= 0 {
ms = 1 ms = 1
} }
if idle < 0 { if idle < 0 {
idle = 0 idle = 0
} }
return &pool{ return &ConnPool{
size: size, size: size,
ttl: int64(ttl.Seconds()), ttl: int64(ttl.Seconds()),
maxStreams: ms, 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") { if strings.HasPrefix(addr, "http") {
addr = addr[strings.Index(addr, ":")+3:] 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 return conn, nil
} }
func (p *pool) release(conn *poolConn, err error) { func (p *ConnPool) Put(conn *poolConn, err error) {
p.Lock() p.Lock()
p, sp, created := conn.pool, conn.sp, conn.created p, sp, created := conn.pool, conn.sp, conn.created
// try to add conn // try to add conn
@ -183,7 +183,7 @@ func (p *pool) release(conn *poolConn, err error) {
} }
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) {