%%%-------------------------------------------------------------------
%%% @author dlive
%%% @copyright (C) 2017, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 28. Dec 2017 7:34 PM
%%%-------------------------------------------------------------------
-module(dubbo_consumer_pool).
-author("dlive").

-behaviour(gen_server).

%% API
-export([start_link/0, start_consumer/2]).

%% gen_server callbacks
-export([init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    terminate/2,
    code_change/3]).

-export([select_connection/1,select_connection/2]).

-include("dubbo.hrl").
-define(SERVER, ?MODULE).

-define(INTERFCE_LIST_TABLE,interface_list).
-define(PROVIDER_NODE_LIST_TABLE,provider_node_list).

-record(state, {}).


%%%===================================================================
%%% API
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Starts the server
%%
%% @end
%%--------------------------------------------------------------------
-spec(start_link() ->
    {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link() ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Initializes the server
%%
%% @spec init(Args) -> {ok, State} |
%%                     {ok, State, Timeout} |
%%                     ignore |
%%                     {stop, Reason}
%% @end
%%--------------------------------------------------------------------
-spec(init(Args :: term()) ->
    {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term()} | ignore).
init([]) ->
    init_ets_table(),
    {ok, #state{}}.
init_ets_table()->
    try ets:new(?INTERFCE_LIST_TABLE, [bag,public,named_table,{keypos,2}]) of
        ?INTERFCE_LIST_TABLE ->
            ok
    catch
        _Type:Reason ->
            logger:error("new ets table error ~p",[Reason]),
            error
    end,
    try ets:new(?PROVIDER_NODE_LIST_TABLE, [bag,public,named_table,{keypos,2}]) of
        ?PROVIDER_NODE_LIST_TABLE ->
            ok
    catch
        _Type1:Reason1 ->
            logger:error("new ets table error ~p",[Reason1]),
            error
    end,
    ok.
%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling call messages
%%
%% @end
%%--------------------------------------------------------------------
-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
    State :: #state{}) ->
    {reply, Reply :: term(), NewState :: #state{}} |
    {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
    {noreply, NewState :: #state{}} |
    {noreply, NewState :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
    {stop, Reason :: term(), NewState :: #state{}}).

handle_call({add_consumer,Interface,ProviderNodeList}, _From, State) ->

    OldProviderList = get_interface_provider_node(Interface),
    NewProviderList = add_consumer(ProviderNodeList,[]),
    DeleteProverList = OldProviderList -- NewProviderList,
    clean_invalid_provider(DeleteProverList),
    {reply, ok, State};
handle_call(_Request, _From, State) ->
    {reply, ok, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling cast messages
%%
%% @end
%%--------------------------------------------------------------------
-spec(handle_cast(Request :: term(), State :: #state{}) ->
    {noreply, NewState :: #state{}} |
    {noreply, NewState :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term(), NewState :: #state{}}).
handle_cast(_Request, State) ->
    {noreply, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling all non call/cast messages
%%
%% @spec handle_info(Info, State) -> {noreply, State} |
%%                                   {noreply, State, Timeout} |
%%                                   {stop, Reason, State}
%% @end
%%--------------------------------------------------------------------
-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
    {noreply, NewState :: #state{}} |
    {noreply, NewState :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term(), NewState :: #state{}}).
handle_info(_Info, State) ->
    {noreply, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% This function is called by a gen_se  rver when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_server terminates
%% with Reason. The return value is ignored.
%%
%% @spec terminate(Reason, State) -> void()
%% @end
%%--------------------------------------------------------------------
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
    State :: #state{}) -> term()).
terminate(_Reason, _State) ->
    ok.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Convert process state when code is changed
%%
%% @spec code_change(OldVsn, State, Extra) -> {ok, NewState}
%% @end
%%--------------------------------------------------------------------
-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
    Extra :: term()) ->
    {ok, NewState :: #state{}} | {error, Reason :: term()}).
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

start_consumer(Interface,ProviderNodeInfo)->
    gen_server:call(?SERVER,{add_consumer,Interface,ProviderNodeInfo}).


%%%===================================================================
%%% Internal functions
%%%===================================================================
add_consumer([],RegisterList)->
    RegisterList;
add_consumer([ProviderNodeInfo|ProviderList],RegisterList)->
    case dubbo_node_config_util:parse_provider_info(ProviderNodeInfo) of
        {ok,ProviderConfig} ->
            HostFlag=get_host_flag(ProviderConfig),
            case ets:lookup(?PROVIDER_NODE_LIST_TABLE,HostFlag) of
                []->
                    ConnectionList = start_provider_process(HostFlag,30,ProviderConfig),
                    ok = update_connection_info(ProviderConfig#provider_config.interface,HostFlag,ConnectionList,true),
                    ok;
                List ->
                    List2 = lists:map(fun(#provider_node_list{connection_info = ConnectionItem })->
                        ConnectionItem
                        end,List),
                    ok = update_connection_info(ProviderConfig#provider_config.interface,HostFlag,List2,false),
                    ok
            end,
            add_consumer(ProviderList,[HostFlag]++RegisterList);
        {error,R1} ->
            logger:error("parse provider info error reason ~p",[R1]),
            add_consumer(ProviderList,RegisterList)
    end.

start_provider_process(HostFlag,Weight,ProviderConfig) ->
    ExecutesList=lists:seq(1,ProviderConfig#provider_config.executes),
    ConnectionList= lists:map(fun(Item) ->
        ConnectionFlag= << HostFlag/binary,(integer_to_binary(Item))/binary >>,
        ConnectionFlagTerm= binary_to_atom(ConnectionFlag,utf8),
        AChild = {ConnectionFlagTerm,{dubbo_netty_client, start_link, [ConnectionFlagTerm,HostFlag,ProviderConfig,Item]}, permanent, 2000, worker, [dubbo_netty_client]},
        {ok,Pid} = dubbo_consumer_pool_sup:add_children(AChild),
        logger:info("start provider ~p pid info ~p~n",[HostFlag,Pid]),
        #connection_info{connection_id = ConnectionFlagTerm,pid = Pid,weight = Weight,host_flag = HostFlag}
        end,ExecutesList),
    ConnectionList.
get_host_flag(ProviderConfig)->
    HostFlag= << (list_to_binary(ProviderConfig#provider_config.host))/binary,<<"_">>/binary,(integer_to_binary(ProviderConfig#provider_config.port))/binary>>,
    HostFlag.

update_connection_info(Interface,HostFlag,ConnectionList,IsUpdateProvideNode)->
    lists:map(fun(Item) ->
        I1 = ets:insert(?INTERFCE_LIST_TABLE,#interface_list{interface = Interface,connection_info = Item}),
        logger:debug("save INTERFCE_LIST_TABLE ~p info:~p",[Interface,I1]),
        case IsUpdateProvideNode of
            true->
                I2 = ets:insert(?PROVIDER_NODE_LIST_TABLE,#provider_node_list{host_flag = HostFlag,connection_info = Item }),
                logger:debug("save PROVIDER_NODE_LIST_TABLE ~p info:~p",[HostFlag,I2]);
            false->
                ok
        end,
        ok
              end,ConnectionList),
    ok.

get_interface_provider_node(Interface)->
    case ets:lookup(?INTERFCE_LIST_TABLE,Interface) of
        []->
            [];
        List->
            ListRet = [Item#interface_list.connection_info#connection_info.host_flag || Item <- List],
            lists_util:del_duplicate(ListRet)
    end.

select_connection(Interface)->
    RandNum = rand:uniform(2048),
    select_connection(Interface,RandNum).
select_connection(Interface,RandNum)->
    case ets:lookup(?INTERFCE_LIST_TABLE,Interface) of
        []->
            {error,none};
        List->
            Len = length(List),
            RemNum = (RandNum rem Len)+1,
%%            RandNum2 = if
%%                           RandNum==Len -> RandNum-1;
%%                           true->RandNum
%%                       end,
            InterfaceListItem = lists:nth(RemNum,List),
            {ok,InterfaceListItem#interface_list.connection_info}
    end.

clean_invalid_provider([])->
    ok;
clean_invalid_provider([HostFlag | DeleteProverList])->
    case ets:lookup(?PROVIDER_NODE_LIST_TABLE,HostFlag) of
        []->
            ok;
        ProviderNodeList->
            io:format("ConnectionList ~p~n",[ProviderNodeList]),
            ProviderNodeList1 = lists_util:del_duplicate(ProviderNodeList),
            clean_connection_info(ProviderNodeList1)
    end,
    clean_invalid_provider(DeleteProverList).

clean_connection_info(ProviderNodeList)->
    lists:map(fun(Item)->
        Pid=Item#provider_node_list.connection_info#connection_info.pid,
        ConnectionId=Item#provider_node_list.connection_info#connection_info.connection_id,
        Pattern=#interface_list{pid=Pid,_='_'},
        ets:delete_object(?INTERFCE_LIST_TABLE,Pattern),
        dubbo_consumer_pool_sup:stop_children(ConnectionId)
              end,ProviderNodeList),
    ok.