/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.dubbo.rpc.cluster.loadbalance;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.LoadbalanceRules;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.AdaptiveMetrics;
import org.apache.dubbo.rpc.Constants;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.support.RpcUtils;

import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_TIMEOUT;
import static org.apache.dubbo.common.constants.CommonConstants.LOADBALANCE_KEY;

/**
 * AdaptiveLoadBalance
 * </p>
 */
public class AdaptiveLoadBalance extends AbstractLoadBalance {

    public static final String NAME = "adaptive";

    //default key
    private String attachmentKey = "mem,load";

    private final AdaptiveMetrics adaptiveMetrics;

    public AdaptiveLoadBalance(ApplicationModel scopeModel){
        adaptiveMetrics = scopeModel.getBeanFactory().getBean(AdaptiveMetrics.class);
    }

    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        Invoker<T> invoker = selectByP2C(invokers,invocation);
        invocation.setAttachment(Constants.ADAPTIVE_LOADBALANCE_ATTACHMENT_KEY,attachmentKey);
        long startTime = System.currentTimeMillis();
        invocation.getAttributes().put(Constants.ADAPTIVE_LOADBALANCE_START_TIME,startTime);
        invocation.getAttributes().put(LOADBALANCE_KEY,LoadbalanceRules.ADAPTIVE);
        adaptiveMetrics.addConsumerReq(getServiceKey(invoker,invocation));
        adaptiveMetrics.setPickTime(getServiceKey(invoker,invocation),startTime);

        return invoker;
    }

    private <T> Invoker<T> selectByP2C(List<Invoker<T>> invokers, Invocation invocation){
        int length = invokers.size();
        if(length == 1) {
            return invokers.get(0);
        }

        if(length == 2) {
            return chooseLowLoadInvoker(invokers.get(0),invokers.get(1),invocation);
        }

        int pos1 = ThreadLocalRandom.current().nextInt(length);
        int pos2 = ThreadLocalRandom.current().nextInt(length - 1);
        if (pos2 >= pos1) {
            pos2 = pos2 + 1;
        }

        return chooseLowLoadInvoker(invokers.get(pos1),invokers.get(pos2),invocation);
    }

    private String getServiceKey(Invoker<?> invoker,Invocation invocation){

        String key = (String) invocation.getAttributes().get(invoker);
        if (StringUtils.isNotEmpty(key)){
            return key;
        }

        key = buildServiceKey(invoker,invocation);
        invocation.getAttributes().put(invoker,key);
        return key;
    }

    private String buildServiceKey(Invoker<?> invoker,Invocation invocation){
        URL url = invoker.getUrl();
        StringBuilder sb = new StringBuilder(128);
        sb.append(url.getAddress()).append(":").append(invocation.getProtocolServiceKey());
        return sb.toString();
    }

    private int getTimeout(Invoker<?> invoker, Invocation invocation) {
        URL url = invoker.getUrl();
        String methodName = RpcUtils.getMethodName(invocation);
        return (int) RpcUtils.getTimeout(url,methodName, RpcContext.getClientAttachment(),invocation, DEFAULT_TIMEOUT);
    }

    private <T> Invoker<T> chooseLowLoadInvoker(Invoker<T> invoker1,Invoker<T> invoker2,Invocation invocation){
        int weight1 = getWeight(invoker1, invocation);
        int weight2 = getWeight(invoker2, invocation);
        int timeout1 = getTimeout(invoker1, invocation);
        int timeout2 = getTimeout(invoker2, invocation);
        long load1 = Double.doubleToLongBits(adaptiveMetrics.getLoad(getServiceKey(invoker1,invocation),weight1,timeout1 ));
        long load2 = Double.doubleToLongBits(adaptiveMetrics.getLoad(getServiceKey(invoker2,invocation),weight2,timeout2 ));

        if (load1 == load2) {
            // The sum of weights
            int totalWeight = weight1 + weight2;
            if (totalWeight > 0) {
                int offset = ThreadLocalRandom.current().nextInt(totalWeight);
                if (offset < weight1) {
                    return invoker1;
                }
                return invoker2;
            }
            return ThreadLocalRandom.current().nextInt(2) == 0 ? invoker1 : invoker2;
        }
        return load1 > load2 ? invoker2 : invoker1;
    }

}
