merge form master
diff --git a/config_example/sys.config b/config_example/sys.config
index b100088..07cfaf9 100644
--- a/config_example/sys.config
+++ b/config_example/sys.config
@@ -17,13 +17,14 @@
]
},
{dubboerl,[
+ {registry,zookeeper},
{zookeeper_list,[{"127.0.0.1",2181}]},
{application,<<"testdubboerl">>},
{registry,true},
{protocol,hessian},
{port,20881},
{consumer,[
- {<<"org.apache.dubbo.erlang.sample.service.facade.UserOperator">>,[]}
+ % {<<"org.apache.dubbo.erlang.sample.service.facade.UserOperator">>,[]}
]},
{provider,[
{dubbo_service_user_impl,userOperator,<<"org.apache.dubbo.erlang.sample.service.facade.UserOperator">>,[]}
diff --git a/include/dubbo.hrl b/include/dubbo.hrl
index ad2277a..282e0da 100644
--- a/include/dubbo.hrl
+++ b/include/dubbo.hrl
@@ -100,10 +100,11 @@
}).
+-record(interface_info, {interface, loadbalance}).
-record(interface_list, {interface, pid, connection_info}).
--record(provider_node_list, {host_flag, connection_info}).
--record(connection_info, {connection_id, pid, weight, host_flag, readonly = false}).
+%%-record(provider_node_list, {host_flag, pid, weight, readonly = false}).
+-record(connection_info, {host_flag, pid, weight, readonly = false}).
-type dubbo_request() :: #dubbo_request{}.
-type dubbo_response() :: #dubbo_response{}.
\ No newline at end of file
diff --git a/include/dubboerl.hrl b/include/dubboerl.hrl
index 31204a5..a9be5f1 100644
--- a/include/dubboerl.hrl
+++ b/include/dubboerl.hrl
@@ -14,8 +14,13 @@
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%------------------------------------------------------------------------------
--define(PROVIDER_IMPL_TABLE,provider_impl_table).
+-define(PROVIDER_IMPL_TABLE, provider_impl_table).
--define(PROVIDER_WORKER,provider_worker).
+-define(PROVIDER_WORKER, provider_worker).
--define(TRAFFIC_CONTROL,traffic_control).
\ No newline at end of file
+-define(TRAFFIC_CONTROL, traffic_control).
+
+
+-record(dubbo_url, {scheme, user_info, host, port, path, parameters, fragment}).
+
+-record(dubbo_invoker, {host_flag, handle}).
\ No newline at end of file
diff --git a/samples/dubbo-sample-service/src/main/resources/applicationProvider.xml b/samples/dubbo-sample-service/src/main/resources/applicationProvider.xml
index 974fabc..425e41e 100644
--- a/samples/dubbo-sample-service/src/main/resources/applicationProvider.xml
+++ b/samples/dubbo-sample-service/src/main/resources/applicationProvider.xml
@@ -10,8 +10,8 @@
<dubbo:consumer check="false" timeout="300000" id="dubboConsumerConfig" retries="0"/>
- <bean id="userService" class="org.apache.dubbo.erlang.sample.service.impl.UserOperatorImpl"/>
- <dubbo:service interface="org.apache.dubbo.erlang.sample.service.facade.UserOperator" ref="userService"/>
+<!-- <bean id="userService" class="org.apache.dubbo.erlang.sample.service.impl.UserOperatorImpl"/>-->
+<!-- <dubbo:service interface="org.apache.dubbo.erlang.sample.service.facade.UserOperator" ref="userService"/>-->
- <!-- <dubbo:reference id="userInterface" interface="UserOperator" retries="0" />-->
+ <dubbo:reference id="userInterface" interface="org.apache.dubbo.erlang.sample.service.facade.UserOperator" retries="0" />
</beans>
diff --git a/samples/dubboerl_demo/apps/dubbo_sample_service/src/userOperator.erl b/samples/dubboerl_demo/apps/dubbo_sample_service/src/userOperator.erl
index b92ca62..d7e0809 100644
--- a/samples/dubboerl_demo/apps/dubbo_sample_service/src/userOperator.erl
+++ b/samples/dubboerl_demo/apps/dubbo_sample_service/src/userOperator.erl
@@ -67,7 +67,7 @@
]
},
Request = dubbo_adapter:reference(Data),
- dubbo_invoker:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
+ dubbo_invoker_old:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
-spec genUserId()->
@@ -96,7 +96,7 @@
]
},
Request = dubbo_adapter:reference(Data),
- dubbo_invoker:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
+ dubbo_invoker_old:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
-spec queryUserInfo(Arg0::#userInfoRequest{})->
@@ -127,7 +127,7 @@
]
},
Request = dubbo_adapter:reference(Data),
- dubbo_invoker:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
+ dubbo_invoker_old:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
-spec queryUserList(Arg0::list())->
@@ -158,5 +158,5 @@
]
},
Request = dubbo_adapter:reference(Data),
- dubbo_invoker:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
+ dubbo_invoker_old:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
diff --git a/src/dubbo_client_default.erl b/src/dubbo_client_default.erl
new file mode 100644
index 0000000..220c30f
--- /dev/null
+++ b/src/dubbo_client_default.erl
@@ -0,0 +1,431 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_client_default).
+
+-behaviour(gen_server).
+
+
+-include("dubbo.hrl").
+
+%% gen_server callbacks
+-export([init/1,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2,
+ terminate/2,
+ code_change/3]).
+-export([start_link/1]).
+
+-export([check_recv_data/2]).
+
+-define(SERVER, ?MODULE).
+
+-record(heartbeat, {last_write = 0, last_read = 0, timeout = 60000, max_timeout = 180000}).
+-record(state, {provider_config, socket = undefined,
+ heartbeat = #heartbeat{},
+ recv_buffer = <<>>, %%从服务端接收的数据
+ host_flag,
+ reconnection_timer,
+ handler
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+%%--------------------------------------------------------------------
+%% @doc
+%% Starts the server
+%%
+%% @end
+%%--------------------------------------------------------------------
+-spec(start_link(Name :: binary(), ProviderConfig :: #provider_config{}) ->
+ {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
+start_link(ProviderConfig) ->
+ gen_server:start_link(?MODULE, [ProviderConfig], []).
+
+%%%===================================================================
+%%% 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([HostFlag, ProviderConfig]) ->
+ #provider_config{host = Host, port = Port} = ProviderConfig,
+ State = case open(Host, Port) of
+ {ok, Socket} ->
+ #state{socket = Socket};
+ {error, _Reason} ->
+ #state{}
+ end,
+ NowStamp = dubbo_time_util:timestamp_ms(),
+ HeartBeatInfo = #heartbeat{last_read = NowStamp, last_write = NowStamp},
+ logger:info("netty client start ~p", [HostFlag]),
+ start_heartbeat_timer(HeartBeatInfo),
+ {ok, State#state{provider_config = ProviderConfig, heartbeat = HeartBeatInfo, host_flag = HostFlag}}.
+
+%%--------------------------------------------------------------------
+%% @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(_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({send_request, Ref, Request, Data, SourcePid, RequestState}, State) ->
+ logger:debug("[send_request begin] send data to provider consumer mid ~p pid ~p sourcePid ~p", [Request#dubbo_request.mid, self(), SourcePid]),
+ NewState = case send_msg(Data, State) of
+ ok ->
+ save_request_info(Request, SourcePid, Ref, RequestState),
+ logger:debug("[send_request end] send data to provider consumer pid ~p state ok", [self()]),
+ State;
+ {error, closed} ->
+ logger:warning("send request error, connection is closed"),
+ State2 = reconnect(State),
+ State2;
+ {error, R1} ->
+ logger:error("[send_request end] send data to provider consumer pid error ~p ~p", [self(), R1]),
+ State
+ end,
+ HeartbeatInfo = update_heartbeat(write, NewState#state.heartbeat),
+ {noreply, NewState#state{heartbeat = HeartbeatInfo}};
+
+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({tcp, _Port, Data}, #state{recv_buffer = RecvBuffer} = State) ->
+%% inet:setopts(State#state.socket, [{active, once}]),
+%% logger:debug("[INFO] recv one data ~w",[Data]),
+ {ok, NextBuffer, NewState} = case check_recv_data(<<RecvBuffer/binary, Data/binary>>, State) of
+ {next_buffer, NextBuffer2, State3} ->
+ logger:debug("[INFO] recv one data state wait next_buffer"),
+ {ok, NextBuffer2, State3}
+ end,
+%% HeartbeatInfo =update_heartbeat(write,NewState#state.heartbeat),
+ {noreply, NewState#state{recv_buffer = NextBuffer}};
+handle_info({tcp_closed, Port}, State) ->
+ logger:info("dubbo connection closed ~p", [Port]),
+ NewState = reconnect(State),
+ {noreply, NewState};
+handle_info({timeout, _TimerRef, {reconnect}}, State) ->
+ NewState = reconnect(State#state{reconnection_timer = undefined}),
+ {noreply, NewState};
+handle_info({timeout, _TimerRef, {heartbeat_timer}}, State) ->
+ {ok, NewState} = case check_heartbeat_state(State) of
+ {normal} -> {ok, State};
+ {send_heart} ->
+ send_heartbeat_msg(undefined, true, State);
+ {reconnect} ->
+ %% @todo reconnect
+ {ok, State}
+ end,
+ HeartbeatInfo = update_heartbeat(write, NewState#state.heartbeat),
+ start_heartbeat_timer(HeartbeatInfo),
+ {noreply, NewState#state{heartbeat = HeartbeatInfo}};
+handle_info(_Info, State) ->
+ logger:warning("[INFO] get one info:~p", [_Info]),
+%% inet:setopts(State#state.socket, [{active, once}]),
+%% case State#state.tmp_pid of
+%% undefined ->ok;
+%% Pid ->
+%% gen_server:cast(Pid,{msg_back})
+%% end,
+ HeartbeatInfo = update_heartbeat(write, State#state.heartbeat),
+ {noreply, State#state{heartbeat = HeartbeatInfo}}.
+
+%%--------------------------------------------------------------------
+%% @private
+%% @doc
+%% This function is called by a gen_server 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) ->
+ logger:warning("terminate reason:~p", [_Reason]),
+ 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}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+open(Host, Port) ->
+ logger:debug("will connect to provider ~p ~p", [Host, Port]),
+ %
+ case gen_tcp:connect(Host, Port, [
+ binary,
+ {packet, 0}, {active, false},
+ {reuseaddr, true},
+ {delay_send, true},
+ {nodelay, true},
+ {high_watermark, 512 * 1024},
+ {low_watermark, 256 * 1024},
+ {sndbuf, 512 * 1024},
+ {recbuf, 512 * 1024}
+ ]) of
+ {ok, Sockets} ->
+ inet:setopts(Sockets, [{active, true}]),
+ {ok, Sockets};
+ Info ->
+ logger:error("start client connection error ~p", [Info]),
+ {error, Info}
+ end.
+
+reconnect(#state{reconnection_timer = Timer} = State) when Timer /= undefined ->
+ State;
+reconnect(State) ->
+ #provider_config{host = Host, port = Port} = State#state.provider_config,
+ case State#state.socket of
+ undefined -> ok;
+ Socket ->
+ gen_tcp:close(Socket)
+ end,
+ case open(Host, Port) of
+ {ok, Socket2} ->
+ logger:warning("reconnect to provider ~p ~p success", [Host, Port]),
+ State#state{socket = Socket2, recv_buffer = <<>>};
+ {error, Reason} ->
+ logger:warning("connect to provider error ~p", [Reason]),
+ TimerRef = erlang:start_timer(2000, self(), {reconnect}),
+ State#state{socket = undefined, reconnection_timer = TimerRef}
+ end.
+
+send_msg(Msg, State) ->
+ case State#state.socket of
+ undefined ->
+ {error, closed};
+ Socket ->
+ case gen_tcp:send(Socket, Msg) of
+ ok ->
+ ok;
+ {error, Reason} ->
+ logger:error("send to server error,reason:~p", [Reason]),
+ {error, Reason}
+ end
+ end.
+
+%%%=================================================================
+%%% 心跳检测
+%%%=================================================================
+start_heartbeat_timer(HeartbeatInfo) ->
+ erlang:start_timer(HeartbeatInfo#heartbeat.timeout, self(), {heartbeat_timer}),
+ ok.
+update_heartbeat(write, Info) ->
+ Info#heartbeat{last_write = dubbo_time_util:timestamp_ms()};
+update_heartbeat(read, Info) ->
+ Info#heartbeat{last_read = dubbo_time_util:timestamp_ms()}.
+
+
+check_heartbeat_state(#state{heartbeat = HeartBeatInfo} = _State) ->
+ Now = dubbo_time_util:timestamp_ms(),
+ #heartbeat{last_read = LastRead, last_write = LastWrite, timeout = Timeout, max_timeout = MaxTimeout} = HeartBeatInfo,
+ if
+ (Now - LastRead) > Timeout ->
+ {send_heart};
+ (Now - LastWrite) > Timeout ->
+ {send_heart};
+ (Now - LastRead) > MaxTimeout ->
+ {reconnect};
+ true ->
+ {normal}
+ end.
+
+
+send_heartbeat_msg(Mid, NeedResponse, State) ->
+ {ok, Bin} = dubbo_heartbeat:generate_request(Mid, NeedResponse),
+ NewState = case send_msg(Bin, State) of
+ ok ->
+ logger:info("send one heartbeat to server"),
+ State;
+ {error, Reason} ->
+ logger:warning("dubbo connection send heartbeat error ~p", [Reason]),
+ State2 = reconnect(State),
+ State2
+ end,
+ {ok, NewState}.
+
+%%%=================================================================
+%%% 接收数据处理
+%%%=================================================================
+-spec check_recv_data(Data :: binary(), State :: #state{}) -> {ready, ReadyData :: binary()} | {ready, ReadyData :: binary(), NextBuffer :: binary()}.
+check_recv_data(<<?DUBBO_MEGIC_HIGH, ?DUBBO_MEGIC_LOW, Rest/binary>> = Data, State) when byte_size(Rest) < 14 ->
+ {next_buffer, Data, State};
+check_recv_data(<<?DUBBO_MEGIC_HIGH, ?DUBBO_MEGIC_LOW, _OtherFlag:80, DataLen:32, Rest/binary>> = Data, State) ->
+ RestSize = byte_size(Rest),
+ if
+ DataLen == RestSize ->
+ {ok, State2} = process_data(Data, State),
+ {next_buffer, <<>>, State2};
+ DataLen > RestSize ->
+ logger:warning("need wait next buffer data ~p", [Data]),
+ {next_buffer, Data, State};
+ DataLen < RestSize ->
+ <<ReadyData:DataLen/binary, NextBuffer/binary>> = Rest,
+ OneData = <<?DUBBO_MEGIC_HIGH:8, ?DUBBO_MEGIC_LOW:8, _OtherFlag:80, DataLen:32, ReadyData/binary>>,
+ {ok, State3} = process_data(OneData, State),
+%% logger:warning("recevi more data ~w ",[NextBuffer]),
+ check_recv_data(NextBuffer, State3)
+ end;
+check_recv_data(<<Error/integer, Data/binary>>, State) ->
+ logger:error("recv bad header data,Begin Byte:~p", [Error]),
+ check_recv_data(Data, State);
+check_recv_data(<<>>, State) ->
+ {next_buffer, <<>>, State}.
+
+
+process_data(Data, State) ->
+ <<Header:16/binary, RestData/binary>> = Data,
+ case dubbo_codec:decode_header(Header) of
+ {ok, response, ResponseInfo} ->
+ process_response(ResponseInfo#dubbo_response.is_event, ResponseInfo, RestData, State),
+%% dubbo_traffic_control:decr_count(State#state.host_flag),
+%% case get_earse_request_info(ResponseInfo#dubbo_response.mid) of
+%% undefined->
+%% logger:error("dubbo response can't find request data,response ~p",[ResponseInfo]);
+%% {SourcePid,Ref,_RequestState} ->
+%% {ok,Res} = dubbo_codec:decode_response(ResponseInfo,RestData),
+%%
+%% logger:info("got one response mid ~p, is_event ~p state ~p",[Res#dubbo_response.mid,Res#dubbo_response.is_event,Res#dubbo_response.state]),
+%% case Res#dubbo_response.is_event of
+%% false ->
+%% %% todo rpccontent need merge response with request
+%% RpcContent=[],
+%% ResponseData = dubbo_type_transfer:response_to_native(Res),
+%% gen_server:cast(SourcePid,{response_process,Ref,RpcContent,ResponseData});
+%% _->
+%% ok
+%% end
+%% end,
+ {ok, State};
+ {ok, request, RequestInfo} ->
+ {ok, Req} = dubbo_codec:decode_request(RequestInfo, RestData),
+ logger:info("get one request mid ~p, is_event ~p", [Req#dubbo_request.mid, Req#dubbo_request.is_event]),
+ {ok, State2} = process_request(Req#dubbo_request.is_event, Req, State),
+ {ok, State2};
+ {error, Type, RelData} ->
+ logger:error("process_data error type ~p RelData ~p", [Type, RelData]),
+ {ok, State}
+ end.
+
+
+%% @doc process event
+-spec process_response(IsEvent :: boolean(), #dubbo_response{}, #state{}, term()) -> ok.
+process_response(false, ResponseInfo, RestData, State) ->
+ dubbo_traffic_control:decr_count(State#state.host_flag),
+ case get_earse_request_info(ResponseInfo#dubbo_response.mid) of
+ undefined ->
+ logger:error("dubbo response can't find request data,response ~p", [ResponseInfo]);
+ {SourcePid, Ref, _RequestState} ->
+ {ok, Res} = dubbo_codec:decode_response(ResponseInfo, RestData),
+ logger:info("got one response mid ~p, is_event ~p state ~p", [Res#dubbo_response.mid, Res#dubbo_response.is_event, Res#dubbo_response.state]),
+ case Res#dubbo_response.is_event of
+ false ->
+ %% todo rpccontent need merge response with request
+ RpcContent = [],
+ ResponseData = dubbo_type_transfer:response_to_native(Res),
+ gen_server:cast(SourcePid, {response_process, Ref, RpcContent, ResponseData});
+ _ ->
+ ok
+ end
+ end,
+ {ok, State};
+process_response(true, _ResponseInfo, _RestData, State) ->
+ {ok, State}.
+
+process_request(true, #dubbo_request{data = <<"R">>}, State) ->
+ {ok, _} = dubbo_provider_consumer_reg_table:update_connection_readonly(self(), true),
+ {ok, State};
+process_request(true, Request, State) ->
+ {ok, NewState} = send_heartbeat_msg(Request#dubbo_request.mid, false, State),
+ {ok, NewState};
+process_request(false, Request, State) ->
+ {ok, State}.
+
+
+save_request_info(Request, SourcePid, Ref, RequestState) ->
+ put(Request#dubbo_request.mid, {SourcePid, Ref, RequestState}).
+get_earse_request_info(Mid) ->
+ erase(Mid).
\ No newline at end of file
diff --git a/src/dubbo_cluster.erl b/src/dubbo_cluster.erl
new file mode 100644
index 0000000..da4031e
--- /dev/null
+++ b/src/dubbo_cluster.erl
@@ -0,0 +1,21 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_cluster).
+-author("dlive").
+
+%% API
+-export([]).
diff --git a/src/dubbo_cluster_failfast.erl b/src/dubbo_cluster_failfast.erl
new file mode 100644
index 0000000..6078517
--- /dev/null
+++ b/src/dubbo_cluster_failfast.erl
@@ -0,0 +1,21 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_cluster_failfast).
+-author("dlive").
+
+%% API
+-export([]).
diff --git a/src/dubbo_common_fun.erl b/src/dubbo_common_fun.erl
index 6744171..5f38fbd 100644
--- a/src/dubbo_common_fun.erl
+++ b/src/dubbo_common_fun.erl
@@ -16,8 +16,9 @@
%%------------------------------------------------------------------------------
-module(dubbo_common_fun).
+-include("dubboerl.hrl").
%% API
--export([local_ip_v4/0, local_ip_v4_str/0]).
+-export([local_ip_v4/0, local_ip_v4_str/0, parse_url/1, url_to_binary/1]).
local_ip_v4() ->
{ok, Addrs} = inet:getifaddrs(),
@@ -29,3 +30,61 @@
local_ip_v4_str() ->
{V1, V2, V3, V4} = local_ip_v4(),
list_to_binary(io_lib:format("~p.~p.~p.~p", [V1, V2, V3, V4])).
+
+
+-spec(parse_url(Url :: binary()|list()) -> {ok, #dubbo_url{}}).
+parse_url(Url) when is_binary(Url) ->
+ parse_url(binary_to_list(Url));
+parse_url(Url) ->
+ case http_uri:parse(Url, []) of
+ {ok, {Scheme, _UserInfo, Host, Port, _Path, Query}} ->
+ QueryStr = case lists:prefix("?", Query) of
+ true ->
+ [_ | Query2] = Query,
+ Query2;
+ false ->
+ Query
+ end,
+ QueryListTmp = string:tokens(QueryStr, "&"),
+ Parameters = parse_url_parameter(QueryListTmp, #{}),
+ Result = #dubbo_url{scheme = Scheme, host = Host, port = Port, parameters = Parameters},
+ {ok, Result};
+ {error, R1} ->
+ {error, R1}
+ end.
+
+
+parse_url_parameter([], Parameters) ->
+ Parameters;
+parse_url_parameter([Item | Rest], Parameters) ->
+ case string:tokens(Item, "=") of
+ KeyPair when length(KeyPair) == 2 ->
+ [Key, Value] = KeyPair,
+ parse_url_parameter(Rest, maps:put(Key, Value, Parameters));
+ KeyPair2 ->
+ logger:error("parse parameter error, keypair ~p", [KeyPair2]),
+ parse_url_parameter(Rest, Parameters)
+ end.
+
+
+url_to_binary(UrlInfo) ->
+ ParameterStr =
+ case UrlInfo#dubbo_url.parameters of
+ undefined ->
+ "";
+ Parameter ->
+ KeyValues = maps:to_list(Parameter),
+ KeyValues2 = [io_lib:format("~s=~s", [Key, http_uri:encode(Value)]) || {Key, Value} <= KeyValues],
+ ParameterStr1 = string:join(KeyValues2, "&"),
+ ParameterStr2 = ["?" | ParameterStr1],
+ list_to_binary(ParameterStr2)
+ end,
+ Value = io_lib:format(<<"~s://~s:~p/~s?~s">>,
+ [
+ UrlInfo#dubbo_url.scheme,
+ UrlInfo#dubbo_url.host,
+ UrlInfo#dubbo_url.port,
+ UrlInfo#dubbo_url.path,
+ ParameterStr
+ ]),
+ list_to_binary(Value).
\ No newline at end of file
diff --git a/src/dubbo_directory.erl b/src/dubbo_directory.erl
new file mode 100644
index 0000000..c667f29
--- /dev/null
+++ b/src/dubbo_directory.erl
@@ -0,0 +1,217 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_directory).
+
+-behaviour(gen_server).
+-include("dubboerl.hrl").
+-include("dubbo.hrl").
+
+-export([subscribe/2,notify/2]).
+%% API
+-export([start_link/0]).
+
+%% gen_server callbacks
+-export([init/1,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2,
+ terminate/2,
+ code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+-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([]) ->
+ {ok, #state{}}.
+
+subscribe(RegistryName,SubcribeUrl)->
+ try gen_server:call(?SERVER,{subscribe,RegistryName,SubcribeUrl},5000) of
+ ok->
+ ok
+ catch
+ Error:Reason->
+ %% todo improve erro type
+ {error,Reason}
+ end.
+
+notify(Interface,UrlList)->
+ %% @todo if UrlList size is 1, and protocol is empty ,need destroyAllInvokers
+
+ refresh_invoker(UrlList),
+%% dubbo_consumer_pool:start_consumer(Interface, UrlList),
+ ok.
+
+
+refresh_invoker(UrlList)->
+ case pick_interface(UrlList) of
+ {error,Reason}->
+ fail;
+ {"empty",Interface,_}->
+ todo_destroy;
+ {_,Interface,LoadBalance} ->
+ OldProviderHosts = dubbo_provider_consumer_reg_table:get_interface_provider_node(Interface),
+ NewInvokers = refresh_invoker(UrlList,[]),
+ NewProviderHosts = [Item#dubbo_invoker.host_flag || Item <- NewInvokers],
+ DeleteProverList = OldProviderHosts -- NewProviderHosts,
+ dubbo_provider_consumer_reg_table:clean_invalid_provider(DeleteProverList),
+ dubbo_provider_consumer_reg_table:update_connection_info(#interface_info{interface = Interface,loadbalance = LoadBalance})
+ end.
+%% OldProviderHosts =
+
+
+refresh_invoker([Url|Rest],Acc)->
+ case dubbo_extension:run_fold(protocol,refer,[Url],undefined) of
+ undefined ->
+ refresh_invoker(Rest,Acc);
+ {ok,Invoker} ->
+ refresh_invoker(Rest,[Invoker|Acc]);
+ {stop,_}->
+ refresh_invoker(Rest,Acc)
+ end.
+
+pick_interface([Url | _]) ->
+ case dubbo_common_fun:parse_url(Url) of
+ {ok,UrlInfo}->
+ Interface = maps:get("interface",UrlInfo#dubbo_url.parameters),
+ LoadBalance = list_to_atom("dubbo_loadbalance_" ++ maps:get("loadbalance",UrlInfo#dubbo_url.parameters,"random")),
+ {UrlInfo#dubbo_url.scheme,Interface,LoadBalance};
+ {error,Reason} ->
+ {error,Reason}
+ end.
+
+
+%%--------------------------------------------------------------------
+%% @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({subscribe,RegistryName,SubcribeUrl}, _From, State) ->
+ NotifyFun= fun dubbo_directory:notify/1,
+ apply(RegistryName,subscribe,[SubcribeUrl,NotifyFun]),
+ {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_server 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}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
diff --git a/src/dubbo_exchanger.erl b/src/dubbo_exchanger.erl
new file mode 100644
index 0000000..6fb8496
--- /dev/null
+++ b/src/dubbo_exchanger.erl
@@ -0,0 +1,34 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_exchanger).
+
+-include("dubbo.hrl").
+
+%% API
+-export([connect/2]).
+
+connect(ProviderConfig, Handler) ->
+ HostFlag = dubbo_provider_consumer_reg_table:get_host_flag(ProviderConfig),
+ {ok, Pid} = dubbo_transport_pool_sup:add_children(ProviderConfig, Handler),
+ logger:info("start provider ~p pid info ~p~n", [HostFlag, Pid]),
+ {ok, #connection_info{pid = Pid, weight = get_weight(ProviderConfig), host_flag = HostFlag}}.
+
+
+
+get_weight(_ProviderConfig) ->
+ %% todo get weight from provider info
+ 30.
\ No newline at end of file
diff --git a/src/dubbo_extension.erl b/src/dubbo_extension.erl
new file mode 100644
index 0000000..4e5eb0c
--- /dev/null
+++ b/src/dubbo_extension.erl
@@ -0,0 +1,189 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_extension).
+-behaviour(gen_server).
+
+%% API
+-export([run/3,run_fold/4,register/3,unregister/3]).
+
+
+-export([start_link/0]).
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+
+-define(TAB, ?MODULE).
+
+-record(state, {}).
+
+
+-spec reg(HookName::hookname(), Module::atom(),Priority::integer()) -> ok | {error, term()}.
+register(HookName, Module,Priority) ->
+ gen_server:call(?MODULE, {register, HookName, {Priority, {Module}}}).
+
+-spec unregister(HookName::hookname(), Module::atom(),Priority::integer()) -> ok.
+unregister(HookName, Module,Priority) ->
+ gen_server:call(?MODULE, {unregister, HookName, {Priority, Module}}).
+
+%% @doc run all hooks registered for the HookName.
+%% Execution can be interrupted if an hook return the atom `stop'.
+-spec run(HookName::hookname(), Args::list()) -> ok.
+run(HookName,Fun, Args) ->
+ case find_hooks(HookName) of
+ no_hook -> ok;
+ Hooks -> run1(Hooks, HookName,Fun, Args)
+ end.
+
+run1([], _HookName,_Fun, _Args) ->
+ ok;
+run1([M | Rest], HookName, Fun, Args) ->
+ Ret = (catch apply(M, Fun, Args)),
+ case Ret of
+ {'EXIT', Reason} ->
+ logger:error("~p~n error running extension: ~p~n", [HookName, Reason]),
+ run1(Rest, HookName,Fun, Args);
+ stop ->
+ ok;
+ _ ->
+ run1(Rest, HookName,Fun, Args)
+ end.
+
+-spec run_fold(HookName::hookname(), Args::list(), Acc::any()) -> Acc2::any().
+run_fold(HookName, Fun, Args, Acc) ->
+ case find_hooks(HookName) of
+ no_hook -> Acc;
+ Hooks -> run_fold1(Hooks,HookName, Fun, Args, Acc)
+ end.
+
+
+run_fold1([], _HookName,_Fun, _Args, Acc) ->
+ Acc;
+run_fold1([M | Rest], HookName,Fun, Args0, Acc) ->
+ Args = Args0 ++ [Acc],
+ Ret = (catch apply(M, Fun, Args)),
+ case Ret of
+ {'EXIT', Reason} ->
+ error_logger:error_msg("~p~n error running hook: ~p~n",
+ [HookName, Reason]),
+ run_fold1(Rest, HookName,Fun,Args0, Acc);
+ stop ->
+ Acc;
+ {stop, NewAcc} ->
+ NewAcc;
+ _ ->
+ run_fold1(Rest, HookName,Fun,Args0, Ret)
+ end.
+
+
+
+
+%% @doc retrieve the lists of registered functions for an hook.
+-spec find(HookName::hookname()) -> {ok, [{atom(), atom()}]} | error.
+find(HookName) ->
+ case ?find_hook(HookName) of
+ no_hook -> error;
+ Hooks -> {ok, Hooks}
+ end.
+
+%% @hidden
+start_link() ->
+ _ = init_tabs(),
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+init_tabs() ->
+ case ets:info(?TAB, name) of
+ undefined ->
+ ets:new(?TAB, [ordered_set, public, named_table,
+ {read_concurrency, true},
+ {write_concurrency, true}]);
+ _ ->
+ true
+ end.
+
+%% @hidden
+init([]) ->
+ {ok, #state{}}.
+
+%% @hidden
+handle_call({register, HookName, {Priority, Module}}, _From, State) ->
+ do_register(HookName, {Priority, Module}),
+ {reply, ok, State};
+handle_call({unregister, HookName, {Priority, Module}}, _From, State) ->
+ do_unregister(HookName, {Priority, Module}),
+ {reply, ok, State};
+handle_call(_Msg, _From, State) ->
+ {reply, badarg, State}.
+
+%% @hidden
+handle_cast(_Msg, State) ->
+ {noreply, State}.
+
+handle_info(_Info, State) ->
+ {noreply, State}.
+
+%% @hidden
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+%% @hidden
+terminate(_Reason, _Srv) ->
+ ok.
+
+do_register(HookName, {_Priority, ModuleName}=Hook) ->
+ check_module(ModuleName),
+ update_hooks(HookName, [Hook]).
+
+
+do_unregister(HookName, Hook) ->
+ remove_hooks(HookName, [Hook]),
+ ok.
+
+update_hooks(HookName, HookFuns) ->
+ case ets:lookup(?TAB, HookName) of
+ [] ->
+ true = ets:insert(?TAB, {HookName, HookFuns});
+ [{_, Funs}] ->
+ Funs2 = lists:keysort(1, Funs ++ HookFuns),
+ true = ets:insert(?TAB, {HookName, Funs2})
+ end.
+
+remove_hooks(HookName, HookFuns) ->
+ case ets:lookup(?TAB, HookName) of
+ [] ->
+ ok;
+ [{_, Funs}] ->
+ Funs2 = Funs -- HookFuns,
+ case Funs2 of
+ [] ->
+ ets:delete(?TAB, HookName);
+ _ ->
+ ets:insert(?TAB, {HookName, Funs2})
+ end
+ end.
+
+check_module(ModuleName) ->
+ _ = code:ensure_loaded(ModuleName),
+ ok.
+
+find_hooks(HookName)->
+ case ets:lookup(?TAB,HookName) of
+ []->
+ no_hook;
+ [{_, Modules}]->
+ Modules
+ end.
diff --git a/src/dubbo_filter.erl b/src/dubbo_filter.erl
new file mode 100644
index 0000000..7fa7950
--- /dev/null
+++ b/src/dubbo_filter.erl
@@ -0,0 +1,21 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_filter).
+-author("dlive").
+
+%% API
+-export([]).
diff --git a/src/dubbo_invoker.erl b/src/dubbo_invoker.erl
index b29751c..f01cec4 100644
--- a/src/dubbo_invoker.erl
+++ b/src/dubbo_invoker.erl
@@ -18,6 +18,13 @@
-include("dubbo.hrl").
%% API
+-export([]).
+
+
+-callback(invoke(Invoker,Invocation) -> ok).
+
+
+%% API
-export([invoke_request/2, invoke_request/3, invoke_request/5]).
-spec invoke_request(Interface :: binary(), Request :: #dubbo_request{}) ->
@@ -40,7 +47,7 @@
{ok, reference(), Data :: any(), RpcContent :: list()}|
{error, Reason :: timeout|no_provider|request_full|any()}.
invoke_request(Interface, Request, RpcContext, RequestState, CallBackPid) ->
- case dubbo_consumer_pool:select_connection(Interface, Request#dubbo_request.mid) of
+ case dubbo_provider_consumer_reg_table:select_connection(Interface, Request#dubbo_request.mid) of
{ok, #connection_info{pid = Pid, host_flag = HostFlag}} ->
case dubbo_traffic_control:check_goon(HostFlag, 199) of
ok ->
diff --git a/src/dubbo_invoker_cluster.erl b/src/dubbo_invoker_cluster.erl
new file mode 100644
index 0000000..906a0fa
--- /dev/null
+++ b/src/dubbo_invoker_cluster.erl
@@ -0,0 +1,21 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_invoker_cluster).
+-author("dlive").
+
+%% API
+-export([]).
diff --git a/src/dubbo_invoker_old.erl b/src/dubbo_invoker_old.erl
new file mode 100644
index 0000000..354cef9
--- /dev/null
+++ b/src/dubbo_invoker_old.erl
@@ -0,0 +1,97 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_invoker_old).
+
+-include("dubbo.hrl").
+%% API
+-export([invoke_request/2, invoke_request/3, invoke_request/5]).
+
+-spec invoke_request(Interface :: binary(), Request :: #dubbo_request{}) ->
+ {ok, reference()}|
+ {ok, reference(), Data :: any(), RpcContent :: list()}|
+ {error, Reason :: timeout|no_provider|any()}.
+invoke_request(Interface, Request) ->
+ invoke_request(Interface, Request, [], #{}, self()).
+
+-spec invoke_request(Interface :: binary(), Request :: #dubbo_request{}, RequestOption :: map()) ->
+ {ok, reference()}|
+ {ok, reference(), Data :: any(), RpcContent :: list()}|
+ {error, Reason :: timeout|no_provider|any()}.
+invoke_request(Interface, Request, RequestOption) ->
+ invoke_request(Interface, Request, maps:get(ctx, RequestOption, []), RequestOption, self()).
+
+
+-spec invoke_request(Interface :: binary(), Request :: #dubbo_request{}, RpcContext :: list(), RequestState :: map(), CallBackPid :: pid()) ->
+ {ok, reference()}|
+ {ok, reference(), Data :: any(), RpcContent :: list()}|
+ {error, Reason :: timeout|no_provider|request_full|any()}.
+invoke_request(Interface, Request, RpcContext, RequestState, CallBackPid) ->
+ case dubbo_provider_consumer_reg_table:select_connection(Interface, Request#dubbo_request.mid) of
+ {ok, #connection_info{pid = Pid, host_flag = HostFlag}} ->
+ case dubbo_traffic_control:check_goon(HostFlag, 199) of
+ ok ->
+ Request2 = merge_attachments(Request, RpcContext),
+ {ok, RequestData} = dubbo_codec:encode_request(Request2),
+ Ref = get_ref(RequestState),
+ gen_server:cast(Pid, {send_request, Ref, Request2, RequestData, CallBackPid, RequestState}),
+ case is_sync(RequestState) of
+ true ->
+ sync_receive(Ref, get_timeout(RequestState));
+ false -> {ok, Ref}
+ end;
+ full ->
+ {error, request_full}
+ end;
+ {error, none} ->
+ logger:error("[INVOKE] ~p error Reason no_provider", [Interface]),
+ {error, no_provider}
+ end.
+
+
+is_sync(Option) ->
+ maps:is_key(sync, Option).
+get_ref(Option) ->
+ maps:get(ref, Option, make_ref()).
+
+get_timeout(Option) ->
+ maps:get(timeout, Option, ?REQUEST_TIME_OUT).
+
+
+sync_receive(Ref, TimeOut) ->
+ receive
+ {'$gen_cast', {response_process, Ref, RpcContent, Response}} ->
+ {ok, Ref, Response, RpcContent}
+ after
+ TimeOut ->
+ {error, timeout}
+ end.
+merge_attachments(#dubbo_request{data = null} = Request, _Option) ->
+ Request;
+merge_attachments(Request, Option) ->
+ Attachements = Request#dubbo_request.data#dubbo_rpc_invocation.attachments,
+ case lists:keyfind(attachments, 1, Option) of
+ false -> OptionAttachments = [];
+ {attachments, OptionAttachments} ->
+ OptionAttachments
+ end,
+ List = [
+ {<<"version">>, <<"0.0.0">>},
+ {<<"timeout">>, <<"5000">>}
+ ],
+ Attachements2 = lists:merge3(Attachements, OptionAttachments, List),
+ Data2 = Request#dubbo_request.data#dubbo_rpc_invocation{attachments = Attachements2},
+ Request#dubbo_request{data = Data2}.
diff --git a/src/dubbo_loadbalance_random.erl b/src/dubbo_loadbalance_random.erl
new file mode 100644
index 0000000..21d4f61
--- /dev/null
+++ b/src/dubbo_loadbalance_random.erl
@@ -0,0 +1,20 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_loadbalance_random).
+
+%% API
+-export([]).
diff --git a/src/dubbo_netty_client.erl b/src/dubbo_netty_client.erl
index 0181d33..06f9d1e 100644
--- a/src/dubbo_netty_client.erl
+++ b/src/dubbo_netty_client.erl
@@ -415,7 +415,7 @@
{ok, State}.
process_request(true, #dubbo_request{data = <<"R">>}, State) ->
- {ok, _} = dubbo_consumer_pool:update_connection_readonly(self(), true),
+ {ok, _} = dubbo_provider_consumer_reg_table:update_connection_readonly(self(), true),
{ok, State};
process_request(true, Request, State) ->
{ok, NewState} = send_heartbeat_msg(Request#dubbo_request.mid, false, State),
diff --git a/src/dubbo_protocol.erl b/src/dubbo_protocol.erl
new file mode 100644
index 0000000..7ecfcd8
--- /dev/null
+++ b/src/dubbo_protocol.erl
@@ -0,0 +1,26 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_protocol).
+
+-callback refer(Url,Acc)->ok.
+
+%%%% API
+%%-export([refer/2]).
+%%
+%%
+%%refer(InterfaceClassInfo,Url)->
+%% dubbo_hooker(protocol_wapper,refer,[InterfaceClassInfo,Url]).
\ No newline at end of file
diff --git a/src/dubbo_protocol_dubbo.erl b/src/dubbo_protocol_dubbo.erl
new file mode 100644
index 0000000..96947cc
--- /dev/null
+++ b/src/dubbo_protocol_dubbo.erl
@@ -0,0 +1,84 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_protocol_dubbo).
+
+-include("dubboerl.hrl").
+-include("dubbo.hrl").
+
+%% API
+-export([refer/2]).
+
+refer(Url, Acc) ->
+ {ok, UrlInfo} = dubbo_common_fun:parse_url(Url),
+ case UrlInfo#dubbo_url.scheme of
+ <<"dubbo">> ->
+ {ok,Invoker} = do_refer(UrlInfo),
+ {ok, Invoker};
+ _ ->
+ {skip, Acc}
+ end.
+
+do_refer(UrlInfo) ->
+ case dubbo_node_config_util:parse_provider_info(UrlInfo) of
+ {ok, ProviderConfig} ->
+%% OldHostList = dubbo_provider_consumer_reg_table:get_interface_provider_node(ProviderConfig#provider_config.interface),
+ case getClients(ProviderConfig) of
+ {ok, ConnectionInfoList} ->
+ dubbo_provider_consumer_reg_table:update_node_conections(ProviderConfig#provider_config.interface,ConnectionInfoList),
+ HostFlag = dubbo_provider_consumer_reg_table:get_host_flag(ProviderConfig),
+ {ok,#dubbo_invoker{host_flag = HostFlag,handle = ?MODULE}};
+ {error, Reason} ->
+ {error, Reason}
+ end;
+ {error, R1} ->
+ logger:error("parse provider info error reason ~p", [R1]),
+ {error, R1}
+ end.
+
+getClients(ProviderConfig) ->
+ %% @todo if connections parameter > 1, need new spec transport
+ case new_transport(ProviderConfig) of
+ {ok, ConnectionInfoList} ->
+%% ConnectionList = start_provider_process(HostFlag, 30, ProviderConfig),
+ {ok, ConnectionInfoList};
+ {error, Reason} ->
+ {error, Reason}
+ end.
+
+
+%%ok = update_connection_info(ProviderConfig#provider_config.interface, HostFlag, ConnectionList, true),
+
+
+new_transport(ProviderConfig) ->
+
+ HostFlag = get_host_flag(ProviderConfig),
+ case dubbo_provider_consumer_reg_table:get_host_connections(ProviderConfig#provider_config) of
+ [] ->
+ case dubbo_exchanger:connect(ProviderConfig, ?MODULE) of
+ {ok, ConnectionInfo} ->
+ {ok, [ConnectionInfo]};
+ {error, Reason} ->
+ logger:warning("start client fail ~p ~p", [Reason, HostFlag]),
+ {error, Reason}
+ end;
+ ConnectionInfoList ->
+ {ok, ConnectionInfoList}
+ end.
+
+
+
+
diff --git a/src/dubbo_protocol_registry.erl b/src/dubbo_protocol_registry.erl
new file mode 100644
index 0000000..b2fde17
--- /dev/null
+++ b/src/dubbo_protocol_registry.erl
@@ -0,0 +1,53 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_protocol_registry).
+-behaviour(dubbo_protocol).
+
+-include("dubboerl.hrl").
+
+%% API
+-export([refer/1]).
+
+refer(Url)->
+ {ok,UrlInfo} = dubbo_common_fun:parse_url(Url),
+
+ {ok,RegistryName} = dubbo_registry:setup_register(UrlInfo),
+
+ ConsumerUrl = gen_consumer_url(UrlInfo),
+ %% 通知directory
+ dubbo_registry:register(RegistryName,ConsumerUrl),
+
+ dubbo_directory:subscribe(RegistryName,ConsumerUrl),
+
+ %% return
+ ok.
+
+
+gen_consumer_url(UrlInfo)->
+ Parameters = UrlInfo#dubbo_url.parameters,
+ #{<<"refer">> := Refer} = Parameters,
+ Refer2 = http_uri:decode(Refer),
+ Parameters2 = dubbo_common_fun:parse_url(Refer2,#{}),
+ #{<<"interface">> := Interface} = Parameters2,
+ ConsumerUrlInfo = UrlInfo#dubbo_url{
+ scheme = <<"consumer">>,
+ host = dubbo_common_fun:local_ip_v4_str(),
+ path = Interface,
+ parameters = Parameters2
+ },
+ ConsumerUrl = dubbo_common_fun:url_to_binary(ConsumerUrlInfo),
+ ConsumerUrl.
\ No newline at end of file
diff --git a/src/dubbo_consumer_pool.erl b/src/dubbo_provider_consumer_reg_table.erl
similarity index 64%
rename from src/dubbo_consumer_pool.erl
rename to src/dubbo_provider_consumer_reg_table.erl
index 0a01d38..1c193b8 100644
--- a/src/dubbo_consumer_pool.erl
+++ b/src/dubbo_provider_consumer_reg_table.erl
@@ -14,7 +14,7 @@
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%------------------------------------------------------------------------------
--module(dubbo_consumer_pool).
+-module(dubbo_provider_consumer_reg_table).
-behaviour(gen_server).
@@ -29,12 +29,16 @@
terminate/2,
code_change/3]).
--export([select_connection/1, select_connection/2, update_connection_readonly/2]).
+-export([update_consumer_connections/2,update_node_conections/2,get_interface_provider_node/1,get_host_connections/2, select_connection/1,
+ select_connection/2, update_connection_readonly/2, get_host_flag/1, get_host_flag/2,clean_invalid_provider/1]).
-include("dubbo.hrl").
-define(SERVER, ?MODULE).
-define(INTERFCE_LIST_TABLE, interface_list).
+
+-define(INTERFAE_INFO_TABLE,dubbo_interface_info).
+
-define(PROVIDER_NODE_LIST_TABLE, provider_node_list).
-record(state, {}).
@@ -86,16 +90,21 @@
ok
catch
_Type:Reason ->
- logger:error("new ets table error ~p", [Reason]),
- error
+ logger:error("new ets table INTERFCE_LIST_TABLE error ~p", [Reason])
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
+ logger:error("new ets table PROVIDER_NODE_LIST_TABLE error ~p", [Reason1])
+ end,
+ try ets:new(?INTERFAE_INFO_TABLE, [public, named_table, {keypos, 2}]) of
+ ?INTERFAE_INFO_TABLE ->
+ ok
+ catch
+ _Type1:Reason1 ->
+ logger:error("new ets table PROVIDER_NODE_LIST_TABLE error ~p", [Reason1])
end,
ok.
%%--------------------------------------------------------------------
@@ -189,47 +198,93 @@
gen_server:call(?SERVER, {add_consumer, Interface, ProviderNodeInfo}).
+
+get_host_connections(Host, Port) ->
+ HostFlag = get_host_flag(Host, Port),
+ List = ets:lookup(?PROVIDER_NODE_LIST_TABLE, HostFlag),
+ List.
+
+update_interface_info(InterfaceInfo)->
+ ets:insert(?INTERFAE_INFO_TABLE,InterfaceInfo).
+
+
%%%===================================================================
%%% 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),
+%%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_transport_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.
+
+
+update_node_conections(HostFlag,Connections)->
+ lists:map(
+ fun(Item) ->
+ HostFlag= Item#connection_info.host_flag,
+ case ets:lookup_element(?PROVIDER_NODE_LIST_TABLE,#connection_info{host_flag = HostFlag,pid = Item#connection_info.pid,_="_"}) of
+ '$end_of_table' ->
+ I2 = ets:insert(?PROVIDER_NODE_LIST_TABLE, Item),
+ logger:debug("insert PROVIDER_NODE_LIST_TABLE ~p info:~p", [HostFlag, I2]);
+ _ ->
+ ok
+ end
+ end, Connections),
+ ok.
+
+update_consumer_connections(Interface, Connections) ->
+ lists:map(
+ fun(Item) ->
+ HostFlag= Item#connection_info.host_flag,
+
+ case ets:lookup_element(?PROVIDER_NODE_LIST_TABLE,#connection_info{host_flag = HostFlag,pid = Item#connection_info.pid,_="_"}) of
+ '$end_of_table' ->
+ I2 = ets:insert(?PROVIDER_NODE_LIST_TABLE, Item),
+ logger:debug("insert PROVIDER_NODE_LIST_TABLE ~p info:~p", [HostFlag, I2]);
+ {_ObjectList,_Continuation} ->
ok
end,
- add_consumer(ProviderList, [HostFlag] ++ RegisterList);
- {error, R1} ->
- logger:error("parse provider info error reason ~p", [R1]),
- add_consumer(ProviderList, RegisterList)
- end.
+ I1 = ets:insert(?INTERFCE_LIST_TABLE, #interface_list{interface = Interface, pid = Item#connection_info.pid, connection_info = Item}),
+ logger:debug("insert interface conection info ~p ~p ~p", [Interface, Item#connection_info.pid, I1]),
+ ok
+ end, Connections),
+ ok.
-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.
+get_host_flag(Host, Port) ->
+ <<(list_to_binary(Host))/binary, <<"_">>/binary, (integer_to_binary(Port))/binary>>.
update_connection_info(Interface, HostFlag, ConnectionList, IsUpdateProvideNode) ->
lists:map(fun(Item) ->
@@ -237,7 +292,7 @@
logger:debug("insert interface conection info ~p ~p ~p", [Interface, Item#connection_info.pid, I1]),
case IsUpdateProvideNode of
true ->
- I2 = ets:insert(?PROVIDER_NODE_LIST_TABLE, #provider_node_list{host_flag = HostFlag, connection_info = Item}),
+ I2 = ets:insert(?PROVIDER_NODE_LIST_TABLE, Item),
logger:debug("insert PROVIDER_NODE_LIST_TABLE ~p info:~p", [HostFlag, I2]);
false ->
ok
@@ -288,18 +343,17 @@
case ets:lookup(?PROVIDER_NODE_LIST_TABLE, HostFlag) of
[] ->
ok;
- ProviderNodeList ->
- ProviderNodeList1 = dubbo_lists_util:del_duplicate(ProviderNodeList),
- clean_connection_info(ProviderNodeList1)
+ ProviderNodeConnections ->
+ ProviderNodeConnections1 = dubbo_lists_util:del_duplicate(ProviderNodeConnections),
+ clean_connection_info(ProviderNodeConnections1)
end,
clean_invalid_provider(DeleteProverList).
-clean_connection_info(ProviderNodeList) ->
+clean_connection_info(ProviderNodeConnections) ->
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,
+ Pid = Item#connection_info.pid,
Pattern = #interface_list{pid = Pid, _ = '_'},
ets:delete_object(?INTERFCE_LIST_TABLE, Pattern),
- dubbo_consumer_pool_sup:stop_children(ConnectionId)
- end, ProviderNodeList),
+ dubbo_transport_pool_sup:stop_children(Pid)
+ end, ProviderNodeConnections),
ok.
\ No newline at end of file
diff --git a/src/dubbo_reference_config.erl b/src/dubbo_reference_config.erl
new file mode 100644
index 0000000..7ab7f86
--- /dev/null
+++ b/src/dubbo_reference_config.erl
@@ -0,0 +1,111 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_reference_config).
+
+-include("dubbo.hrl").
+-include("dubboerl.hrl").
+
+-record(dubbo_interface_info,{}).
+
+%% API
+-export([init_reference/1]).
+
+init_reference(ConsumerInfo)->
+%% InitConfigMap= #{
+%%
+%% },
+ %% 组装各类需要数据
+ create_proxy(ConsumerInfo),
+ ok.
+
+
+create_proxy(ConsumerInfo)->
+
+
+
+ Para = gen_parameter(ConsumerInfo),
+ Url = gen_registry_url(Para),
+ dubbo_extension:run(protocol_wapper,refer,[Url]),
+ ok.
+
+ %%application=hello-world&dubbo=2.0.2&pid=68901&refer=application=hello-world&default.check=false&default.lazy=false&default.retries=0&default.sticky=false&default.timeout=300000&dubbo=2.0.2&interface=org.apache.dubbo.erlang.sample.service.facade.UserOperator&lazy=false&methods=queryUserInfo,queryUserList,genUserId,getUserInfo&pid=68901®ister.ip=127.0.0.1&release=2.7.1&retries=0&side=consumer&sticky=false×tamp=1559727789953®istry=zookeeper&release=2.7.1×tamp=1559727842451
+
+
+gen_registry_url(Para)->
+ %%todo 组装para & url
+ {Host,Port} = get_registry_host_port(),
+ UrlInfo = #dubbo_url{
+ scheme = <<"registry">>,
+ host = list_to_binary(Host),
+ port = integer_to_binary(Port),
+ path = <<"org.apache.dubbo.registry.RegistryService">>,
+ parameters = Para
+ },
+ dubbo_common_fun:url_to_binary(UrlInfo).
+%% Url = "registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=hello-world&dubbo=2.0.2&pid=68901&refer=application%3Dhello-world%26default.check%3Dfalse%26default.lazy%3Dfalse%26default.retries%3D0%26default.sticky%3Dfalse%26default.timeout%3D300000%26dubbo%3D2.0.2%26interface%3Dorg.apache.dubbo.erlang.sample.service.facade.UserOperator%26lazy%3Dfalse%26methods%3DqueryUserInfo%2CqueryUserList%2CgenUserId%2CgetUserInfo%26pid%3D68901%26register.ip%3D127.0.0.1%26release%3D2.7.1%26retries%3D0%26side%3Dconsumer%26sticky%3Dfalse%26timestamp%3D1559727789953®istry=zookeeper&release=2.7.1×tamp=1559727842451",
+%% Url.
+
+get_registry_host_port()->
+ %% @todo need adapter other registry
+ RegistryList = application:get_env(dubboerl,zookeeper_list,[{"127.0.0.1",2181}]),
+ [Item|_] = RegistryList,
+ Item.
+
+gen_parameter(ConsumerInfo)->
+ Para = #{
+ <<"application">> => get_appname(ConsumerInfo),
+ <<"dubbo">> => <<"2.0.2">>,
+ <<"pid">> => get_pid(),
+ <<"refer">> => get_refinfo(ConsumerInfo),
+ <<"registry">> => get_registry_type(),
+ <<"release">> => <<"2.7.1">>,
+ <<"timestamp">> => integer_to_binary(dubbo_time_util:timestamp_ms())
+ },
+
+ Para.
+
+get_appname(ConsumerInfo)->
+ ConsumerInfo#consumer_config.application.
+get_pid()->
+ os:getpid().
+get_refinfo(ConsumerInfo)->
+ KeyValues=[
+ {"application",ConsumerInfo#consumer_config.application},
+ {"default.check",ConsumerInfo#consumer_config.check},
+ {"default.lazy","false"},
+ {"default.retries","0"},
+ {"default.sticky","false"},
+ {"default.timeout","300000"},
+ {"dubbo","2.0.2"},
+ {"interface",ConsumerInfo#consumer_config.interface},
+ {"lazy","false"},
+ {"methods",ConsumerInfo#consumer_config.methods},
+ {"register.ip",ConsumerInfo#consumer_config.application},
+ {"release","2.7.1"},
+ {"pid",get_pid()},
+ {"side","consumer"},
+ {"sticky","false"},
+ {"timestamp",dubbo_time_util:timestamp_ms()}
+ ],
+ KeyValues2 = [io_lib:format("~s=~p", [Key, Value]) || {Key, Value} <= KeyValues],
+ ParameterStr1 = string:join(KeyValues2, "&"),
+ list_to_binary(http_uri:encode(ParameterStr1)).
+%% <<"application%3Dhello-world%26default.check%3Dfalse%26default.lazy%3Dfalse%26default.retries%3D0%26default.sticky%3Dfalse%26default.timeout%3D300000%26dubbo%3D2.0.2%26interface%3Dorg.apache.dubbo.erlang.sample.service.facade.UserOperator%26lazy%3Dfalse%26methods%3DqueryUserInfo%2CqueryUserList%2CgenUserId%2CgetUserInfo%26pid%3D68901%26register.ip%3D127..0.1%26release%3D2.7.1%26retries%3D0%26side%3Dconsumer%26sticky%3Dfalse%26timestamp%3D1559727789953">>.
+
+get_registry_type()->
+ %%todo
+ atom_to_binary(application:get_env(dubboerl,registry,zookeeper)).
\ No newline at end of file
diff --git a/src/dubbo_registry.erl b/src/dubbo_registry.erl
new file mode 100644
index 0000000..89eb64b
--- /dev/null
+++ b/src/dubbo_registry.erl
@@ -0,0 +1,47 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_registry).
+-include("dubboerl.hrl").
+
+-callback start(Url :: binary) -> ok.
+-callback register(Url::binary())-> term().
+-callback subscribe(SubcribeUrl::binary(),NotifyFun::function())->ok.
+
+%% API
+-export([setup_register/1,register/2]).
+
+-spec(setup_register(UrlInfo :: map()) -> {ok, RegistryProcessName :: atom()}|{error, term()}).
+setup_register(UrlInfo) ->
+ RegistryModuleName = get_registry_module(UrlInfo),
+ case whereis(RegistryModuleName) of
+ undefined ->
+ apply(RegistryModuleName, start, [UrlInfo]),
+ {ok, RegistryModuleName};
+ _ ->
+ {ok, RegistryModuleName}
+ end.
+
+register(RegistryName,Url) ->
+ logger:info("call ~p register url ~p",[RegistryName,Url]),
+ Result = apply(RegistryName,register,[Url]),
+ Result.
+
+
+get_registry_module(Info) ->
+ RegistryName = Info#dubbo_url.scheme,
+ FullName = << <<"dubbo_registry_">>, RegistryName/binary>>,
+ binary_to_existing_atom(FullName).
\ No newline at end of file
diff --git a/src/dubbo_registry_zookeeper.erl b/src/dubbo_registry_zookeeper.erl
new file mode 100644
index 0000000..5d14588
--- /dev/null
+++ b/src/dubbo_registry_zookeeper.erl
@@ -0,0 +1,314 @@
+%%------------------------------------------------------------------------------
+%% 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.
+%%------------------------------------------------------------------------------
+-module(dubbo_registry_zookeeper).
+-behaviour(gen_server).
+-behaviour(dubbo_registry).
+
+-include("dubbo.hrl").
+-include("dubboerl.hrl").
+%% API
+-export([start_link/0, register_consumer/1, register_consumer/2, gen_consumer_node_info/1, register_provider/1, provider_watcher/1]).
+
+-export([start/1,register/1,subscribe/2]).
+%% gen_server callbacks
+-export([init/1,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2,
+ terminate/2,
+ code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+-record(state, {zk_pid,notify_fun}).
+
+%%%===================================================================
+%%% 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([]) ->
+ {ok, Pid} = connection(),
+ {ok, #state{zk_pid = Pid}}.
+
+%%--------------------------------------------------------------------
+%% @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,ConsumerUrl}, _From, State) ->
+ add_consumer(Interface,ConsumerUrl, State),
+ {reply, ok, State};
+handle_call({add_provider, Provider}, _From, State) ->
+ register_provider_path(Provider, State),
+ {reply, ok, State};
+handle_call({subscribe_provider,InterfaceName,NotifyFun}, _From, #state{zk_pid = ZkPid} = State) ->
+ NewState=State#state{notify_fun = NotifyFun},
+ get_provider_list(InterfaceName,ZkPid,NotifyFun),
+ {reply, ok, NewState};
+
+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({provider_node_change, Interface, Path}, #state{zk_pid = Pid} = State) ->
+ get_provider_and_start(Pid, Interface, Path),
+ {noreply, 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) ->
+ logger:info("zk server recv msg:~p", [_Info]),
+ {noreply, State}.
+
+%%--------------------------------------------------------------------
+%% @private
+%% @doc
+%% This function is called by a gen_server 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}.
+
+
+
+%%----------------------------------------------
+%% dubbo_registry
+%%----------------------------------------------
+start(Url) ->
+ ok.
+register(Url)->
+ {ok,UrlInfo} = dubbo_common_fun:parse_url(Url),
+ InterfaceName = maps:get(<<"interface">>,UrlInfo#dubbo_url.parameters),
+ register(UrlInfo#dubbo_url.scheme,InterfaceName,Url),
+ ok.
+
+register(<<"consumer">>,InterfaceName,Url)->
+ gen_server:call(?SERVER, {add_consumer,InterfaceName, Url}),
+ ok;
+register(<<"provider">>,InterfaceName,Url)->
+
+ ok.
+
+subscribe(SubcribeUrl,NotifyFun)->
+ {ok,UrlInfo} = dubbo_common_fun:parse_url(SubcribeUrl),
+ InterfaceName = maps:get(<<"interface">>,UrlInfo#dubbo_url.parameters),
+ try gen_server:call(?SERVER,{subscribe_provider,InterfaceName,NotifyFun},5000) of
+ ok->
+ ok
+ catch
+ Error:Reason->
+ %%todo improve error type
+ {error,Reason}
+ end.
+
+register_consumer(Consumer) ->
+ gen_server:call(?SERVER, {add_consumer, Consumer}),
+ ok.
+register_consumer(Name, Option) ->
+ Consumer = #consumer_config{interface = Name, methods = [<<"testa">>, <<"testb">>]},
+ register_consumer(Consumer),
+ ok.
+register_provider(Provider) ->
+ gen_server:call(?SERVER, {add_provider, Provider}),
+ ok.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+connection() ->
+ {ok, List} = application:get_env(dubboerl, zookeeper_list),
+ {ok, Pid} = erlzk:connect(List, 30000, [
+ {chroot, "/"},
+ {monitor, self()}]),
+ {ok, Pid}.
+
+add_consumer(InterfaceName,ConsumerUrl, State) ->
+ Pid = State#state.zk_pid,
+%% ConsumerNode = gen_consumer_node_info(Consumer),
+ ConsumerNode2 = list_to_binary(edoc_lib:escape_uri(binary_to_list(ConsumerUrl))),
+ check_and_create_path(Pid, <<"">>, [{<<"dubbo">>, p}, {InterfaceName, p}, {<<"consumers">>, p}, {ConsumerNode2, e}]),
+ %% todo
+%% get_provider_list(Consumer, State),
+ ok.
+register_provider_path(Provider, State) ->
+ #state{zk_pid = Pid} = State,
+ ProviderNode = dubbo_node_config_util:gen_provider_info(Provider),
+ check_and_create_path(Pid, <<"">>, [{<<"dubbo">>, p}, {Provider#provider_config.interface, p}, {<<"providers">>, p}, {ProviderNode, e}]),
+ ok.
+
+
+get_provider_list(InterfaceName,ZkPid,NotifyFun) ->
+ InterfacePath = <<<<"/dubbo/">>/binary, InterfaceName/binary, <<"/providers">>/binary>>,
+ ChildList= get_provider_and_start(ZkPid, InterfaceName, InterfacePath),
+ NotifyFun(InterfaceName,ChildList),
+ ok.
+get_provider_and_start(Pid, Interface, Path) ->
+ case erlzk:get_children(Pid, Path, spawn(dubbo_registry_zookeeper, provider_watcher, [Interface])) of
+ {ok, ChildList} ->
+ logger:debug("get provider list ~p", [ChildList]),
+%% start_provider_process(Interface, ChildList),
+ ChildList;
+ {error, R1} ->
+ logger:debug("[add_consumer] get_provider_list error ~p ~p", [R1]),
+ []
+ end.
+
+provider_watcher(Interface) ->
+ receive
+ {node_children_changed, Path} ->
+ gen_server:cast(?SERVER, {provider_node_change, Interface, Path}),
+ logger:debug("provider_watcher get event ~p ~p", [node_children_changed, Path]);
+ {Event, Path} ->
+%% Path = "/a",
+%% Event = node_created
+ logger:debug("provider_watcher get event ~p ~p", [Event, Path])
+ end,
+ ok.
+
+
+create_path(Pid, Path, CreateType) ->
+ case erlzk:create(Pid, Path, CreateType) of
+ {ok, ActualPath} ->
+ logger:debug("[add_consumer] create zk path success ~p", [ActualPath]),
+ ok;
+ {error, R1} ->
+ logger:debug("[add_consumer] create zk path error ~p ~p", [Path, R1])
+ end,
+ ok.
+check_and_create_path(_Pid, _RootPath, []) ->
+ ok;
+check_and_create_path(Pid, RootPath, [{Item, CreateType} | Rst]) ->
+ CheckPath = <<RootPath/binary, <<"/">>/binary, Item/binary>>,
+ case erlzk:exists(Pid, CheckPath) of
+ {ok, Stat} ->
+ check_and_create_path(Pid, CheckPath, Rst);
+ {error, no_node} ->
+ logger:debug("[add_consumer] check_and_create_path unexist no_node ~p", [CheckPath]),
+ create_path(Pid, CheckPath, CreateType),
+ check_and_create_path(Pid, CheckPath, Rst);
+ {error, R1} ->
+ logger:debug("[add_consumer] check_and_create_path unexist ~p", [R1]),
+ check_and_create_path(Pid, CheckPath, Rst)
+ end.
+
+gen_consumer_node_info(Consumer) ->
+ %% revision参数字段的作用是什么？ 暂时不添加
+ Methods = dubbo_lists_util:join(Consumer#consumer_config.methods, <<",">>),
+ Value = io_lib:format(<<"consumer://~s/~s?application=~s&category=~s&check=~p&default.timeout=~p&dubbo=~s&interface=~s&methods=~s&side=~s×tamp=~p">>,
+ [dubbo_common_fun:local_ip_v4_str(),
+ Consumer#consumer_config.interface,
+ Consumer#consumer_config.application,
+ Consumer#consumer_config.category,
+ Consumer#consumer_config.check,
+ Consumer#consumer_config.default_timeout,
+ Consumer#consumer_config.dubbo_version,
+ Consumer#consumer_config.interface,
+ Methods,
+ Consumer#consumer_config.side,
+ dubbo_time_util:timestamp_ms()
+ ]),
+ list_to_binary(Value).
+
+%%dubbo_zookeeper:register_consumer(<<"com.ifcoder.abcd">>,[]).
+start_provider_process(Interface, ProviderList) ->
+ dubbo_provider_consumer_reg_table:start_consumer(Interface, ProviderList).
\ No newline at end of file
diff --git a/src/dubbo_consumer_pool_sup.erl b/src/dubbo_transport_pool_sup.erl
similarity index 87%
rename from src/dubbo_consumer_pool_sup.erl
rename to src/dubbo_transport_pool_sup.erl
index 77a6dbe..019c57d 100644
--- a/src/dubbo_consumer_pool_sup.erl
+++ b/src/dubbo_transport_pool_sup.erl
@@ -14,12 +14,12 @@
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%------------------------------------------------------------------------------
--module(dubbo_consumer_pool_sup).
+-module(dubbo_transport_pool_sup).
-behaviour(supervisor).
%% API
--export([start_link/0, add_children/1, stop_children/1]).
+-export([start_link/0, add_children/2, stop_children/1]).
%% Supervisor callbacks
-export([init/1]).
@@ -63,17 +63,18 @@
ignore |
{error, Reason :: term()}).
init([]) ->
- RestartStrategy = one_for_one,
+ RestartStrategy = simple_one_for_one,
MaxRestarts = 1000,
MaxSecondsBetweenRestarts = 3600,
SupFlags = {RestartStrategy, MaxRestarts, MaxSecondsBetweenRestarts},
-
- {ok, {SupFlags, []}}.
+ Child = {dubbo_client_default, {dubbo_client_default, start_link, []}, permanent, 2000, worker, [dubbo_client_default]},
+ {ok, {SupFlags, [Child]}}.
-add_children(ChildSpec) ->
- supervisor:start_child(?SERVER, ChildSpec).
+add_children(ProvideConfig, Handler) ->
+ supervisor:start_child(?SERVER, [ProvideConfig, Handler]).
+
stop_children(ChildID) ->
supervisor:terminate_child(?SERVER, ChildID).
%%%===================================================================
diff --git a/src/dubbo_zookeeper.erl b/src/dubbo_zookeeper.erl
index f62ace1..84a95d6 100644
--- a/src/dubbo_zookeeper.erl
+++ b/src/dubbo_zookeeper.erl
@@ -272,5 +272,5 @@
%%dubbo_zookeeper:register_consumer(<<"com.ifcoder.abcd">>,[]).
start_provider_process(Interface, ProviderList) ->
- dubbo_consumer_pool:start_consumer(Interface, ProviderList).
+ dubbo_provider_consumer_reg_table:start_consumer(Interface, ProviderList).
diff --git a/src/dubboerl.erl b/src/dubboerl.erl
index 0a5f5bc..03cc8a0 100644
--- a/src/dubboerl.erl
+++ b/src/dubboerl.erl
@@ -33,8 +33,9 @@
ApplicationName = application:get_env(dubboerl, application, <<"defaultApplication">>),
lists:map(fun({Interface, Option}) ->
ConsumerInfo = dubbo_config_util:gen_consumer(ApplicationName, Interface, Option),
- dubbo_zookeeper:register_consumer(ConsumerInfo),
- logger:info("register consumer success ~p", [Interface])
+%% dubbo_zookeeper:register_consumer(ConsumerInfo),
+ dubbo_reference_config:init_reference(ConsumerInfo),
+ logger:info("consumer refer success ~p", [Interface])
end, ConsumerList),
ok.
diff --git a/src/dubboerl_app.erl b/src/dubboerl_app.erl
index 223e842..0f51768 100644
--- a/src/dubboerl_app.erl
+++ b/src/dubboerl_app.erl
@@ -27,9 +27,14 @@
%%====================================================================
start(_StartType, _StartArgs) ->
- io:format("[START] dubbo framework server start~n"),
-%% env_init(),
- dubboerl_sup:start_link().
+ logger:info("[START] dubbo framework server start"),
+ case dubboerl_sup:start_link() of
+ {ok,Pid} ->
+ init_default_hooks(),
+ {ok,Pid};
+ Result ->
+ Result
+ end.
%%--------------------------------------------------------------------
stop(_State) ->
@@ -38,6 +43,11 @@
%%====================================================================
%% Internal functions
%%====================================================================
+init_default_hooks()->
+ dubbo_extension:register(protocol,dubbo_protocol_dubbo,10),
+ dubbo_extension:register(protocol_wapper,dubbo_protocol_registry,10),
+
+ ok.
env_init() ->
ets:new(?PROVIDER_IMPL_TABLE, [public, named_table]),
dubbo_traffic_control:init(),
diff --git a/src/dubboerl_sup.erl b/src/dubboerl_sup.erl
index a829015..a0d2fb0 100644
--- a/src/dubboerl_sup.erl
+++ b/src/dubboerl_sup.erl
@@ -45,8 +45,8 @@
%% NettySer = {dubbo_netty_client,{dubbo_netty_client, start_link, []},transient,5000,worker,[dubbo_netty_client]},
Id_count = {dubbo_id_generator, {dubbo_id_generator, start_link, []}, transient, 5000, worker, [dubbo_id_generator]},
ProviderPoolSup = {dubbo_provider_worker_sup, {dubbo_provider_worker_sup, start_link, []}, transient, 5000, supervisor, [dubbo_provider_worker_sup]},
- ConsumerPoolSup = {dubbo_consumer_pool_sup, {dubbo_consumer_pool_sup, start_link, []}, transient, 5000, supervisor, [dubbo_consumer_pool_sup]},
- ConsumerPool = {dubbo_consumer_pool, {dubbo_consumer_pool, start_link, []}, transient, 5000, worker, [dubbo_consumer_pool]},
+ ConsumerPoolSup = {dubbo_transport_pool_sup, {dubbo_transport_pool_sup, start_link, []}, transient, 5000, supervisor, [dubbo_transport_pool_sup]},
+ ConsumerPool = {dubbo_provider_consumer_reg_table, {dubbo_provider_consumer_reg_table, start_link, []}, transient, 5000, worker, [dubbo_provider_consumer_reg_table]},
ListNew1 =
case application:get_env(dubboerl, registry, false) of
true ->
diff --git a/test/dubbo_consumer_pool_tests.erl b/test/dubbo_consumer_pool_tests.erl
index 740ed84..0a67f1a 100644
--- a/test/dubbo_consumer_pool_tests.erl
+++ b/test/dubbo_consumer_pool_tests.erl
@@ -21,13 +21,13 @@
-include("dubbo.hrl").
update_readonly_test() ->
- dubbo_consumer_pool:start_link(),
+ dubbo_provider_consumer_reg_table:start_link(),
InterfaceName= <<"testinterfacename">>,
HostFalg= <<"127.0.0.1/20880">>,
ConnectionList = [
#connection_info{connection_id=1,pid= testpid,weight = 30,host_flag = HostFalg},
#connection_info{connection_id=2,pid= testpid2,weight = 30,host_flag = HostFalg}
],
- dubbo_consumer_pool:update_connection_info(InterfaceName,HostFalg,ConnectionList,true),
- {ok,Size} = dubbo_consumer_pool:update_connection_readonly(testpid,false),
+ dubbo_provider_consumer_reg_table:update_connection_info(InterfaceName,HostFalg,ConnectionList,true),
+ {ok,Size} = dubbo_provider_consumer_reg_table:update_connection_readonly(testpid,false),
?assertEqual(1,Size).
diff --git a/test/userOperator.erl b/test/userOperator.erl
index 0d35917..e8567ae 100644
--- a/test/userOperator.erl
+++ b/test/userOperator.erl
@@ -67,7 +67,7 @@
]
},
Request = dubbo_adapter:reference(Data),
- dubbo_invoker:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
+ dubbo_invoker_old:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
-spec genUserId()->
@@ -96,7 +96,7 @@
]
},
Request = dubbo_adapter:reference(Data),
- dubbo_invoker:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
+ dubbo_invoker_old:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
-spec queryUserInfo(Arg0::#userInfoRequest{})->
@@ -127,7 +127,7 @@
]
},
Request = dubbo_adapter:reference(Data),
- dubbo_invoker:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
+ dubbo_invoker_old:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
-spec queryUserList(Arg0::list())->
@@ -158,7 +158,7 @@
]
},
Request = dubbo_adapter:reference(Data),
- dubbo_invoker:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
+ dubbo_invoker_old:invoke_request(?CURRENT_CLASS_NAME,Request,RequestOption).
test() ->
queryUserInfo(#userInfoRequest{username = "name", requestId = "111"}, #{sync=> true}).
\ No newline at end of file
