diff options
Diffstat (limited to 'lib')
11 files changed, 0 insertions, 1060 deletions
diff --git a/lib/dialyzer/test/small_SUITE_data/results/comm_layer b/lib/dialyzer/test/small_SUITE_data/results/comm_layer deleted file mode 100644 index cb4bf14eb4..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/results/comm_layer +++ /dev/null @@ -1,2 +0,0 @@ - -comm_layer.erl:76: Invalid type specification for function 'comm_layer_dir.comm_layer':this/0. The success typing is () -> {_,integer(),pid()} diff --git a/lib/dialyzer/test/small_SUITE_data/results/pubsub b/lib/dialyzer/test/small_SUITE_data/results/pubsub deleted file mode 100644 index e69de29bb2..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/results/pubsub +++ /dev/null diff --git a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_acceptor.erl b/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_acceptor.erl deleted file mode 100644 index 2ca1468911..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_acceptor.erl +++ /dev/null @@ -1,119 +0,0 @@ -% Copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -% -% Licensed 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. -%%%------------------------------------------------------------------- -%%% File : comm_acceptor.erl -%%% Author : Thorsten Schuett <[email protected]> -%%% Description : Acceptor -%%% This module accepts new connections and starts corresponding -%%% comm_connection processes. -%%% -%%% Created : 18 Apr 2008 by Thorsten Schuett <[email protected]> -%%%------------------------------------------------------------------- -%% @author Thorsten Schuett <[email protected]> -%% @copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -%% @version $Id $ --module(comm_layer_dir.comm_acceptor). - --export([start_link/1, init/2]). - --import(config). --import(gen_tcp). --import(inet). --import(log). --import(lists). --import(process_dictionary). - -start_link(InstanceId) -> - Pid = spawn_link(comm_layer_dir.comm_acceptor, init, [InstanceId, self()]), - receive - {started} -> - {ok, Pid} - end. - -init(InstanceId, Supervisor) -> - process_dictionary:register_process(InstanceId, acceptor, self()), - erlang:register(comm_layer_acceptor, self()), - log:log(info,"[ CC ] listening on ~p:~p", [config:listenIP(), config:listenPort()]), - LS = case config:listenIP() of - undefined -> - open_listen_port(config:listenPort(), first_ip()); - _ -> - open_listen_port(config:listenPort(), config:listenIP()) - end, - {ok, {_LocalAddress, LocalPort}} = inet:sockname(LS), - comm_port:set_local_address(undefined, LocalPort), - %io:format("this() == ~w~n", [{LocalAddress, LocalPort}]), - Supervisor ! {started}, - server(LS). - -server(LS) -> - case gen_tcp:accept(LS) of - {ok, S} -> - case comm_port:get_local_address_port() of - {undefined, LocalPort} -> - {ok, {MyIP, _LocalPort}} = inet:sockname(S), - comm_port:set_local_address(MyIP, LocalPort); - _ -> - ok - end, - receive - {tcp, S, Msg} -> - {endpoint, Address, Port} = binary_to_term(Msg), - % auto determine remote address, when not sent correctly - NewAddress = if Address =:= {0,0,0,0} orelse Address =:= {127,0,0,1} -> - case inet:peername(S) of - {ok, {PeerAddress, _Port}} -> - % io:format("Sent Address ~p\n",[Address]), - % io:format("Peername is ~p\n",[PeerAddress]), - PeerAddress; - {error, _Why} -> - % io:format("Peername error ~p\n",[Why]). - Address - end; - true -> - % io:format("Address is ~p\n",[Address]), - Address - end, - NewPid = comm_connection:new(NewAddress, Port, S), - gen_tcp:controlling_process(S, NewPid), - inet:setopts(S, [{active, once}, {send_timeout, config:read(tcp_send_timeout)}]), - comm_port:register_connection(NewAddress, Port, NewPid, S) - end, - server(LS); - Other -> - log:log(warn,"[ CC ] unknown message ~p", [Other]) - end. - -open_listen_port({From, To}, IP) -> - open_listen_port(lists:seq(From, To), IP); -open_listen_port([Port | Rest], IP) -> - case gen_tcp:listen(Port, [binary, {packet, 4}, {reuseaddr, true}, - {active, once}, {ip, IP}]) of - {ok, Socket} -> - Socket; - {error, Reason} -> - log:log(error,"[ CC ] can't listen on ~p: ~p~n", [Port, Reason]), - open_listen_port(Rest, IP) - end; -open_listen_port([], _) -> - abort; -open_listen_port(Port, IP) -> - open_listen_port([Port], IP). - --include_lib("kernel/include/inet.hrl"). - -first_ip() -> - {ok, Hostname} = inet:gethostname(), - {ok, HostEntry} = inet:gethostbyname(Hostname), - erlang:hd(HostEntry#hostent.h_addr_list). diff --git a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_connection.erl b/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_connection.erl deleted file mode 100644 index 5a8f9710d6..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_connection.erl +++ /dev/null @@ -1,207 +0,0 @@ -%% -*- coding: utf-8 -*- -% Copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -% -% Licensed 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. -%%%------------------------------------------------------------------- -%%% File : comm_connection.erl -%%% Author : Thorsten Schuett <[email protected]> -%%% Description : creates and destroys connections and represents the -%%% endpoint of a connection where messages are received and -%% send from/to the network. -%%% -%%% Created : 18 Apr 2008 by Thorsten Schuett <[email protected]> -%%%------------------------------------------------------------------- -%% @author Thorsten Schuett <[email protected]> -%% @copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -%% @version $Id $ --module(comm_layer_dir.comm_connection). - --export([send/3, open_new/4, new/3, open_new_async/4]). - --import(config). --import(gen_tcp). --import(inet). --import(io). --import(io_lib). --import(log). --import(timer). - --include("comm_layer.hrl"). - -%% @doc new accepted connection. called by comm_acceptor -%% @spec new(inet:ip_address(), int(), socket()) -> pid() -new(Address, Port, Socket) -> - spawn(fun () -> loop(Socket, Address, Port) end). - -%% @doc open new connection -%% @spec open_new(inet:ip_address(), int(), inet:ip_address(), int()) -> -%% {local_ip, inet:ip_address(), int(), pid(), inet:socket()} -%% | fail -%% | {connection, pid(), inet:socket()} -open_new(Address, Port, undefined, MyPort) -> - Myself = self(), - LocalPid = spawn(fun () -> - case new_connection(Address, Port, MyPort) of - fail -> - Myself ! {new_connection_failed}; - Socket -> - {ok, {MyIP, _MyPort}} = inet:sockname(Socket), - Myself ! {new_connection_started, MyIP, MyPort, Socket}, - loop(Socket, Address, Port) - end - end), - receive - {new_connection_failed} -> - fail; - {new_connection_started, MyIP, MyPort, S} -> - {local_ip, MyIP, MyPort, LocalPid, S} - end; -open_new(Address, Port, _MyAddress, MyPort) -> - Owner = self(), - LocalPid = spawn(fun () -> - case new_connection(Address, Port, MyPort) of - fail -> - Owner ! {new_connection_failed}; - Socket -> - Owner ! {new_connection_started, Socket}, - loop(Socket, Address, Port) - end - end), - receive - {new_connection_failed} -> - fail; - {new_connection_started, Socket} -> - {connection, LocalPid, Socket} - end. - -% =============================================================================== -% @doc open a new connection asynchronously -% =============================================================================== --spec(open_new_async/4 :: (any(), any(), any(), any()) -> pid()). -open_new_async(Address, Port, _MyAddr, MyPort) -> - Pid = spawn(fun () -> - case new_connection(Address, Port, MyPort) of - fail -> - comm_port:unregister_connection(Address, Port), - ok; - Socket -> - loop(Socket, Address, Port) - end - end), - Pid. - - -send({Address, Port, Socket}, Pid, Message) -> - BinaryMessage = term_to_binary({deliver, Pid, Message}), - SendTimeout = config:read(tcp_send_timeout), - {Time, Result} = timer:tc(gen_tcp, send, [Socket, BinaryMessage]), - if - Time > 1200 * SendTimeout -> - log:log(error,"[ CC ] send to ~p took ~p: ~p", - [Address, Time, inet:getopts(Socket, [keep_alive, send_timeout])]); - true -> - ok - end, - case Result of - ok -> - ?LOG_MESSAGE(erlang:element(1, Message), byte_size(BinaryMessage)), - ok; - {error, closed} -> - comm_port:unregister_connection(Address, Port), - close_connection(Socket); - {error, _Reason} -> - %log:log(error,"[ CC ] couldn't send to ~p:~p (~p)", [Address, Port, Reason]), - comm_port:unregister_connection(Address, Port), - close_connection(Socket) - end. - -loop(fail, Address, Port) -> - comm_port:unregister_connection(Address, Port), - ok; -loop(Socket, Address, Port) -> - receive - {send, Pid, Message} -> - case send({Address, Port, Socket}, Pid, Message) of - ok -> loop(Socket, Address, Port); - _ -> ok - end; - {tcp_closed, Socket} -> - comm_port:unregister_connection(Address, Port), - gen_tcp:close(Socket); - {tcp, Socket, Data} -> - case binary_to_term(Data) of - {deliver, Process, Message} -> - Process ! Message, - inet:setopts(Socket, [{active, once}]), - loop(Socket, Address, Port); - {user_close} -> - comm_port:unregister_connection(Address, Port), - gen_tcp:close(Socket); - {youare, _Address, _Port} -> - %% @TODO what do we get from this information? - inet:setopts(Socket, [{active, once}]), - loop(Socket, Address, Port); - Unknown -> - log:log(warn,"[ CC ] unknown message ~p", [Unknown]), - inet:setopts(Socket, [{active, once}]), - loop(Socket, Address, Port) - end; - - {youare, _IP, _Port} -> - loop(Socket, Address, Port); - - Unknown -> - log:log(warn,"[ CC ] unknown message2 ~p", [Unknown]) , - loop(Socket, Address, Port) - end. - -% =============================================================================== - --spec(new_connection(inet:ip_address(), integer(), integer()) -> inet:socket() | fail). -new_connection(Address, Port, MyPort) -> - case gen_tcp:connect(Address, Port, [binary, {packet, 4}, {nodelay, true}, {active, once}, - {send_timeout, config:read(tcp_send_timeout)}], - config:read(tcp_connect_timeout)) of - {ok, Socket} -> - % send end point data - case inet:sockname(Socket) of - {ok, {MyAddress, _MyPort}} -> - Message = term_to_binary({endpoint, MyAddress, MyPort}), - gen_tcp:send(Socket, Message), - case inet:peername(Socket) of - {ok, {RemoteIP, RemotePort}} -> - YouAre = term_to_binary({youare, RemoteIP, RemotePort}), - gen_tcp:send(Socket, YouAre), - Socket; - {error, _Reason} -> - %log:log(error,"[ CC ] reconnect to ~p because socket is ~p", - % [Address, Reason]), - close_connection(Socket), - new_connection(Address, Port, MyPort) - end; - {error, _Reason} -> - %log:log(error,"[ CC ] reconnect to ~p because socket is ~p", - % [Address, Reason]), - close_connection(Socket), - new_connection(Address, Port, MyPort) - end; - {error, _Reason} -> - %log:log(error,"[ CC ] couldn't connect to ~p:~p (~p)", - %[Address, Port, Reason]), - fail - end. - -close_connection(Socket) -> - spawn( fun () -> - gen_tcp:close(Socket) - end ). diff --git a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_layer.erl b/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_layer.erl deleted file mode 100644 index b7fdd183e1..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_layer.erl +++ /dev/null @@ -1,83 +0,0 @@ -% Copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -% -% Licensed 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. -%%%------------------------------------------------------------------- -%%% File : comm_layer.erl -%%% Author : Thorsten Schuett <[email protected]> -%%% Description : Public interface to Communication Layer. -%%% Generic functions to send messages. -%%% Distinguishes on runtime whether the destination is in the -%%% same Erlang virtual machine (use ! for sending) or on a remote -%%% site (use comm_port:send()). -%%% -%%% Created : 04 Feb 2008 by Thorsten Schuett <[email protected]> -%%%------------------------------------------------------------------- -%% @author Thorsten Schuett <[email protected]> -%% @copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -%% @version $Id $ --module(comm_layer_dir.comm_layer). - --author('[email protected]'). --vsn('$Id: comm_layer.erl,v 1.1 2009/11/06 12:41:36 maria Exp $ '). - --export([start_link/0, send/2, this/0, here/1]). - --import(io). --import(util). --import(log). - --include("comm_layer.hrl"). - - -% @TODO: should be ip --type(process_id() :: {any(), integer(), pid()}). -%%==================================================================== -%% public functions -%%==================================================================== - -%% @doc starts the communication port (for supervisor) -%% @spec start_link() -> {ok,Pid} | ignore | {error,Error} -start_link() -> - comm_port_sup:start_link(). - -%% @doc a process descriptor has to specify the erlang vm -%% + the process inside. {IP address, port, pid} -%% @type process_id() = {inet:ip_address(), int(), pid()}. -%% @spec send(process_id(), term()) -> ok - -send({{_IP1, _IP2, _IP3, _IP4} = _IP, _Port, _Pid} = Target, Message) -> - {MyIP,MyPort} = comm_port:get_local_address_port(), - %io:format("send: ~p:~p -> ~p:~p(~p) : ~p\n", [MyIP, MyPort, _IP, _Port, _Pid, Message]), - IsLocal = (MyIP == _IP) and (MyPort == _Port), - if - IsLocal -> - ?LOG_MESSAGE(erlang:element(1, Message), byte_size(term_to_binary(Message))), - _Pid ! Message; - true -> - comm_port:send(Target, Message) - end; - -send(Target, Message) -> - log:log(error,"[ CC ] wrong call to cs_send:send: ~w ! ~w", [Target, Message]), - log:log(error,"[ CC ] stacktrace: ~w", [util:get_stacktrace()]), - ok. - -%% @doc returns process descriptor for the calling process --spec(this/0 :: () -> atom()).%process_id()). -this() -> - here(self()). - --spec(here/1 :: (pid()) -> process_id()). -here(Pid) -> - {LocalIP, LocalPort} = comm_port:get_local_address_port(), - {LocalIP, LocalPort, Pid}. diff --git a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_layer.hrl b/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_layer.hrl deleted file mode 100644 index 54f31b7c55..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_layer.hrl +++ /dev/null @@ -1,29 +0,0 @@ -% Copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -% -% Licensed 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. -%%%------------------------------------------------------------------- -%%% File : comm_layer.hrl -%%% Author : Thorsten Schuett <[email protected]> -%%% Description : -%%% -%%% Created : 31 Jul 2008 by Thorsten Schuett <[email protected]> -%%%------------------------------------------------------------------- -%% @author Thorsten Schuett <[email protected]> -%% @copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -%% @version $Id: comm_layer.hrl,v 1.1 2009/11/06 12:41:36 maria Exp $ --author('[email protected]'). --vsn('$Id: comm_layer.hrl,v 1.1 2009/11/06 12:41:36 maria Exp $ '). - -% enable logging of message statistics -%-define(LOG_MESSAGE(TAG, SIZE), comm_layer.comm_logger:log(TAG, SIZE)). --define(LOG_MESSAGE(TAG, SIZE), ok). diff --git a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_logger.erl b/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_logger.erl deleted file mode 100644 index b8882758af..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_logger.erl +++ /dev/null @@ -1,143 +0,0 @@ -% Copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -% -% Licensed 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. -%%%------------------------------------------------------------------- -%%% File : comm_logger.erl -%%% Author : Thorsten Schuett <[email protected]> -%%% Description : -%%% -%%% Created : 31 Jul 2008 by Thorsten Schuett <[email protected]> -%%%------------------------------------------------------------------- -%% @author Thorsten Schuett <[email protected]> -%% @copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -%% @version $Id: comm_logger.erl,v 1.1 2009/11/06 12:41:36 maria Exp $ --module(comm_layer_dir.comm_logger). - --author('[email protected]'). --vsn('$Id: comm_logger.erl,v 1.1 2009/11/06 12:41:36 maria Exp $ '). - --behaviour(gen_server). - --import(gb_trees). --import(gen_server). - -%% API --export([start_link/0]). - --export([log/2, dump/0]). - -%% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). - --record(state, {start, map}). - -%%==================================================================== -%% API -%%==================================================================== -%%-------------------------------------------------------------------- -%% Function: start_link() -> {ok,Pid} | ignore | {error,Error} -%% Description: Starts the server -%%-------------------------------------------------------------------- -start_link() -> - gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). - -%%-------------------------------------------------------------------- -%% Function: log(Tag, Size) -> ok -%% Description: logs a message type with its size -%%-------------------------------------------------------------------- -log(Tag, Size) -> - gen_server:cast(?MODULE, {log, Tag, Size}). - -%%-------------------------------------------------------------------- -%% Function: dump() -> {gb_tree:gb_trees(), {Date, Time}} -%% Description: gets the logging state -%%-------------------------------------------------------------------- -dump() -> - gen_server:call(?MODULE, {dump}). - -%%==================================================================== -%% gen_server callbacks -%%==================================================================== - -%%-------------------------------------------------------------------- -%% Function: init(Args) -> {ok, State} | -%% {ok, State, Timeout} | -%% ignore | -%% {stop, Reason} -%% Description: Initiates the server -%%-------------------------------------------------------------------- -init([]) -> - {ok, #state{start=erlang:now(), map=gb_trees:empty()}}. - -%%-------------------------------------------------------------------- -%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} | -%% {reply, Reply, State, Timeout} | -%% {noreply, State} | -%% {noreply, State, Timeout} | -%% {stop, Reason, Reply, State} | -%% {stop, Reason, State} -%% Description: Handling call messages -%%-------------------------------------------------------------------- -handle_call({dump}, _From, State) -> - Reply = {State#state.map, State#state.start}, - {reply, Reply, State}; -handle_call(_Request, _From, State) -> - Reply = ok, - {reply, Reply, State}. - -%%-------------------------------------------------------------------- -%% Function: handle_cast(Msg, State) -> {noreply, State} | -%% {noreply, State, Timeout} | -%% {stop, Reason, State} -%% Description: Handling cast messages -%%-------------------------------------------------------------------- -handle_cast({log, Tag, Size}, State) -> - case gb_trees:lookup(Tag, State#state.map) of - none -> - {noreply, State#state{map=gb_trees:insert(Tag, {Size, 1}, State#state.map)}}; - {value, {OldSize, OldCount}} -> - {noreply, State#state{map=gb_trees:update(Tag, {Size + OldSize, OldCount + 1}, State#state.map)}} - end; -handle_cast(_Msg, State) -> - {noreply, State}. - -%%-------------------------------------------------------------------- -%% Function: handle_info(Info, State) -> {noreply, State} | -%% {noreply, State, Timeout} | -%% {stop, Reason, State} -%% Description: Handling all non call/cast messages -%%-------------------------------------------------------------------- -handle_info(_Info, State) -> - {noreply, State}. - -%%-------------------------------------------------------------------- -%% Function: terminate(Reason, State) -> void() -%% Description: 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. -%%-------------------------------------------------------------------- -terminate(_Reason, _State) -> - ok. - -%%-------------------------------------------------------------------- -%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState} -%% Description: Convert process state when code is changed -%%-------------------------------------------------------------------- -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - -%%-------------------------------------------------------------------- -%%% Internal functions -%%-------------------------------------------------------------------- diff --git a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_port.erl b/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_port.erl deleted file mode 100644 index d9fcb5e625..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_port.erl +++ /dev/null @@ -1,241 +0,0 @@ -%% -*- coding: utf-8 -*- -% Copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -% -% Licensed 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. -%%%------------------------------------------------------------------- -%%% File : comm_port.erl -%%% Author : Thorsten Schuett <[email protected]> -%%% Description : Main CommLayer Interface -%%% Maps remote addresses to comm_connection PIDs. -%%% -%%% Created : 18 Apr 2008 by Thorsten Schuett <[email protected]> -%%%------------------------------------------------------------------- -%% @author Thorsten Schuett <[email protected]> -%% @copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -%% @version $Id $ --module(comm_layer_dir.comm_port). - --author('[email protected]'). --vsn('$Id: comm_port.erl,v 1.1 2009/11/06 12:41:36 maria Exp $ '). - --behaviour(gen_server). - --import(ets). --import(gen_server). --import(io). --import(log). - --define(ASYNC, true). -%-define(SYNC, true). - -%% API --export([start_link/0, - send/2, - unregister_connection/2, register_connection/4, - set_local_address/2, get_local_address_port/0]). - -%% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). - -%%==================================================================== -%% API -%%==================================================================== - -%% @doc -%% @spec send({inet:ip_address(), int(), pid()}, term()) -> ok --ifdef(ASYNC). -send({Address, Port, Pid}, Message) -> - gen_server:call(?MODULE, {send, Address, Port, Pid, Message}, 20000). --endif. --ifdef(SYNC). -send({Address, Port, Pid}, Message) -> - case ets:lookup(?MODULE, {Address, Port}) of - [{{Address, Port}, {_LPid, Socket}}] -> - comm_connection:send({Address, Port, Socket}, Pid, Message), - ok; - [] -> - gen_server:call(?MODULE, {send, Address, Port, Pid, Message}, 20000) - end. --endif. - - -%% @doc -%% @spec unregister_connection(inet:ip_address(), int()) -> ok -unregister_connection(Adress, Port) -> - gen_server:call(?MODULE, {unregister_conn, Adress, Port}, 20000). - -%% @doc -%% @spec register_connection(inet:ip_address(), int(), pid(), gen_tcp:socket()) -> ok | duplicate -register_connection(Adress, Port, Pid, Socket) -> - gen_server:call(?MODULE, {register_conn, Adress, Port, Pid, Socket}, 20000). - -%% @doc -%% @spec set_local_address(inet:ip_address(), int()) -> ok -set_local_address(Address, Port) -> - gen_server:call(?MODULE, {set_local_address, Address, Port}, 20000). - - -%% @doc -%% @spec get_local_address_port() -> {inet:ip_address(),int()} -get_local_address_port() -> - case ets:lookup(?MODULE, local_address_port) of - [{local_address_port, Value}] -> - Value; - [] -> - undefined - end. - -%%-------------------------------------------------------------------- -%% Function: start_link() -> {ok,Pid} | ignore | {error,Error} -%% Description: Starts the server -%%-------------------------------------------------------------------- -start_link() -> - gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). - -%%==================================================================== -%% gen_server callbacks -%%==================================================================== - -%%-------------------------------------------------------------------- -%% Function: init(Args) -> {ok, State} | -%% {ok, State, Timeout} | -%% ignore | -%% {stop, Reason} -%% Description: Initiates the server -%%-------------------------------------------------------------------- -init([]) -> - ets:new(?MODULE, [set, protected, named_table]), - {ok, ok}. % empty state. - -%%-------------------------------------------------------------------- -%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} | -%% {reply, Reply, State, Timeout} | -%% {noreply, State} | -%% {noreply, State, Timeout} | -%% {stop, Reason, Reply, State} | -%% {stop, Reason, State} -%% Description: Handling call messages -%%-------------------------------------------------------------------- -handle_call({send, Address, Port, Pid, Message}, _From, State) -> - send(Address, Port, Pid, Message, State); - -handle_call({unregister_conn, Address, Port}, _From, State) -> - ets:delete(?MODULE, {Address, Port}), - {reply, ok, State}; - -handle_call({register_conn, Address, Port, Pid, Socket}, _From, State) -> - case ets:lookup(?MODULE, {Address, Port}) of - [{{Address, Port}, _}] -> - {reply, duplicate, State}; - [] -> - ets:insert(?MODULE, {{Address, Port}, {Pid, Socket}}), - {reply, ok, State} - end; - -handle_call({set_local_address, Address, Port}, _From, State) -> - ets:insert(?MODULE, {local_address_port, {Address,Port}}), - {reply, ok, State}. - -%%-------------------------------------------------------------------- -%% Function: handle_cast(Msg, State) -> {noreply, State} | -%% {noreply, State, Timeout} | -%% {stop, Reason, State} -%% Description: Handling cast messages -%%-------------------------------------------------------------------- -handle_cast(_Msg, State) -> - {noreply, State}. - -%%-------------------------------------------------------------------- -%% Function: handle_info(Info, State) -> {noreply, State} | -%% {noreply, State, Timeout} | -%% {stop, Reason, State} -%% Description: Handling all non call/cast messages -%%-------------------------------------------------------------------- -handle_info(_Info, State) -> - {noreply, State}. - -%%-------------------------------------------------------------------- -%% Function: terminate(Reason, State) -> void() -%% Description: 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. -%%-------------------------------------------------------------------- -terminate(_Reason, _State) -> - ok. - -%%-------------------------------------------------------------------- -%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState} -%% Description: Convert process state when code is changed -%%-------------------------------------------------------------------- -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - -%%-------------------------------------------------------------------- -%%% Internal functions -%%-------------------------------------------------------------------- - --ifdef(ASYNC). -send(Address, Port, Pid, Message, State) -> - {DepAddr,DepPort} = get_local_address_port(), - if - DepAddr == undefined -> - open_sync_connection(Address, Port, Pid, Message, State); - true -> - case ets:lookup(?MODULE, {Address, Port}) of - [{{Address, Port}, {ConnPid, _Socket}}] -> - ConnPid ! {send, Pid, Message}, - {reply, ok, State}; - [] -> - ConnPid = comm_connection:open_new_async(Address, Port, - DepAddr, DepPort), - ets:insert(?MODULE, {{Address, Port}, {ConnPid, undef}}), - ConnPid ! {send, Pid, Message}, - {reply, ok, State} - end - end. --endif. - --ifdef(SYNC). -send(Address, Port, Pid, Message, State) -> - case ets:lookup(?MODULE, {Address, Port}) of - [{{Address, Port}, {_LPid, Socket}}] -> - comm_connection:send({Address, Port, Socket}, Pid, Message), - {reply, ok, State}; - [] -> - open_sync_connection(Address, Port, Pid, Message, State) - end. --endif. - - -open_sync_connection(Address, Port, Pid, Message, State) -> - {DepAddr,DepPort} = get_local_address_port(), - case comm_connection:open_new(Address, Port, DepAddr, DepPort) of - {local_ip, MyIP, MyPort, MyPid, MySocket} -> - comm_connection:send({Address, Port, MySocket}, Pid, Message), - log:log(info,"[ CC ] this() == ~w", [{MyIP, MyPort}]), - % set_local_address(t, {MyIP,MyPort}}), - % register_connection(Address, Port, MyPid, MySocket), - ets:insert(?MODULE, {local_address_port, {MyIP,MyPort}}), - ets:insert(?MODULE, {{Address, Port}, {MyPid, MySocket}}), - {reply, ok, State}; - fail -> - % drop message (remote node not reachable, failure detector will notice) - {reply, ok, State}; - {connection, LocalPid, NewSocket} -> - comm_connection:send({Address, Port, NewSocket}, Pid, Message), - ets:insert(?MODULE, {{Address, Port}, {LocalPid, NewSocket}}), - % register_connection(Address, Port, LPid, NewSocket), - {reply, ok, State} - end. diff --git a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_port_sup.erl b/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_port_sup.erl deleted file mode 100644 index d7a25b14ab..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/src/comm_layer/comm_port_sup.erl +++ /dev/null @@ -1,88 +0,0 @@ -% Copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -% -% Licensed 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. -%%%------------------------------------------------------------------- -%%% File : comm_port_sup.erl -%%% Author : Thorsten Schuett <[email protected]> -%%% Description : -%%% -%%% Created : 04 Feb 2008 by Thorsten Schuett <[email protected]> -%%%------------------------------------------------------------------- -%% @author Thorsten Schuett <[email protected]> -%% @copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -%% @version $Id: comm_port_sup.erl,v 1.1 2009/11/06 12:41:36 maria Exp $ --module(comm_layer_dir.comm_port_sup). - --author('[email protected]'). --vsn('$Id: comm_port_sup.erl,v 1.1 2009/11/06 12:41:36 maria Exp $ '). - --behaviour(supervisor). - --import(supervisor). --import(randoms). --import(string). --import(config). - --export([start_link/0, init/1]). - -%%==================================================================== -%% API functions -%%==================================================================== -%%-------------------------------------------------------------------- -%% Function: start_link() -> {ok,Pid} | ignore | {error,Error} -%% Description: Starts the supervisor -%%-------------------------------------------------------------------- -start_link() -> - supervisor:start_link(?MODULE, []). - -%%==================================================================== -%% Supervisor callbacks -%%==================================================================== -%%-------------------------------------------------------------------- -%% Func: init(Args) -> {ok, {SupFlags, [ChildSpec]}} | -%% ignore | -%% {error, Reason} -%% Description: Whenever a supervisor is started using -%% supervisor:start_link/[2,3], this function is called by the new process -%% to find out about restart strategy, maximum restart frequency and child -%% specifications. -%%-------------------------------------------------------------------- -init([]) -> - InstanceId = string:concat("comm_port_", randoms:getRandomId()), - CommPort = - {comm_port, - {comm_layer_dir.comm_port, start_link, []}, - permanent, - brutal_kill, - worker, - []}, - CommAcceptor = - {comm_acceptor, - {comm_layer_dir.comm_acceptor, start_link, [InstanceId]}, - permanent, - brutal_kill, - worker, - []}, - CommLogger = - {comm_logger, - {comm_layer_dir.comm_logger, start_link, []}, - permanent, - brutal_kill, - worker, - []}, - {ok, {{one_for_all, 10, 1}, - [ - CommPort, - CommLogger, - CommAcceptor - ]}}. diff --git a/lib/dialyzer/test/small_SUITE_data/src/pubsub/pubsub_api.erl b/lib/dialyzer/test/small_SUITE_data/src/pubsub/pubsub_api.erl deleted file mode 100644 index 85ea292077..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/src/pubsub/pubsub_api.erl +++ /dev/null @@ -1,99 +0,0 @@ -% Copyright 2007-2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -% -% Licensed 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. -%%%------------------------------------------------------------------- -%%% File : pubsub_api.erl -%%% Author : Thorsten Schuett <[email protected]> -%%% Description : Publish API function -%%% -%%% Created : 17 Sep 2007 by Thorsten Schuett <[email protected]> -%%%------------------------------------------------------------------- -%% @author Thorsten Schuett <[email protected]> -%% @copyright 2007-2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -%% @version $Id $ --module(pubsub_dir.pubsub_api). - --author('[email protected]'). --vsn('$Id: pubsub_api.erl,v 1.1 2009/11/06 12:39:55 maria Exp $ '). - --export([publish/2, subscribe/2, unsubscribe/2, get_subscribers/1]). - --import(transstore.transaction_api). --import(io). --import(lists). - -%%==================================================================== -%% public functions -%%==================================================================== - -%% @doc publishs an event under a given topic. -%% called e.g. from the java-interface -%% @spec publish(string(), string()) -> ok -publish(Topic, Content) -> - Subscribers = get_subscribers(Topic), - io:format("calling subscribers ~p~n", [Subscribers]), - lists:foreach(fun (Subscriber) -> - io:format("calling ~p~n", [Subscriber]), - pubsub_publish:publish(Subscriber, Topic, Content) - end, - Subscribers), - ok. - -%% @doc subscribes a url for a topic. -%% called e.g. from the java-interface -%% @spec subscribe(string(), string()) -> ok | {fail, term()} -subscribe(Topic, URL) -> - TFun = fun(TransLog) -> - {{Success, _ValueOrReason} = Result, TransLog1} = transaction_api:read(Topic, TransLog), - {Result2, TransLog2} = if - Success == fail -> - transaction_api:write(Topic, [URL], TransLog); %obacht: muss TransLog sein! - true -> - {value, Subscribers} = Result, - transaction_api:write(Topic, [URL | Subscribers], TransLog1) - end, - if - Result2 == ok -> - {{ok, ok}, TransLog2}; - true -> - {Result2, TransLog2} - end - end, - transaction_api:do_transaction(TFun, fun (_) -> ok end, fun (X) -> {fail, X} end). - -%% @doc unsubscribes a url for a topic. --spec(unsubscribe/2 :: (string(), string()) -> ok | {fail, any()}). -unsubscribe(Topic, URL) -> - TFun = fun(TransLog) -> - {Subscribers, TransLog1} = transaction_api:read2(TransLog, Topic), - case lists:member(URL, Subscribers) of - true -> - NewSubscribers = lists:delete(URL, Subscribers), - TransLog2 = transaction_api:write2(TransLog1, Topic, NewSubscribers), - {{ok, ok}, TransLog2}; - false -> - {{fail, not_found}, TransLog} - end - end, - transaction_api:do_transaction(TFun, fun (_) -> ok end, fun (X) -> {fail, X} end). - -%% @doc queries the subscribers of a query -%% @spec get_subscribers(string()) -> [string()] -get_subscribers(Topic) -> - {Fl, _Value} = transaction_api:quorum_read(Topic), - if - Fl == fail -> %% Fl is either Fail or the Value/Subscribers - []; - true -> - Fl - end. diff --git a/lib/dialyzer/test/small_SUITE_data/src/pubsub/pubsub_publish.erl b/lib/dialyzer/test/small_SUITE_data/src/pubsub/pubsub_publish.erl deleted file mode 100644 index 601dbad74b..0000000000 --- a/lib/dialyzer/test/small_SUITE_data/src/pubsub/pubsub_publish.erl +++ /dev/null @@ -1,49 +0,0 @@ -% Copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -% -% Licensed 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. -%%%------------------------------------------------------------------- -%%% File : pubsub_publish.erl -%%% Author : Thorsten Schuett <[email protected]> -%%% Description : Publish function -%%% -%%% Created : 26 Mar 2008 by Thorsten Schuett <[email protected]> -%%%------------------------------------------------------------------- -%% @author Thorsten Schuett <[email protected]> -%% @copyright 2008 Konrad-Zuse-Zentrum für Informationstechnik Berlin -%% @version $Id $ --module(pubsub_dir.pubsub_publish). - --author('[email protected]'). --vsn('$Id: pubsub_publish.erl,v 1.1 2009/11/06 12:39:55 maria Exp $ '). - --export([publish/3, publish_internal/3]). - --import(json). --import(io). --import(http). --import(jsonrpc). - -%%==================================================================== -%% public functions -%%==================================================================== - -%% @doc publishs an event to a given url. -%% @spec publish(string(), string(), string()) -> ok -%% @todo use pool:pspawn -publish(URL, Topic, Content) -> - spawn(fun () -> pubsub_publish:publish_internal(URL, Topic, Content) end), - ok. - -publish_internal(URL, Topic, Content) -> - Res = jsonrpc:call(URL, [], {call, notify, [Topic, Content]}), - io:format("~p ~p~n", [Res, URL]). |