refactor: invocation interface

cluster/cluster/adaptivesvc/cluster_invoker.go

@@ -36,6 +36,8 @@ import (
  "dubbo.apache.org/dubbo-go/v3/protocol"
 )
 
+var _ protocol.Invoker = (*adaptiveServiceClusterInvoker)(nil)
+
 type adaptiveServiceClusterInvoker struct {
  base.BaseClusterInvoker
 }

cluster/cluster/zoneaware/cluster_interceptor.go

@@ -38,11 +38,11 @@ func (z *interceptor) Invoke(ctx context.Context, invoker protocol.Invoker, invo
  switch value := force.(type) {
  case bool:
  if value {
- invocation.SetAttachments(key, "true")
+ invocation.SetAttachment(key, "true")
  }
  case string:
  if "true" == value {
- invocation.SetAttachments(key, "true")
+ invocation.SetAttachment(key, "true")
  }
  default:
  // ignore

cluster/cluster/zoneaware/cluster_invoker.go

@@ -65,15 +65,15 @@ func (invoker *zoneawareClusterInvoker) Invoke(ctx context.Context, invocation p
 
  // providers in the registry with the same zone
  key := constant.RegistryKey + "." + constant.RegistryZoneKey
- zone := invocation.AttachmentsByKey(key, "")
+ zone := invocation.GetAttachmentWithDefaultValue(key, "")
  if "" != zone {
  for _, invoker := range invokers {
  if invoker.IsAvailable() && matchParam(zone, key, "", invoker) {
  return invoker.Invoke(ctx, invocation)
  }
  }
 
- force := invocation.AttachmentsByKey(constant.RegistryKey+"."+constant.RegistryZoneForceKey, "")
+ force := invocation.GetAttachmentWithDefaultValue(constant.RegistryKey+"."+constant.RegistryZoneForceKey, "")
  if "true" == force {
  return &protocol.RPCResult{
  Err: fmt.Errorf("no registry instance in zone or "+

cluster/cluster/zoneaware/cluster_invoker_test.go

@@ -177,7 +177,7 @@ func TestZoneWareInvokerWithZoneSuccess(t *testing.T) {
  inv := &invocation.RPCInvocation{}
  // zone hangzhou
  hz := zoneArray[0]
- inv.SetAttachments(constant.RegistryKey+"."+constant.RegistryZoneKey, hz)
+ inv.SetAttachment(constant.RegistryKey+"."+constant.RegistryZoneKey, hz)
 
  result := clusterInvoker.Invoke(context.Background(), inv)
 
@@ -206,9 +206,9 @@ func TestZoneWareInvokerWithZoneForceFail(t *testing.T) {
 
  inv := &invocation.RPCInvocation{}
  // zone hangzhou
- inv.SetAttachments(constant.RegistryKey+"."+constant.RegistryZoneKey, "hangzhou")
+ inv.SetAttachment(constant.RegistryKey+"."+constant.RegistryZoneKey, "hangzhou")
  // zone force
- inv.SetAttachments(constant.RegistryKey+"."+constant.RegistryZoneForceKey, "true")
+ inv.SetAttachment(constant.RegistryKey+"."+constant.RegistryZoneForceKey, "true")
 
  result := clusterInvoker.Invoke(context.Background(), inv)
 

common/proxy/proxy.go

@@ -181,19 +181,19 @@ func DefaultProxyImplementFunc(p *Proxy, v common.RPCService) {
  }
 
  for k, value := range p.attachments {
- inv.SetAttachments(k, value)
+ inv.SetAttachment(k, value)
  }
 
  // add user setAttachment. It is compatibility with previous versions.
  atm := invCtx.Value(constant.AttachmentKey)
  if m, ok := atm.(map[string]string); ok {
  for k, value := range m {
- inv.SetAttachments(k, value)
+ inv.SetAttachment(k, value)
  }
  } else if m2, ok2 := atm.(map[string]interface{}); ok2 {
  // it is support to transfer map[string]interface{}. It refers to dubbo-java 2.7.
  for k, value := range m2 {
- inv.SetAttachments(k, value)
+ inv.SetAttachment(k, value)
  }
  }
 

common/proxy/proxy_factory/pass_through.go

@@ -102,7 +102,7 @@ func (pi *PassThroughProxyInvoker) Invoke(ctx context.Context, invocation protoc
  in := make([]reflect.Value, 5)
  in = append(in, srv.Rcvr())
  in = append(in, reflect.ValueOf(invocation.MethodName()))
- in = append(in, reflect.ValueOf(invocation.Attachment(constant.ParamsTypeKey)))
+ in = append(in, reflect.ValueOf(invocation.GetAttachmentInterface(constant.ParamsTypeKey)))
  in = append(in, reflect.ValueOf(args))
  in = append(in, reflect.ValueOf(invocation.Attachments()))
 

filter/active/filter.go

@@ -59,14 +59,14 @@ func newActiveFilter() filter.Filter {
 
 // Invoke starts to record the requests status
 func (f *activeFilter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
- invocation.(*invocation2.RPCInvocation).SetAttachments(dubboInvokeStartTime, strconv.FormatInt(protocol.CurrentTimeMillis(), 10))
+ invocation.(*invocation2.RPCInvocation).SetAttachment(dubboInvokeStartTime, strconv.FormatInt(protocol.CurrentTimeMillis(), 10))
  protocol.BeginCount(invoker.GetURL(), invocation.MethodName())
  return invoker.Invoke(ctx, invocation)
 }
 
 // OnResponse update the active count base on the request result.
 func (f *activeFilter) OnResponse(ctx context.Context, result protocol.Result, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
- startTime, err := strconv.ParseInt(invocation.(*invocation2.RPCInvocation).AttachmentsByKey(dubboInvokeStartTime, "0"), 10, 64)
+ startTime, err := strconv.ParseInt(invocation.(*invocation2.RPCInvocation).GetAttachmentWithDefaultValue(dubboInvokeStartTime, "0"), 10, 64)
  if err != nil {
  result.SetError(err)
  logger.Errorf("parse dubbo_invoke_start_time to int64 failed")

filter/active/filter_test.go

@@ -47,7 +47,7 @@ func TestFilterInvoke(t *testing.T) {
  invoker.EXPECT().Invoke(gomock.Any()).Return(nil)
  invoker.EXPECT().GetUrl().Return(url).Times(1)
  filter.Invoke(context.Background(), invoker, invoc)
- assert.True(t, invoc.AttachmentsByKey(dubboInvokeStartTime, "") != "")
+ assert.True(t, invoc.GetAttachmentWithDefaultValue(dubboInvokeStartTime, "") != "")
 }
 
 func TestFilterOnResponse(t *testing.T) {

filter/adaptivesvc/filter.go

@@ -91,7 +91,7 @@ func (f *adaptiveServiceProviderFilter) Invoke(ctx context.Context, invoker prot
 func (f *adaptiveServiceProviderFilter) OnResponse(_ context.Context, result protocol.Result, invoker protocol.Invoker,
  invocation protocol.Invocation) protocol.Result {
  // get updater from the attributes
- updaterIface := invocation.AttributeByKey(constant.AdaptiveServiceUpdaterKey, nil)
+ updaterIface, _ := invocation.GetAttribute(constant.AdaptiveServiceUpdaterKey)
  if updaterIface == nil {
  return &protocol.RPCResult{Err: ErrUpdaterNotFound}
  }

filter/auth/default_authenticator.go

@@ -69,10 +69,10 @@ func (authenticator *defaultAuthenticator) Sign(invocation protocol.Invocation,
  if err != nil {
  return err
  }
- inv.SetAttachments(constant.RequestSignatureKey, signature)
- inv.SetAttachments(constant.RequestTimestampKey, currentTimeMillis)
- inv.SetAttachments(constant.AKKey, accessKeyPair.AccessKey)
- inv.SetAttachments(constant.Consumer, consumer)
+ inv.SetAttachment(constant.RequestSignatureKey, signature)
+ inv.SetAttachment(constant.RequestTimestampKey, currentTimeMillis)
+ inv.SetAttachment(constant.AKKey, accessKeyPair.AccessKey)
+ inv.SetAttachment(constant.Consumer, consumer)
  return nil
 }
 
@@ -97,11 +97,11 @@ func getSignature(url *common.URL, invocation protocol.Invocation, secrectKey st
 
 // Authenticate verifies whether the signature sent by the requester is correct
 func (authenticator *defaultAuthenticator) Authenticate(invocation protocol.Invocation, url *common.URL) error {
- accessKeyId := invocation.AttachmentsByKey(constant.AKKey, "")
+ accessKeyId := invocation.GetAttachmentWithDefaultValue(constant.AKKey, "")
 
- requestTimestamp := invocation.AttachmentsByKey(constant.RequestTimestampKey, "")
- originSignature := invocation.AttachmentsByKey(constant.RequestSignatureKey, "")
- consumer := invocation.AttachmentsByKey(constant.Consumer, "")
+ requestTimestamp := invocation.GetAttachmentWithDefaultValue(constant.RequestTimestampKey, "")
+ originSignature := invocation.GetAttachmentWithDefaultValue(constant.RequestSignatureKey, "")
+ consumer := invocation.GetAttachmentWithDefaultValue(constant.Consumer, "")
  if IsEmpty(accessKeyId, false) || IsEmpty(consumer, false) ||
  IsEmpty(requestTimestamp, false) || IsEmpty(originSignature, false) {
  return errors.New("failed to authenticate your ak/sk, maybe the consumer has not enabled the auth")

filter/auth/default_authenticator_test.go

@@ -79,10 +79,10 @@ func TestDefaultAuthenticator_Sign(t *testing.T) {
  testurl.SetParam(constant.ParameterSignatureEnableKey, "false")
  inv := invocation.NewRPCInvocation("test", []interface{}{"OK"}, nil)
  _ = authenticator.Sign(inv, testurl)
- assert.NotEqual(t, inv.AttachmentsByKey(constant.RequestSignatureKey, ""), "")
- assert.NotEqual(t, inv.AttachmentsByKey(constant.Consumer, ""), "")
- assert.NotEqual(t, inv.AttachmentsByKey(constant.RequestTimestampKey, ""), "")
- assert.Equal(t, inv.AttachmentsByKey(constant.AKKey, ""), "akey")
+ assert.NotEqual(t, inv.GetAttachmentWithDefaultValue(constant.RequestSignatureKey, ""), "")
+ assert.NotEqual(t, inv.GetAttachmentWithDefaultValue(constant.Consumer, ""), "")
+ assert.NotEqual(t, inv.GetAttachmentWithDefaultValue(constant.RequestTimestampKey, ""), "")
+ assert.Equal(t, inv.GetAttachmentWithDefaultValue(constant.AKKey, ""), "akey")
 }
 
 func Test_getAccessKeyPairSuccess(t *testing.T) {

filter/generic/filter.go

@@ -67,7 +67,7 @@ func (f *genericFilter) Invoke(ctx context.Context, invoker protocol.Invoker, in
  args := make([]hessian.Object, 0, len(oldargs))
 
  // get generic info from attachments of invocation, the default value is "true"
- generic := invocation.AttachmentsByKey(constant.GenericKey, constant.GenericSerializationDefault)
+ generic := invocation.GetAttachmentWithDefaultValue(constant.GenericKey, constant.GenericSerializationDefault)
  // get generalizer according to value in the `generic`
  g := getGeneralizer(generic)
 

filter/generic/filter_test.go

@@ -60,7 +60,7 @@ func TestFilter_Invoke(t *testing.T) {
  assert.Equal(t, "Hello", args[0])
  assert.Equal(t, "java.lang.String", args[1].([]string)[0])
  assert.Equal(t, "arg1", args[2].([]hessian.Object)[0].(string))
- assert.Equal(t, constant.GenericSerializationDefault, invocation.AttachmentsByKey(constant.GenericKey, ""))
+ assert.Equal(t, constant.GenericSerializationDefault, invocation.GetAttachmentWithDefaultValue(constant.GenericKey, ""))
  return &protocol.RPCResult{}
  })
 
@@ -93,7 +93,7 @@ func TestFilter_InvokeWithGenericCall(t *testing.T) {
  assert.Equal(t, "hello", args[0])
  assert.Equal(t, "java.lang.String", args[1].([]string)[0])
  assert.Equal(t, "arg1", args[2].([]string)[0])
- assert.Equal(t, constant.GenericSerializationDefault, invocation.AttachmentsByKey(constant.GenericKey, ""))
+ assert.Equal(t, constant.GenericSerializationDefault, invocation.GetAttachmentWithDefaultValue(constant.GenericKey, ""))
  return &protocol.RPCResult{}
  })
 

filter/generic/service_filter.go

@@ -87,7 +87,7 @@ func (f *genericServiceFilter) Invoke(ctx context.Context, invoker protocol.Invo
  argsType := method.ArgsType()
 
  // get generic info from attachments of invocation, the default value is "true"
- generic := invocation.AttachmentsByKey(constant.GenericKey, constant.GenericSerializationDefault)
+ generic := invocation.GetAttachmentWithDefaultValue(constant.GenericKey, constant.GenericSerializationDefault)
  // get generalizer according to value in the `generic`
  g := getGeneralizer(generic)
 
@@ -126,7 +126,7 @@ func (f *genericServiceFilter) Invoke(ctx context.Context, invoker protocol.Invo
 func (f *genericServiceFilter) OnResponse(_ context.Context, result protocol.Result, _ protocol.Invoker, invocation protocol.Invocation) protocol.Result {
  if invocation.IsGenericInvocation() && result.Result() != nil {
  // get generic info from attachments of invocation, the default value is "true"
- generic := invocation.AttachmentsByKey(constant.GenericKey, constant.GenericSerializationDefault)
+ generic := invocation.GetAttachmentWithDefaultValue(constant.GenericKey, constant.GenericSerializationDefault)
  // get generalizer according to value in the `generic`
  g := getGeneralizer(generic)
 

filter/seata/filter.go

@@ -58,7 +58,7 @@ func newSeataFilter() filter.Filter {
 
 // Invoke Get Xid by attachment key `SEATA_XID`. When use Seata, transfer xid by attachments
 func (f *seataFilter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
- xid := invocation.AttachmentsByKey(string(SEATA_XID), "")
+ xid := invocation.GetAttachmentWithDefaultValue(string(SEATA_XID), "")
  if len(strings.TrimSpace(xid)) > 0 {
  logger.Debugf("Method: %v,Xid: %v", invocation.MethodName(), xid)
  return invoker.Invoke(context.WithValue(ctx, SEATA_XID, xid), invocation)

protocol/dubbo/dubbo_codec.go

@@ -65,20 +65,20 @@ func (c *DubboCodec) EncodeRequest(request *remoting.Request) (*bytes.Buffer, er
  invocation := *invoc
 
  svc := impl.Service{}
- svc.Path = invocation.AttachmentsByKey(constant.PathKey, "")
- svc.Interface = invocation.AttachmentsByKey(constant.InterfaceKey, "")
- svc.Version = invocation.AttachmentsByKey(constant.VersionKey, "")
- svc.Group = invocation.AttachmentsByKey(constant.GroupKey, "")
+ svc.Path = invocation.GetAttachmentWithDefaultValue(constant.PathKey, "")
+ svc.Interface = invocation.GetAttachmentWithDefaultValue(constant.InterfaceKey, "")
+ svc.Version = invocation.GetAttachmentWithDefaultValue(constant.VersionKey, "")
+ svc.Group = invocation.GetAttachmentWithDefaultValue(constant.GroupKey, "")
  svc.Method = invocation.MethodName()
- timeout, err := strconv.Atoi(invocation.AttachmentsByKey(constant.TimeoutKey, strconv.Itoa(constant.DefaultRemotingTimeout)))
+ timeout, err := strconv.Atoi(invocation.GetAttachmentWithDefaultValue(constant.TimeoutKey, strconv.Itoa(constant.DefaultRemotingTimeout)))
  if err != nil {
  // it will be wrapped in readwrite.Write .
  return nil, perrors.WithStack(err)
  }
  svc.Timeout = time.Duration(timeout)
 
  header := impl.DubboHeader{}
- serialization := invocation.AttachmentsByKey(constant.SerializationKey, constant.Hessian2Serialization)
+ serialization := invocation.GetAttachmentWithDefaultValue(constant.SerializationKey, constant.Hessian2Serialization)
  if serialization == constant.ProtobufSerialization {
  header.SerialID = constant.SProto
  } else {
@@ -254,20 +254,20 @@ func (c *DubboCodec) decodeResponse(data []byte) (*remoting.Response, int, error
  Status: pkg.Header.ResponseStatus,
  Event: (pkg.Header.Type & impl.PackageHeartbeat) != 0,
  }
- var error error
+ var pkgerr error
  if pkg.Header.Type&impl.PackageHeartbeat != 0x00 {
  if pkg.Header.Type&impl.PackageResponse != 0x00 {
  logger.Debugf("get rpc heartbeat response{header: %#v, body: %#v}", pkg.Header, pkg.Body)
  if pkg.Err != nil {
  logger.Errorf("rpc heartbeat response{error: %#v}", pkg.Err)
- error = pkg.Err
+ pkgerr = pkg.Err
  }
  } else {
  logger.Debugf("get rpc heartbeat request{header: %#v, service: %#v, body: %#v}", pkg.Header, pkg.Service, pkg.Body)
  response.Status = hessian.Response_OK
  // reply(session, p, hessian.PackageHeartbeat)
  }
- return response, hessian.HEADER_LENGTH + pkg.Header.BodyLen, error
+ return response, hessian.HEADER_LENGTH + pkg.Header.BodyLen, pkgerr
  }
  logger.Debugf("get rpc response{header: %#v, body: %#v}", pkg.Header, pkg.Body)
  rpcResult := &protocol.RPCResult{}

protocol/dubbo/dubbo_invoker.go

@@ -117,10 +117,10 @@ func (di *DubboInvoker) Invoke(ctx context.Context, invocation protocol.Invocati
 
  inv := invocation.(*invocation_impl.RPCInvocation)
  // init param
- inv.SetAttachments(constant.PathKey, di.GetURL().GetParam(constant.InterfaceKey, ""))
+ inv.SetAttachment(constant.PathKey, di.GetURL().GetParam(constant.InterfaceKey, ""))
  for _, k := range attachmentKey {
  if v := di.GetURL().GetParam(k, ""); len(v) > 0 {
- inv.SetAttachments(k, v)
+ inv.SetAttachment(k, v)
  }
  }
 
@@ -133,7 +133,7 @@ func (di *DubboInvoker) Invoke(ctx context.Context, invocation protocol.Invocati
  url.SetParam(constant.SerializationKey, constant.Hessian2Serialization)
  }
  // async
- async, err := strconv.ParseBool(inv.AttachmentsByKey(constant.AsyncKey, "false"))
+ async, err := strconv.ParseBool(inv.GetAttachmentWithDefaultValue(constant.AsyncKey, "false"))
  if err != nil {
  logger.Errorf("ParseBool - error: %v", err)
  async = false
@@ -172,12 +172,12 @@ func (di *DubboInvoker) getTimeout(invocation *invocation_impl.RPCInvocation) ti
  if len(timeout) != 0 {
  if t, err := time.ParseDuration(timeout); err == nil {
  // config timeout into attachment
- invocation.SetAttachments(constant.TimeoutKey, strconv.Itoa(int(t.Milliseconds())))
+ invocation.SetAttachment(constant.TimeoutKey, strconv.Itoa(int(t.Milliseconds())))
  return t
  }
  }
  // set timeout into invocation at method level
- invocation.SetAttachments(constant.TimeoutKey, strconv.Itoa(int(di.timeout.Milliseconds())))
+ invocation.SetAttachment(constant.TimeoutKey, strconv.Itoa(int(di.timeout.Milliseconds())))
  return di.timeout
 }
 

protocol/dubbo/dubbo_invoker_test.go

@@ -63,7 +63,7 @@ package dubbo
 // assert.Equal(t, User{ID: "1", Name: "username"}, *res.Result().(*User))
 //
 // // CallOneway
-// inv.SetAttachments(constant.ASYNC_KEY, "true")
+// inv.SetAttachment(constant.ASYNC_KEY, "true")
 // res = invoker.Invoke(context.Background(), inv)
 // assert.NoError(t, res.Error())
 //
@@ -81,7 +81,7 @@ package dubbo
 // assert.NoError(t, res.Error())
 //
 // // Err_No_Reply
-// inv.SetAttachments(constant.ASYNC_KEY, "false")
+// inv.SetAttachment(constant.ASYNC_KEY, "false")
 // inv.SetReply(nil)
 // res = invoker.Invoke(context.Background(), inv)
 // assert.EqualError(t, res.Error(), "request need @response")

protocol/dubbo3/dubbo3_invoker.go

@@ -205,12 +205,12 @@ func (di *DubboInvoker) getTimeout(invocation *invocation_impl.RPCInvocation) ti
  if len(timeout) != 0 {
  if t, err := time.ParseDuration(timeout); err == nil {
  // config timeout into attachment
- invocation.SetAttachments(constant.TimeoutKey, strconv.Itoa(int(t.Milliseconds())))
+ invocation.SetAttachment(constant.TimeoutKey, strconv.Itoa(int(t.Milliseconds())))
  return t
  }
  }
  // set timeout into invocation at method level
- invocation.SetAttachments(constant.TimeoutKey, strconv.Itoa(int(di.timeout.Milliseconds())))
+ invocation.SetAttachment(constant.TimeoutKey, strconv.Itoa(int(di.timeout.Milliseconds())))
  return di.timeout
 }
 

protocol/invocation.go

@@ -38,18 +38,23 @@ type Invocation interface {
  // Reply gets response of request
  Reply() interface{}
  // Attachments gets all attachments
- Attachments() map[string]interface{}
- // AttachmentsByKey gets attachment by key , if nil then return default value. （It will be deprecated in the future）
- AttachmentsByKey(string, string) string
- Attachment(string) interface{}
- // Attributes refers to dubbo 2.7.6. It is different from attachment. It is used in internal process.
- Attributes() map[string]interface{}
- // AttributeByKey gets attribute by key , if nil then return default value
- AttributeByKey(string, interface{}) interface{}
- // SetAttachments sets attribute by @key and @value.
- SetAttachments(key string, value interface{})
+
  // Invoker gets the invoker in current context.
  Invoker() Invoker
  // IsGenericInvocation gets if this is a generic invocation
  IsGenericInvocation() bool
+
+ Attachments() map[string]interface{}
+ SetAttachment(key string, value interface{})
+ GetAttachment(key string) (string, bool)
+ GetAttachmentInterface(string) interface{}
+ GetAttachmentWithDefaultValue(key string, defaultValue string) string
+
+ // Attributes firstly introduced on dubbo-java 2.7.6. It is
+ // used in internal invocation, that is, it's not passed between
+ // server and client.
+ Attributes() map[string]interface{}
+ SetAttribute(key string, value interface{})
+ GetAttribute(key string) (interface{}, bool)
+ GetAttributeWithDefaultValue(key string, defaultValue interface{}) interface{}
 }