﻿/**
 * Tencent is pleased to support the open source community by making Tars available.
 *
 * Copyright (C) 2016THL A29 Limited, a Tencent company. All rights reserved.
 *
 * Licensed under the BSD 3-Clause License (the "License"); you may not use this file except 
 * in compliance with the License. You may obtain a copy of the License at
 *
 * https://opensource.org/licenses/BSD-3-Clause
 *
 * 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.
 */

#ifndef __TARS_ADAPTER_PROXY_H_
#define __TARS_ADAPTER_PROXY_H_

#include "util/tc_timeout_queue_new.h"
#include "util/tc_timeout_queue_map.h"
#include "util/tc_transceiver.h"
#include "servant/Global.h"
#include "servant/EndpointInfo.h"
#include "servant/ObjectProxy.h"
#include "servant/Message.h"
#include "servant/StatReport.h"
#include "servant/AuthLogic.h"
#include "servant/AuthF.h"

#include <queue>
#include <unordered_map>

// #ifdef TARS_OPENTRACKING
// #include <opentracing/span.h>
// #endif
namespace tars
{
/**
 * 每个Adapter对应一个Endpoint，也就是一个服务端口
 */
class AdapterProxy
{
public:
    /**
     * 构造函数
     * @param ep
     * @param op
     */
    AdapterProxy(ObjectProxy * pObjectProxy, const EndpointInfo &ep,Communicator* pCom);

    /**
     * 析构函数
     */
    ~AdapterProxy();

    /**
     * 调用server端对象方法
     * @param req
     * @return int
     */
	int invoke(ReqMessage * msg);

	/**
	 *
	 */
	void onConnect();

    /**
     * 发送请求
     * 发送挤压的数据
     * @param req
     * @return
     */
    void doInvoke();

    /**
     * server端的响应包返回
     */
    void finishInvoke(shared_ptr<ResponsePacket> &rsp);

    /**
     * 发起建立连接请求, 并返回是否成功
     * @connecting : false: 已经建立连接的才返回true(hash模式下使用, 保证hash不会乱), true: 发起连接就返回true
     * @return bool, 没有可用连接则返回false, 否则返回true
     */
    bool checkActive(bool connecting);

    /**
     * 重置try time
     * @param next: true: 设置为下一次重试时间, 这之前不能发起连接, false: 设置为当前时间(马上就可以发起连接)
     */
    void resetRetryTime(bool next = true);

    /**
     * 记录连接是否异常
     */
    void addConnExc(bool bExc);

    /**
     * 处理超时
     */
    void doTimeout();

    /**
     * 发送tars_ping
     */
    void doKeepAlive();

    /**
     * 处理stat
     */
    void mergeStat(map<StatMicMsgHead, StatMicMsgBody> & mStatMicMsg);

// #ifdef TARS_OPENTRACKING
// 	/** 
// 	 * Zipkin调用链
// 	 */
//     void startTrack(ReqMessage * msg);

//     void finishTrack(ReqMessage * msg);
// #endif	
	
    /**
     * 获取ObjectProxy
     */
    inline ObjectProxy * getObjProxy() { return _objectProxy; }

    /**
     * 获取端口信息
     * @return const EndpointInfo&
     */
    inline const EndpointInfo & endpoint() const { return _ep; }

    /**
     * 连接超时的时间
     */
    int getConTimeout();

    /**
     * 连接是否超时
     */
    inline bool isConnTimeout() { return _trans->isConnTimeout(); }

    /**
     * 设置连接是否超时
     */
    void   setConTimeout(bool bConTimeout);

    /**
     * 连接是否异常
     */
    inline bool isConnExc() { return _connExc; }

    /**
     * 连接异常的次数
     */
    inline int ConnExcCnt() const { return _connExcCnt; }

    /**
     * 服务在主控的状态是否为active
     */
    inline bool isActiveInReg() { return _activeStateInReg; }

    /**
     * 服务在主控的状态是否为active
     */
    inline void setActiveInReg(bool bActive) { _activeStateInReg = bActive; }

    /**
     * 获取连接
     *
     * @return TC_Transceiver*
     */
    inline TC_Transceiver* trans() { return _trans.get(); }

    /**
     * 设置节点的静态权重值
     */
    inline void setWeight(int iWeight)
    {
        if(_staticWeight != iWeight)
            _staticWeightChanged = true;
        _staticWeight = iWeight;
    }

    /**
     * 获取节点的静态权重值
     */
    inline int getWeight() { return _staticWeight; }
    /**
     * 将权重变化标识重置为false
     */
    inline void resetWeightChanged() { _staticWeightChanged = false; }

    /**
     * 判断权重静态权重值是否变化, 参数reset为true时将权重变化标识重置为false
     */
     inline bool checkWeightChanged(bool reset = false)
     {
         bool changed = _staticWeightChanged;
         if(reset)
             _staticWeightChanged = false;
         return changed;
     }

    /**
     * 获取id
     */
    inline int getId() const { return _id; }


	/**
	 * 屏蔽结点
	 */
	void onSetInactive();

	/**
	 * get timeout queue
	 * @return
	 */
	TC_TimeoutQueueNew<ReqMessage*> * getTimeoutQueue() { return _timeoutQueue.get(); }

protected:

    //创建完网络句柄后的回调
    shared_ptr<TC_ProxyInfo> onCreateCallback(TC_Transceiver*);

    std::shared_ptr<TC_OpenSSL> onOpensslCallback(TC_Transceiver*);

    void onCloseCallback(TC_Transceiver*, TC_Transceiver::CloseReason reason, const string &err);

    void onConnectCallback(TC_Transceiver*);

    void onRequestCallback(TC_Transceiver*);

    shared_ptr<TC_NetWorkBuffer::Buffer> onSendAuthCallback(TC_Transceiver*);

    TC_NetWorkBuffer::PACKET_TYPE onVerifyAuthCallback(TC_NetWorkBuffer &, TC_Transceiver*);

    TC_NetWorkBuffer::PACKET_TYPE onParserCallback(TC_NetWorkBuffer&, TC_Transceiver*);

    void onCompletePackage(TC_Transceiver*);

	void doInvoke_serial();

private:


    /**
     * 屏蔽结点
     */
    void setInactive();


    /**
     * 请求的响应处理
     */
    void finishInvoke(ReqMessage * msg);

    /**
     * 超时统计处理
     */
    void finishInvoke(bool bTimeout);

    /**
     * 初始化模块间调用信息的头部
     */
    void initStatHead();

    /**
     * 模块间调用信息的统计
     */
    void stat(ReqMessage * msg);

    /**
     * 模块间调用信息的合并
     */
    void merge(const StatMicMsgBody& inBody, StatMicMsgBody& outBody);

    /**
     * 连接串行模式
     * @param msg
     * @return
     */
	int invoke_connection_serial(ReqMessage * msg);

	/**
	 * 连接并行模式
	 * @param msg
	 * @return
	 */
	int invoke_connection_parallel(ReqMessage * msg);

    /**
     * 完成串行连接请求
     * @param msg
     * @return
     */
	void finishInvoke_serial(shared_ptr<ResponsePacket> & rsp);

    /**
     * 完成连接复用请求
     * @param msg
     * @return
     */
	void finishInvoke_parallel(shared_ptr<ResponsePacket> & rsp);

	/**
	 * 并行发送的情况(连接复用)
	 */
	void doInvoke_parallel();

	/**
	 * slave 名称(去掉set等信息)
	 * @param sSlaveName
	 * @return
	 */
	string getSlaveName(const string& sSlaveName);

private:

    /*
     * 通信器
     */
    Communicator*                           _communicator;

    /*
     * ObjectProxy
     */
    ObjectProxy*                           _objectProxy;

    /**
     * in request
     */
    ReqMessage*                             _requestMsg = NULL;

    /**
     * ep
     */ 
    EndpointInfo                            _ep;

    /*
     * 收发包处理
     */
    std::unique_ptr<TC_Transceiver>         _trans;

    /*
     * 超时队列
     */
    std::unique_ptr<TC_TimeoutQueueNew<ReqMessage*>> _timeoutQueue;

    /*
     * 节点在主控的存活状态
     */
    bool                                   _activeStateInReg;

    /*
     * 节点当前的存活状态
     */
    bool                                   _activeStatus;

    /*
     * 单位时间内的总调用次数
     */
    uint32_t                               _totalInvoke;

    /*
     * 单位时间内的调用超时次数
     */
    uint32_t                               _timeoutInvoke;

    /*
     * 下一次进行超时比率检测的时间
     */
    time_t                                 _nextFinishInvokeTime;

    /*
     * 连续超时的次数
     */
    uint32_t                               _frequenceFailInvoke;

    /*
     * 连续失败的检查时间点
     */
    time_t                                 _frequenceFailTime;

    /*
     * 下一次重试的时间
     */
    time_t                                 _nextRetryTime;

    /*
     * 下一次发起tars_ping请求的时间
     */
    time_t                                 _nextKeepAliveTime;

    /*
     * 是否连接异常
     */
    bool                                   _connExc;

    /*
     * 连接异常的次数
     */
    uint32_t                               _connExcCnt;

    /*
     * 静态权重值
     */
    int                                       _staticWeight;
    
    /*
     * 静态权重是否变更过
     */
    bool                                      _staticWeightChanged;

    /*
     * 超时请求的回包回来后，是否打印超时的日志信息
     */
    bool                                   _timeoutLogFlag;

    /*
     * 非发送队列的大小限制，用于发送过载判断
     */
    size_t                                   _noSendQueueLimit;

    /*
     * 模块间调用统计信息的head信息
     */
    StatMicMsgHead                           _statHead;

    /*
     * 模块间调用统计信息的body信息
     */
    unordered_map<string,StatMicMsgBody>    _statBody;

    /*
     * 调用链信息
     */
// #ifdef TARS_OPENTRACKING
//     map<int,std::unique_ptr<opentracing::Span>> _spanMap;
// #endif
    int                                    _id;
    static  atomic<int>                    _idGen;
};
////////////////////////////////////////////////////////////////////
}
#endif
