grpc重写插件完成

.goreleaser.yaml

@@ -36,6 +36,8 @@ builds:
       - -X "github.com/eolinker/apinto/utils/version.buildUser=goreleaser"
       - -X "github.com/eolinker/apinto/utils/version.goVersion={{.Env.GOVERSION}}"
       - -X "github.com/eolinker/apinto/utils/version.eoscVersion={{.Env.EoscVersion}}"
+    env:
+      - CGO_ENABLED=0
 archives:
   - id: default
     format: tar.gz

drivers/plugins/grpc-proxy-rewrite/driver.go

@@ -21,10 +21,11 @@ func Create(id, name string, conf *Config, workers map[eosc.RequireId]eosc.IWork
 
 	pw := &ProxyRewrite{
 		WorkerBase:      drivers.Worker(id, name),
-		host:            strings.TrimSpace(conf.Authority),
 		headers:         conf.Headers,
 		tls:             conf.Tls,
 		skipCertificate: conf.SkipCertificate,
+		service:         conf.Service,
+		method:          conf.Method,
 	}
 
 	return pw, nil

drivers/plugins/grpc-proxy-rewrite/proxy-rewrite.go

@@ -1,8 +1,6 @@
 package grpc_proxy_rewrite
 
 import (
-	"strings"
-
 	grpc_context "github.com/eolinker/eosc/eocontext/grpc-context"
 
 	"github.com/eolinker/apinto/drivers"
@@ -21,8 +19,8 @@ var (
 
 type ProxyRewrite struct {
 	drivers.WorkerBase
-
-	host            string
+	service         string
+	method          string
 	headers         map[string]string
 	tls             bool
 	skipCertificate bool
@@ -33,8 +31,12 @@ func (p *ProxyRewrite) DoFilter(ctx eocontext.EoContext, next eocontext.IChain)
 }
 
 func (p *ProxyRewrite) DoGrpcFilter(ctx grpc_context.IGrpcContext, next eocontext.IChain) (err error) {
-	if p.host != "" {
-		ctx.Proxy().SetHost(p.host)
+
+	if p.service != "" {
+		ctx.Proxy().SetService(p.service)
+	}
+	if p.method != "" {
+		ctx.Proxy().SetMethod(p.method)
 	}
 	ctx.EnableTls(p.tls)
 	ctx.InsecureCertificateVerify(p.skipCertificate)
@@ -59,7 +61,8 @@ func (p *ProxyRewrite) Reset(v interface{}, workers map[eosc.RequireId]eosc.IWor
 	p.skipCertificate = conf.SkipCertificate
 	p.headers = conf.Headers
 	p.tls = conf.Tls
-	p.host = strings.TrimSpace(conf.Authority)
+	p.service = conf.Service
+	p.method = conf.Method
 	return nil
 }
 

node/grpc-context/client.go

@@ -1,6 +1,7 @@
 package grpc_context
 
 import (
+	"fmt"
 	"sync"
 	"time"
 )
@@ -9,24 +10,17 @@ var _ IClient = (*Client)(nil)
 
 var (
 	clientPool IClient = NewClient()
-	defaultClientOption         = ClientOption{
-		ClientPoolConnSize: defaultClientPoolConnsSizeCap,
-		DialTimeOut:        defaultDialTimeout,
-		KeepAlive:          defaultKeepAlive,
-		KeepAliveTimeout:   defaultKeepAliveTimeout,
-	}
 )
 
 type IClient interface {
-	Get(target string, isTls bool) (IClientPool, bool)
-	Set(target string, isTls bool, pool IClientPool)
-	Del(target string, isTls bool)
+	Get(target string, isTls bool, host ...string) IClientPool
 	Close()
 }
 
 type Client struct {
 	clients    map[string]IClientPool
 	tlsClients map[string]IClientPool
+	stop       bool
 	locker     sync.RWMutex
 }
 
@@ -44,6 +38,10 @@ func (c *Client) clean() {
 	sleep := time.Second * 10
 	for {
 		c.locker.Lock()
+		if c.stop {
+			c.locker.Unlock()
+			return
+		}
 		for key, client := range c.clients {
 			if client.ConnCount() < 1 {
 				delete(c.clients, key)
@@ -59,33 +57,41 @@ func (c *Client) clean() {
 	}
 }
 
-func (c *Client) Get(target string, isTls bool) (IClientPool, bool) {
+func (c *Client) Get(target string, isTls bool, host ...string) IClientPool {
+	key := target
+	authority := ""
+	if len(host) > 0 && host[0] != "" {
+		key = fmt.Sprintf("%s|%s", target, host[0])
+		authority = host[0]
+	}
 	c.locker.RLock()
-	defer c.locker.RUnlock()
 	clients := c.clients
 	if isTls {
 		clients = c.tlsClients
 	}
-	client, ok := clients[target]
-
-	return client, ok
-}
-
-func (c *Client) Set(target string, isTls bool, pool IClientPool) {
-	c.locker.Lock()
-	defer c.locker.Unlock()
-	c.del(target, isTls)
-	clients := c.clients
-	if isTls {
-		clients = c.tlsClients
+	client, ok := clients[key]
+	c.locker.RUnlock()
+	if ok {
+		return client
 	}
-	clients[target] = pool
-}
-
-func (c *Client) Del(target string, isTls bool) {
 	c.locker.Lock()
 	defer c.locker.Unlock()
-	c.del(target, isTls)
+	client, ok = clients[key]
+	if ok {
+		return client
+	}
+	p := NewClientPoolWithOption(target, &ClientOption{
+		ClientPoolConnSize: defaultClientPoolConnsSizeCap,
+		DialTimeOut:        defaultDialTimeout,
+		KeepAlive:          defaultKeepAlive,
+		KeepAliveTimeout:   defaultKeepAliveTimeout,
+		IsTls:              isTls,
+		Authority:          authority,
+	})
+
+	clients[key] = p
+
+	return p
 }
 
 func (c *Client) del(target string, isTls bool) {
@@ -111,6 +117,7 @@ func (c *Client) Close() {
 	for _, client := range tlsClients {
 		client.Close()
 	}
+	c.stop = true
 	c.clients = nil
 	c.tlsClients = nil
 }

node/grpc-context/context.go

@@ -41,8 +41,8 @@ type Context struct {
 	srv                       interface{}
 	acceptTime                time.Time
 	requestId                 string
-	request                   *Request
-	proxy                     *Request
+	request                   grpc_context.IRequest
+	proxy                     grpc_context.IRequest
 	response                  grpc_context.IResponse
 	completeHandler           eocontext.CompleteHandler
 	finishHandler             eocontext.FinishHandler
@@ -199,19 +199,18 @@ func (c *Context) SetResponse(response grpc_context.IResponse) {
 }
 
 func (c *Context) Invoke(address string, timeout time.Duration) error {
-	clientConn, err := c.dial(address)
+	passHost, targetHost := c.GetUpstreamHostHandler().PassHost()
+	switch passHost {
+	case eocontext.NodeHost:
+		c.proxy.SetHost(address)
+	case eocontext.ReWriteHost:
+		c.proxy.SetHost(targetHost)
+	}
+	clientConn, err := clientPool.Get(address, c.tls, c.proxy.Host()).Get()
 	if err != nil {
 		return err
 	}
-	passHost, targetHost := c.GetUpstreamHostHandler().PassHost()
-	switch passHost {
-	case eocontext.PassHost:
 
-	case eocontext.NodeHost:
-		c.proxy.Headers().Set(":authority", address)
-	case eocontext.ReWriteHost:
-		c.proxy.Headers().Set(":authority", targetHost)
-	}
 	c.proxy.Headers().Set("grpc-timeout", fmt.Sprintf("%dn", timeout))
 	clientCtx, _ := context.WithCancel(metadata.NewOutgoingContext(c.Context(), c.proxy.Headers().Copy()))
 	clientStream, err := grpc.NewClientStream(clientCtx, clientStreamDescForProxying, clientConn, c.proxy.FullMethodName())
@@ -255,11 +254,31 @@ func (c *Context) reset() {
 	pool.Put(c)
 }
 
-func (c *Context) dial(address string) (*grpc.ClientConn, error) {
-	p, has := clientPool.Get(address, c.tls)
-	if !has {
-		p = NewClientPoolWithOption(address, &defaultClientOption)
-		defer clientPool.Set(address, c.tls, p)
-	}
-	return p.Get()
-}
+//func (c *Context) dial(address string, timeout time.Duration) (*grpc.ClientConn, error) {
+//	return clientPool.Get(address, c.tls, c.proxy.Host()).Get()
+//
+//	//opts := make([]grpc.DialOption, 0, 5)
+//	//if c.tls {
+//	//	opts = append(opts, grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: c.insecureCertificateVerify})))
+//	//} else {
+//	//	opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+//	//}
+//	//opts = append(opts,
+//	//	//grpc.WithBlock(),
+//	//	grpc.WithKeepaliveParams(keepalive.ClientParameters{
+//	//		Time:    defaultKeepAlive,
+//	//		Timeout: defaultKeepAliveTimeout,
+//	//	}),
+//	//)
+//	//
+//	//if c.proxy.Host() != "" {
+//	//	opts = append(opts, grpc.WithAuthority(c.proxy.Host()))
+//	//}
+//	////ctx, cancel := context.WithTimeout(context.TODO(), timeout)
+//	////defer cancel()
+//	//conn, err := grpc.Dial(address, opts...)
+//	//if err != nil {
+//	//	return nil, err
+//	//}
+//	//return conn, nil
+//}

node/grpc-context/pool.go

@@ -29,6 +29,7 @@ var (
 type ClientOption struct {
 	ClientPoolConnSize int
 	IsTls              bool
+	Authority          string
 	DialTimeOut        time.Duration
 	KeepAlive          time.Duration
 	KeepAliveTimeout   time.Duration
@@ -120,18 +121,17 @@ func (cc *ClientPool) connect() (*grpc.ClientConn, error) {
 		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	}
 	opts = append(opts,
-		grpc.WithBlock(),
+		//grpc.WithBlock(),
 		grpc.WithKeepaliveParams(keepalive.ClientParameters{
 			Time:    cc.option.KeepAlive,
 			Timeout: cc.option.KeepAliveTimeout,
 		}),
 	)
-	conn, err := grpc.DialContext(ctx, cc.target, opts...)
-	if err != nil {
-		return nil, err
+	if cc.option.Authority != "" {
+		opts = append(opts, grpc.WithAuthority(cc.option.Authority))
 	}
 
-	return conn, nil
+	return grpc.DialContext(ctx, cc.target, opts...)
 }
 
 func (cc *ClientPool) Close() {