update

Author: chenjie199234

cgrpc/client.go

@@ -184,16 +184,12 @@ func (c *CGrpcClient) UpdateDiscovery(all map[string]*RegisterData) {
 	c.resolver.cc.UpdateState(s)
 }
 func (c *CGrpcClient) Call(ctx context.Context, path string, req interface{}, resp interface{}, metadata map[string]string) error {
-	start := time.Now()
 	if c.c.GlobalTimeout != 0 {
 		var cancel context.CancelFunc
-		ctx, cancel = context.WithDeadline(ctx, start.Add(c.c.GlobalTimeout))
+		ctx, cancel = context.WithDeadline(ctx, time.Now().Add(c.c.GlobalTimeout))
 		defer cancel()
 	}
 	dl, ok := ctx.Deadline()
-	if ok && dl.UnixNano() <= start.UnixNano()+int64(5*time.Millisecond) {
-		return cerror.ErrDeadlineExceeded
-	}
 	md := gmetadata.New(nil)
 	if len(metadata) != 0 {
 		d, _ := json.Marshal(metadata)
@@ -206,6 +202,11 @@ func (c *CGrpcClient) Call(ctx context.Context, path string, req interface{}, re
 	md.Set("core_target", c.serverappname)
 	ctx = gmetadata.NewOutgoingContext(ctx, md)
 	for {
+		start := time.Now()
+		if ok && dl.UnixNano() <= start.UnixNano()+int64(5*time.Millisecond) {
+			//at least 5ms for net lag and server logic
+			return cerror.ErrDeadlineExceeded
+		}
 		p := &peer.Peer{}
 		e := transGrpcError(c.conn.Invoke(ctx, path, req, resp, grpc.Peer(p)))
 		end := time.Now()

codegen/tml/dao/template_dao.go

@@ -59,7 +59,7 @@ func NewApi() error {
 	//if e != nil {
 	// 	return e
 	//}
-	//ExampleWebApi = example.NewExampleWebClient(exampleweb)
+	//ExampleWebApi = example.NewExampleWebClient(exampleweb,"http://examplehost:exampleport")
 
 	return nil
 }
@@ -158,36 +158,6 @@ func getWebClientConfig() *web.ClientConfig {
 		MaxHeader:     1024,
 		SocketRBuf:    2048,
 		SocketWBuf:    2048,
-		Discover:      webDNS,
-	}
-}
-
-func webDNS(group, name string, manually <-chan *struct{}, client *web.WebClient) {
-	tker := time.NewTicker(time.Second * 10)
-	for{
-		select {
-		case <-tker.C:
-		case <-manually:
-			tker.Reset(time.Second * 10)
-		}
-		result := make(map[string]*web.RegisterData)
-		addrs, e := net.LookupHost(name + "-service-headless" + "." + group)
-		if e != nil {
-			log.Error(nil,"[web.dns] get:", name+"-service-headless", "addrs error:", e)
-			continue
-		}
-		for i := range addrs {
-			addrs[i] = addrs[i] + ":8000"
-		}
-		dserver := make(map[string]struct{})
-		dserver["dns"] = struct{}{}
-		for _, addr := range addrs {
-			result[addr] = &web.RegisterData{DServers: dserver}
-		}
-		for len(tker.C) > 0 {
-			<-tker.C
-		}
-		client.UpdateDiscovery(result)
 	}
 }`
 

codegen/tml/gomod/template_gomod.go

@@ -11,8 +11,8 @@ const text = `module {{.}}
 go 1.17
 
 require (
-	github.com/chenjie199234/Config v0.0.25
-	github.com/chenjie199234/Corelib v0.0.39
+	github.com/chenjie199234/Config v0.0.26
+	github.com/chenjie199234/Corelib v0.0.40
 	github.com/fsnotify/fsnotify v1.5.1
 	github.com/go-sql-driver/mysql v1.6.0
 	github.com/segmentio/kafka-go v0.4.25

crpc/client.go

@@ -107,10 +107,10 @@ func NewCrpcClient(c *ClientConfig, selfgroup, selfname, servergroup, servername
 		for _, cert := range c.CAs {
 			certPEM, e := os.ReadFile(cert)
 			if e != nil {
-				return nil, errors.New("[web.client] read cert file:" + cert + " error:" + e.Error())
+				return nil, errors.New("[crpc.client] read cert file:" + cert + " error:" + e.Error())
 			}
 			if !certpool.AppendCertsFromPEM(certPEM) {
-				return nil, errors.New("[web.client] load cert file:" + cert + " error:" + e.Error())
+				return nil, errors.New("[crpc.client] load cert file:" + cert + " error:" + e.Error())
 			}
 		}
 	}
@@ -238,26 +238,12 @@ func (c *CrpcClient) offlinefunc(p *stream.Peer) {
 var errPickAgain = errors.New("[crpc.client] picked server closed")
 
 func (c *CrpcClient) Call(ctx context.Context, path string, in []byte, metadata map[string]string) ([]byte, error) {
-	start := time.Now()
-	if c.c.GlobalTimeout != 0 {
-		var cancel context.CancelFunc
-		ctx, cancel = context.WithDeadline(ctx, start.Add(c.c.GlobalTimeout))
-		defer cancel()
-	}
-	dl, ok := ctx.Deadline()
-	if ok && dl.UnixNano() <= start.UnixNano()+int64(5*time.Millisecond) {
-		return nil, cerror.ErrDeadlineExceeded
-	}
 	msg := &Msg{
-		//Callid:   atomic.AddUint64(&c.callid, 1),
 		Type:     MsgType_CALL,
 		Path:     path,
 		Body:     in,
 		Metadata: metadata,
 	}
-	if ok {
-		msg.Deadline = dl.UnixNano()
-	}
 	traceid, _, _, selfmethod, selfpath, selfdeep := trace.GetTrace(ctx)
 	if traceid != "" {
 		msg.Tracedata = map[string]string{
@@ -267,8 +253,22 @@ func (c *CrpcClient) Call(ctx context.Context, path string, in []byte, metadata
 			"Deep":         strconv.Itoa(selfdeep),
 		}
 	}
+	if c.c.GlobalTimeout != 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithDeadline(ctx, time.Now().Add(c.c.GlobalTimeout))
+		defer cancel()
+	}
+	dl, ok := ctx.Deadline()
+	if ok {
+		msg.Deadline = dl.UnixNano()
+	}
 	r := c.getreq(msg)
 	for {
+		start := time.Now()
+		if ok && dl.UnixNano() <= start.UnixNano()+int64(5*time.Millisecond) {
+			//at least 5ms for net lag and server logic
+			return nil, cerror.ErrDeadlineExceeded
+		}
 		server, e := c.balancer.Pick(ctx)
 		if e != nil {
 			return nil, e