Merge pull request #1147 from cloudwego/release/v0.7.3

chore: release v0.7.3

client/client.go

@@ -361,7 +361,13 @@ func (kc *kClient) Call(ctx context.Context, method string, request, response in
 			recycleRI = true
 		}
 	} else {
-		ri, recycleRI, err = kc.opt.RetryContainer.WithRetryIfNeeded(ctx, callOptRetry, kc.rpcCallWithRetry(ri, method, request, response), ri, request)
+		var lastRI rpcinfo.RPCInfo
+		lastRI, recycleRI, err = kc.opt.RetryContainer.WithRetryIfNeeded(ctx, callOptRetry, kc.rpcCallWithRetry(ri, method, request, response), ri, request)
+		if ri != lastRI {
+			// reset ri of ctx to lastRI
+			ctx = rpcinfo.NewCtxWithRPCInfo(ctx, lastRI)
+		}
+		ri = lastRI
 	}
 
 	// do fallback if with setup

client/service_inline.go

@@ -140,10 +140,16 @@ func (kc *serviceInlineClient) Call(ctx context.Context, method string, request,
 	var reportErr error
 	defer func() {
 		if panicInfo := recover(); panicInfo != nil {
-			reportErr = rpcinfo.ClientPanicToErr(ctx, panicInfo, ri, false)
+			reportErr = rpcinfo.ClientPanicToErr(ctx, panicInfo, ri, true)
 		}
 		kc.opt.TracerCtl.DoFinish(ctx, ri, reportErr)
-		rpcinfo.PutRPCInfo(ri)
+		// If the user start a new goroutine and return before endpoint finished, it may cause panic.
+		// For example,, if the user writes a timeout Middleware and times out, rpcinfo will be recycled,
+		// but in fact, rpcinfo is still being used when it is executed inside
+		// So if endpoint returns err, client won't recycle rpcinfo.
+		if reportErr == nil {
+			rpcinfo.PutRPCInfo(ri)
+		}
 		callOpts.Recycle()
 	}()
 	reportErr = kc.eps(ctx, request, response)

go.mod

@@ -13,8 +13,8 @@ require (
 	github.com/cloudwego/fastpb v0.0.4
 	github.com/cloudwego/frugal v0.1.8
 	github.com/cloudwego/localsession v0.0.2
-	github.com/cloudwego/netpoll v0.5.0
-	github.com/cloudwego/thriftgo v0.3.0
+	github.com/cloudwego/netpoll v0.5.1
+	github.com/cloudwego/thriftgo v0.3.2-0.20230828085742-edaddf2c17af
 	github.com/golang/mock v1.6.0
 	github.com/google/pprof v0.0.0-20220608213341-c488b8fa1db3
 	github.com/jhump/protoreflect v1.8.2

go.sum

@@ -62,18 +62,14 @@ github.com/cloudwego/localsession v0.0.2/go.mod h1:kiJxmvAcy4PLgKtEnPS5AXed3xCiX
 github.com/cloudwego/netpoll v0.2.4/go.mod h1:1T2WVuQ+MQw6h6DpE45MohSvDTKdy2DlzCx2KsnPI4E=
 github.com/cloudwego/netpoll v0.3.1/go.mod h1:1T2WVuQ+MQw6h6DpE45MohSvDTKdy2DlzCx2KsnPI4E=
 github.com/cloudwego/netpoll v0.4.0/go.mod h1:xVefXptcyheopwNDZjDPcfU6kIjZXZ4nY550k1yH9eQ=
-github.com/cloudwego/netpoll v0.4.2-0.20230913081710-1a27688e2033 h1:/VYzCYH+Brp8CW1u475U+mPS7lHv5ulKx0vFJbp3YZ0=
-github.com/cloudwego/netpoll v0.4.2-0.20230913081710-1a27688e2033/go.mod h1:xVefXptcyheopwNDZjDPcfU6kIjZXZ4nY550k1yH9eQ=
-github.com/cloudwego/netpoll v0.4.2-0.20230918061532-5719b5310f34 h1:AbZPQaXr7MzOiUf1OZauww5rjmBpeLlyhM+hD7UsCn8=
-github.com/cloudwego/netpoll v0.4.2-0.20230918061532-5719b5310f34/go.mod h1:xVefXptcyheopwNDZjDPcfU6kIjZXZ4nY550k1yH9eQ=
-github.com/cloudwego/netpoll v0.5.0 h1:oRrOp58cPCvK2QbMozZNDESvrxQaEHW2dCimmwH1lcU=
-github.com/cloudwego/netpoll v0.5.0/go.mod h1:xVefXptcyheopwNDZjDPcfU6kIjZXZ4nY550k1yH9eQ=
+github.com/cloudwego/netpoll v0.5.1 h1:zDUF7xF0C97I10fGlQFJ4jg65khZZMUvSu/TWX44Ohc=
+github.com/cloudwego/netpoll v0.5.1/go.mod h1:xVefXptcyheopwNDZjDPcfU6kIjZXZ4nY550k1yH9eQ=
 github.com/cloudwego/thriftgo v0.1.2/go.mod h1:LzeafuLSiHA9JTiWC8TIMIq64iadeObgRUhmVG1OC/w=
 github.com/cloudwego/thriftgo v0.2.4/go.mod h1:8i9AF5uDdWHGqzUhXDlubCjx4MEfKvWXGQlMWyR0tM4=
 github.com/cloudwego/thriftgo v0.2.7/go.mod h1:8i9AF5uDdWHGqzUhXDlubCjx4MEfKvWXGQlMWyR0tM4=
 github.com/cloudwego/thriftgo v0.2.11/go.mod h1:dAyXHEmKXo0LfMCrblVEY3mUZsdeuA5+i0vF5f09j7E=
-github.com/cloudwego/thriftgo v0.3.0 h1:BBb9hVcqmu9p4iKUP/PSIaDB21Vfutgd7k2zgK37Q9Q=
-github.com/cloudwego/thriftgo v0.3.0/go.mod h1:AvH0iEjvKHu3cdxG7JvhSAaffkS4h2f4/ZxpJbm48W4=
+github.com/cloudwego/thriftgo v0.3.2-0.20230828085742-edaddf2c17af h1:xsNmlAdSnh6zuovEON4Ab0iT+fTfQUWqZ50tk+6OGW8=
+github.com/cloudwego/thriftgo v0.3.2-0.20230828085742-edaddf2c17af/go.mod h1:AvH0iEjvKHu3cdxG7JvhSAaffkS4h2f4/ZxpJbm48W4=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

pkg/remote/compression.go

@@ -31,26 +31,30 @@ const (
 )
 
 func SetRecvCompressor(ri rpcinfo.RPCInfo, compressorName string) {
-	if ri == nil {
+	if ri == nil || compressorName == "" {
 		return
 	}
-	rpcinfo.AsMutableEndpointInfo(ri.From()).SetTag("recv-compressor", compressorName)
+	if v, ok := ri.Invocation().(rpcinfo.InvocationSetter); ok {
+		v.SetExtra("recv-compressor", compressorName)
+	}
 }
 
 func SetSendCompressor(ri rpcinfo.RPCInfo, compressorName string) {
-	if ri == nil {
+	if ri == nil || compressorName == "" {
 		return
 	}
-	rpcinfo.AsMutableEndpointInfo(ri.From()).SetTag("send-compressor", compressorName)
+	if v, ok := ri.Invocation().(rpcinfo.InvocationSetter); ok {
+		v.SetExtra("send-compressor", compressorName)
+	}
 }
 
 func GetSendCompressor(ri rpcinfo.RPCInfo) string {
 	if ri == nil {
 		return ""
 	}
-	v, exist := ri.From().Tag("send-compressor")
-	if exist {
-		return v
+	v := ri.Invocation().Extra("send-compressor")
+	if name, ok := v.(string); ok {
+		return name
 	}
 	return ""
 }
@@ -59,9 +63,9 @@ func GetRecvCompressor(ri rpcinfo.RPCInfo) string {
 	if ri == nil {
 		return ""
 	}
-	v, exist := ri.From().Tag("recv-compressor")
-	if exist {
-		return v
+	v := ri.Invocation().Extra("recv-compressor")
+	if name, ok := v.(string); ok {
+		return name
 	}
 	return ""
 }

pkg/rpcinfo/interface.go

@@ -83,6 +83,7 @@ type Invocation interface {
 	MethodName() string
 	SeqID() int32
 	BizStatusErr() kerrors.BizStatusErrorIface
+	Extra(key string) interface{}
 }
 
 // RPCInfo is the core abstraction of information about an RPC in Kitex.

pkg/rpcinfo/invocation.go

@@ -41,15 +41,16 @@ type InvocationSetter interface {
 	SetMethodName(name string)
 	SetSeqID(seqID int32)
 	SetBizStatusErr(err kerrors.BizStatusErrorIface)
+	SetExtra(key string, value interface{})
 	Reset()
 }
-
 type invocation struct {
 	packageName string
 	serviceName string
 	methodName  string
 	seqID       int32
 	bizErr      kerrors.BizStatusErrorIface
+	extra       map[string]interface{}
 }
 
 // NewInvocation creates a new Invocation with the given service, method and optional package.
@@ -130,6 +131,20 @@ func (i *invocation) SetBizStatusErr(err kerrors.BizStatusErrorIface) {
 	i.bizErr = err
 }
 
+func (i *invocation) SetExtra(key string, value interface{}) {
+	if i.extra == nil {
+		i.extra = map[string]interface{}{}
+	}
+	i.extra[key] = value
+}
+
+func (i *invocation) Extra(key string) interface{} {
+	if i.extra == nil {
+		return nil
+	}
+	return i.extra[key]
+}
+
 // Reset implements the InvocationSetter interface.
 func (i *invocation) Reset() {
 	i.zero()
@@ -147,4 +162,7 @@ func (i *invocation) zero() {
 	i.serviceName = ""
 	i.methodName = ""
 	i.bizErr = nil
+	for key := range i.extra {
+		delete(i.extra, key)
+	}
 }

pkg/serviceinfo/serviceinfo.go

@@ -27,6 +27,7 @@ type PayloadCodec int
 const (
 	Thrift PayloadCodec = iota
 	Protobuf
+	Hessian2
 )
 
 const (

tool/cmd/kitex/args/args.go

@@ -114,6 +114,8 @@ func (a *Arguments) buildFlags(version string) *flag.FlagSet {
 		"Specify a code gen path.")
 	f.BoolVar(&a.DeepCopyAPI, "deep-copy-api", false,
 		"Generate codes with injecting deep copy method.")
+	f.StringVar(&a.Protocol, "protocol", "",
+		"Specify a protocol for codec")
 	a.RecordCmd = os.Args
 	a.Version = version
 	a.ThriftOptions = append(a.ThriftOptions,
@@ -298,12 +300,16 @@ func (a *Arguments) BuildCmd(out io.Writer) *exec.Cmd {
 		Stdout: io.MultiWriter(out, os.Stdout),
 		Stderr: io.MultiWriter(out, os.Stderr),
 	}
+
 	if a.IDLType == "thrift" {
 		os.Setenv(EnvPluginMode, thriftgo.PluginName)
 		cmd.Args = append(cmd.Args, "thriftgo")
 		for _, inc := range a.Includes {
 			cmd.Args = append(cmd.Args, "-i", inc)
 		}
+		if strings.EqualFold(a.Protocol, "hessian2") {
+			a.ThriftOptions = append(a.ThriftOptions, "template=slim")
+		}
 		a.ThriftOptions = append(a.ThriftOptions, "package_prefix="+a.PackagePrefix)
 		gas := "go:" + strings.Join(a.ThriftOptions, ",")
 		if a.Verbose {

tool/internal_pkg/generator/generator.go

@@ -27,6 +27,7 @@ import (
 	"github.com/cloudwego/kitex/tool/internal_pkg/log"
 	"github.com/cloudwego/kitex/tool/internal_pkg/tpl"
 	"github.com/cloudwego/kitex/tool/internal_pkg/util"
+	"github.com/cloudwego/kitex/transport"
 )
 
 // Constants .
@@ -132,6 +133,7 @@ type Config struct {
 	GenPath string
 
 	DeepCopyAPI bool
+	Protocol    string
 }
 
 // Pack packs the Config into a slice of "key=val" strings.
@@ -456,6 +458,9 @@ func (g *generator) updatePackageInfo(pkg *PackageInfo) {
 	pkg.ExternalKitexGen = g.Use
 	pkg.FrugalPretouch = g.FrugalPretouch
 	pkg.Module = g.ModuleName
+	if strings.EqualFold(g.Protocol, transport.HESSIAN2.String()) {
+		pkg.Protocol = transport.HESSIAN2
+	}
 	if pkg.Dependencies == nil {
 		pkg.Dependencies = make(map[string]string)
 	}
@@ -481,10 +486,10 @@ func (g *generator) setImports(name string, pkg *PackageInfo) {
 		}
 		fallthrough
 	case HandlerFileName:
-		if len(pkg.AllMethods()) > 0 {
-			pkg.AddImports("context")
-		}
 		for _, m := range pkg.ServiceInfo.AllMethods() {
+			if !m.ServerStreaming && !m.ClientStreaming {
+				pkg.AddImports("context")
+			}
 			for _, a := range m.Args {
 				for _, dep := range a.Deps {
 					pkg.AddImport(dep.PkgRefName, dep.ImportPath)

tool/internal_pkg/generator/generator_test.go

@@ -53,6 +53,7 @@ func TestConfig_Pack(t *testing.T) {
 		RecordCmd             string
 		ThriftPluginTimeLimit time.Duration
 		TemplateDir           string
+		Protocol              string
 	}
 	tests := []struct {
 		name    string
@@ -63,7 +64,7 @@ func TestConfig_Pack(t *testing.T) {
 		{
 			name:    "some",
 			fields:  fields{Features: []feature{feature(999)}, ThriftPluginTimeLimit: 30 * time.Second},
-			wantRes: []string{"Verbose=false", "GenerateMain=false", "GenerateInvoker=false", "Version=", "NoFastAPI=false", "ModuleName=", "ServiceName=", "Use=", "IDLType=", "Includes=", "ThriftOptions=", "ProtobufOptions=", "IDL=", "OutputPath=", "PackagePrefix=", "CombineService=false", "CopyIDL=false", "ProtobufPlugins=", "Features=999", "FrugalPretouch=false", "ThriftPluginTimeLimit=30s", "ExtensionFile=", "Record=false", "RecordCmd=", "TemplateDir=", "GenPath=", "DeepCopyAPI=false"},
+			wantRes: []string{"Verbose=false", "GenerateMain=false", "GenerateInvoker=false", "Version=", "NoFastAPI=false", "ModuleName=", "ServiceName=", "Use=", "IDLType=", "Includes=", "ThriftOptions=", "ProtobufOptions=", "IDL=", "OutputPath=", "PackagePrefix=", "CombineService=false", "CopyIDL=false", "ProtobufPlugins=", "Features=999", "FrugalPretouch=false", "ThriftPluginTimeLimit=30s", "ExtensionFile=", "Record=false", "RecordCmd=", "TemplateDir=", "GenPath=", "DeepCopyAPI=false", "Protocol="},
 		},
 	}
 	for _, tt := range tests {
@@ -91,6 +92,7 @@ func TestConfig_Pack(t *testing.T) {
 				FrugalPretouch:        tt.fields.FrugalPretouch,
 				ThriftPluginTimeLimit: tt.fields.ThriftPluginTimeLimit,
 				TemplateDir:           tt.fields.TemplateDir,
+				Protocol:              tt.fields.Protocol,
 			}
 			if gotRes := c.Pack(); !reflect.DeepEqual(gotRes, tt.wantRes) {
 				t.Errorf("Config.Pack() = \n%v\nwant\n%v", gotRes, tt.wantRes)
@@ -121,6 +123,7 @@ func TestConfig_Unpack(t *testing.T) {
 		Features        []feature
 		FrugalPretouch  bool
 		TemplateDir     string
+		Protocol        string
 	}
 	type args struct {
 		args []string
@@ -162,6 +165,7 @@ func TestConfig_Unpack(t *testing.T) {
 				Features:        tt.fields.Features,
 				FrugalPretouch:  tt.fields.FrugalPretouch,
 				TemplateDir:     tt.fields.TemplateDir,
+				Protocol:        tt.fields.Protocol,
 			}
 			if err := c.Unpack(tt.args.args); (err != nil) != tt.wantErr {
 				t.Errorf("Config.Unpack() error = %v, wantErr %v", err, tt.wantErr)

tool/internal_pkg/generator/type.go

@@ -21,6 +21,7 @@ import (
 	"text/template"
 
 	"github.com/cloudwego/kitex/tool/internal_pkg/util"
+	"github.com/cloudwego/kitex/transport"
 )
 
 // File .
@@ -45,6 +46,7 @@ type PackageInfo struct {
 	Features         []feature
 	FrugalPretouch   bool
 	Module           string
+	Protocol         transport.Protocol
 }
 
 // AddImport .
@@ -97,6 +99,7 @@ type ServiceInfo struct {
 	CombineServices []*ServiceInfo
 	HasStreaming    bool
 	ServiceFilePath string
+	Protocol        string
 }
 
 // AllMethods returns all methods that the service have.

tool/internal_pkg/pluginmode/thriftgo/convertor.go

@@ -25,6 +25,7 @@ import (
 	"strings"
 
 	"github.com/cloudwego/kitex/tool/internal_pkg/util"
+	"github.com/cloudwego/kitex/transport"
 
 	"github.com/cloudwego/thriftgo/generator/backend"
 	"github.com/cloudwego/thriftgo/generator/golang"
@@ -188,10 +189,9 @@ func (c *converter) copyTypeWithRef(t *parser.Type, ref string) (res *parser.Typ
 	default:
 		if strings.Contains(t.Name, ".") {
 			return &parser.Type{
-				Name:        t.Name,
-				KeyType:     t.KeyType,
-				ValueType:   t.ValueType,
-				Annotations: t.Annotations,
+				Name:      t.Name,
+				KeyType:   t.KeyType,
+				ValueType: t.ValueType,
 			}
 		}
 		return &parser.Type{
@@ -353,8 +353,10 @@ func (c *converter) convertTypes(req *plugin.Request) error {
 		}
 		// combine service
 		if ast == req.AST && c.Config.CombineService && len(ast.Services) > 0 {
-			var svcs []*generator.ServiceInfo
-			var methods []*generator.MethodInfo
+			var (
+				svcs    []*generator.ServiceInfo
+				methods []*generator.MethodInfo
+			)
 			for _, s := range all[ast.Filename] {
 				svcs = append(svcs, s)
 				methods = append(methods, s.AllMethods()...)
@@ -376,6 +378,11 @@ func (c *converter) convertTypes(req *plugin.Request) error {
 				Methods:         methods,
 				ServiceFilePath: ast.Filename,
 			}
+
+			if c.IsHessian2() {
+				si.Protocol = transport.HESSIAN2.String()
+			}
+
 			si.ServiceTypeName = func() string { return si.ServiceName }
 			all[ast.Filename] = append(all[ast.Filename], si)
 			c.svc2ast[si] = ast
@@ -404,6 +411,10 @@ func (c *converter) makeService(pkg generator.PkgInfo, svc *golang.Service) (*ge
 		}
 		si.Methods = append(si.Methods, mi)
 	}
+
+	if c.IsHessian2() {
+		si.Protocol = transport.HESSIAN2.String()
+	}
 	return si, nil
 }
 
@@ -484,3 +495,7 @@ func (c *converter) getCombineServiceName(name string, svcs []*generator.Service
 	}
 	return name
 }
+
+func (c *converter) IsHessian2() bool {
+	return strings.EqualFold(c.Config.Protocol, transport.HESSIAN2.String())
+}