聊聊dubbo-go的failfastCluster

type failfastCluster struct{}

const failfast = "failfast"

func init() {
    extension.SetCluster(failfast, NewFailFastCluster)
}

// NewFailFastCluster ...
func NewFailFastCluster() cluster.Cluster {
    return &failfastCluster{}
}

func (cluster *failfastCluster) Join(directory cluster.Directory) protocol.Invoker {
    return newFailFastClusterInvoker(directory)
}

type failfastClusterInvoker struct {
    baseClusterInvoker
}

func newFailFastClusterInvoker(directory cluster.Directory) protocol.Invoker {
    return &failfastClusterInvoker{
        baseClusterInvoker: newBaseClusterInvoker(directory),
    }
}

func (invoker *failfastClusterInvoker) Invoke(ctx context.Context, invocation protocol.Invocation) protocol.Result {
    invokers := invoker.directory.List(invocation)
    err := invoker.checkInvokers(invokers, invocation)
    if err != nil {
        return &protocol.RPCResult{Err: err}
    }

    loadbalance := getLoadBalance(invokers[0], invocation)

    err = invoker.checkWhetherDestroyed()
    if err != nil {
        return &protocol.RPCResult{Err: err}
    }

    ivk := invoker.doSelect(loadbalance, invocation, invokers, nil)
    return ivk.Invoke(ctx, invocation)
}

func getLoadBalance(invoker protocol.Invoker, invocation protocol.Invocation) cluster.LoadBalance {
    url := invoker.GetUrl()

    methodName := invocation.MethodName()
    //Get the service loadbalance config
    lb := url.GetParam(constant.LOADBALANCE_KEY, constant.DEFAULT_LOADBALANCE)

    //Get the service method loadbalance config if have
    if v := url.GetMethodParam(methodName, constant.LOADBALANCE_KEY, ""); len(v) > 0 {
        lb = v
    }
    return extension.GetLoadbalance(lb)
}

func (invoker *baseClusterInvoker) doSelect(lb cluster.LoadBalance, invocation protocol.Invocation, invokers []protocol.Invoker, invoked []protocol.Invoker) protocol.Invoker {

    var selectedInvoker protocol.Invoker
    url := invokers[0].GetUrl()
    sticky := url.GetParamBool(constant.STICKY_KEY, false)
    //Get the service method sticky config if have
    sticky = url.GetMethodParamBool(invocation.MethodName(), constant.STICKY_KEY, sticky)

    if invoker.stickyInvoker != nil && !isInvoked(invoker.stickyInvoker, invokers) {
        invoker.stickyInvoker = nil
    }

    if sticky && invoker.stickyInvoker != nil && (invoked == nil || !isInvoked(invoker.stickyInvoker, invoked)) {
        if invoker.availablecheck && invoker.stickyInvoker.IsAvailable() {
            return invoker.stickyInvoker
        }
    }

    selectedInvoker = invoker.doSelectInvoker(lb, invocation, invokers, invoked)

    if sticky {
        invoker.stickyInvoker = selectedInvoker
    }
    return selectedInvoker

}

func (invoker *baseClusterInvoker) doSelectInvoker(lb cluster.LoadBalance, invocation protocol.Invocation, invokers []protocol.Invoker, invoked []protocol.Invoker) protocol.Invoker {
    if len(invokers) == 1 {
        return invokers[0]
    }

    selectedInvoker := lb.Select(invokers, invocation)

    //judge to if the selectedInvoker is invoked

    if (!selectedInvoker.IsAvailable() && invoker.availablecheck) || isInvoked(selectedInvoker, invoked) {
        // do reselect
        var reslectInvokers []protocol.Invoker

        for _, invoker := range invokers {
            if !invoker.IsAvailable() {
                continue
            }

            if !isInvoked(invoker, invoked) {
                reslectInvokers = append(reslectInvokers, invoker)
            }
        }

        if len(reslectInvokers) > 0 {
            selectedInvoker = lb.Select(reslectInvokers, invocation)
        } else {
            return nil
        }
    }
    return selectedInvoker
}