Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Ftr: enable filter and cluster when client consumer provider directly #1181

Merged
merged 10 commits into from
May 15, 2021
Merged

Ftr: enable filter and cluster when client consumer provider directly #1181

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
5d4a96d
Merge pull request #4 from apache/3.0
kzhan Mar 21, 2021
610f374
Merge pull request #11 from apache/3.0
kzhan Mar 28, 2021
6c05180
Merge pull request #17 from apache/3.0
kzhan Apr 27, 2021
37780c4
URL directly call add filter and cluster
Apr 27, 2021
b94d03d
update
Apr 28, 2021
c456461
update
Apr 29, 2021
1a63f86
Merge branch '3.0' of https://github.com/apache/dubbo-go into upstrea…
May 6, 2021
02aa711
Merge branch 'upstream/3.0' into feature/client_direct_call
May 6, 2021
3ef9cc7
add mockFilter
May 6, 2021
928206d
Merge branch '3.0' into feature/client_direct_call
AlexStocks May 15, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
34 changes: 33 additions & 1 deletion config/reference_config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ package config
import (
"context"
"fmt"
"github.com/apache/dubbo-go/protocol/protocolwrapper"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

move it to the 3rd import block

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

move it to the 3rd import block

ok

"net/url"
"strconv"
"time"
Expand Down Expand Up @@ -134,11 +135,42 @@ func (c *ReferenceConfig) Refer(_ interface{}) {

if len(c.urls) == 1 {
c.invoker = extension.GetProtocol(c.urls[0].Protocol).Refer(c.urls[0])
// c.URL != "" is direct call
if c.URL != "" {
//filter
c.invoker = protocolwrapper.BuildInvokerChain(c.invoker, constant.REFERENCE_FILTER_KEY)

// cluster
invokers := make([]protocol.Invoker, 0, len(c.urls))
invokers = append(invokers, c.invoker)
// TODO(decouple from directory, config should not depend on directory module)
var hitClu string
// not a registry url, must be direct invoke.
hitClu = constant.FAILOVER_CLUSTER_NAME
if len(invokers) > 0 {
u := invokers[0].GetURL()
if nil != &u {
hitClu = u.GetParam(constant.CLUSTER_KEY, constant.ZONEAWARE_CLUSTER_NAME)
}
}

cluster := extension.GetCluster(hitClu)
// If 'zone-aware' policy select, the invoker wrap sequence would be:
// ZoneAwareClusterInvoker(StaticDirectory) ->
// FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
c.invoker = cluster.Join(directory.NewStaticDirectory(invokers))
}
} else {
invokers := make([]protocol.Invoker, 0, len(c.urls))
var regURL *common.URL
for _, u := range c.urls {
invokers = append(invokers, extension.GetProtocol(u.Protocol).Refer(u))
invoker := extension.GetProtocol(u.Protocol).Refer(u)
// c.URL != "" is direct call
if c.URL != "" {
//filter
invoker = protocolwrapper.BuildInvokerChain(invoker, constant.REFERENCE_FILTER_KEY)
}
invokers = append(invokers, invoker)
if u.Protocol == constant.REGISTRY_PROTOCOL {
regURL = u
}
Expand Down
6 changes: 3 additions & 3 deletions protocol/protocolwrapper/protocol_filter_wrapper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ func (pfw *ProtocolFilterWrapper) Export(invoker protocol.Invoker) protocol.Expo
if pfw.protocol == nil {
pfw.protocol = extension.GetProtocol(invoker.GetURL().Protocol)
}
invoker = buildInvokerChain(invoker, constant.SERVICE_FILTER_KEY)
invoker = BuildInvokerChain(invoker, constant.SERVICE_FILTER_KEY)
return pfw.protocol.Export(invoker)
}

Expand All @@ -63,15 +63,15 @@ func (pfw *ProtocolFilterWrapper) Refer(url *common.URL) protocol.Invoker {
if invoker == nil {
return nil
}
return buildInvokerChain(invoker, constant.REFERENCE_FILTER_KEY)
return BuildInvokerChain(invoker, constant.REFERENCE_FILTER_KEY)
}

// Destroy will destroy all invoker and exporter.
func (pfw *ProtocolFilterWrapper) Destroy() {
pfw.protocol.Destroy()
}

func buildInvokerChain(invoker protocol.Invoker, key string) protocol.Invoker {
func BuildInvokerChain(invoker protocol.Invoker, key string) protocol.Invoker {
filterName := invoker.GetURL().GetParam(key, "")
if filterName == "" {
return invoker
Expand Down