TCP Server Application (tcp_server_app.erl)

-module(tcp_server_app).

-author(‘saleyn@gmail.com’).

-behaviour(application).

%% Internal API

-export([start_client/0]).

%% Application and Supervisor callbacks

-export([start/2, stop/1, init/1]).

-define(MAX_RESTART, 5).

-define(MAX_TIME, 60).

-define(DEF_PORT, 2222).

%% A startup function for spawning new client connection handling FSM.

%% To be called by the TCP listener process.

start_client() ->

supervisor:start_child(tcp_client_sup, []).

%%———————————————————————-

%% Application behaviour callbacks

%%———————————————————————-

start(_Type, _Args) ->

ListenPort = get_app_env(listen_port, ?DEF_PORT),

supervisor:start_link({local, ?MODULE}, ?MODULE, [ListenPort, tcp_echo_fsm]).

stop(_S) ->

ok.

%%———————————————————————-

%% Supervisor behaviour callbacks

%%———————————————————————-

init([Port, Module]) ->

{ok,

{_SupFlags = {one_for_one, ?MAX_RESTART, ?MAX_TIME},

[

% TCP Listener

{ tcp_server_sup, % Id = internal id

{tcp_listener,start_link,[Port,Module]}, % StartFun = {M, F, A}

permanent, % Restart = permanent | transient | temporary

2000, % Shutdown = brutal_kill | int() >= 0 | infinity

worker, % Type = worker | supervisor

[tcp_listener] % Modules = [Module] | dynamic

},

% Client instance supervisor

{ tcp_client_sup,

{supervisor,start_link,[{local, tcp_client_sup}, ?MODULE, [Module]]},

permanent, % Restart = permanent | transient | temporary

infinity, % Shutdown = brutal_kill | int() >= 0 | infinity

supervisor, % Type = worker | supervisor

[] % Modules = [Module] | dynamic

}

]

}

};

init([Module]) ->

{ok,

{_SupFlags = {simple_one_for_one, ?MAX_RESTART, ?MAX_TIME},

[

% TCP Client

{ undefined, % Id = internal id

{Module,start_link,[]}, % StartFun = {M, F, A}

temporary, % Restart = permanent | transient | temporary

2000, % Shutdown = brutal_kill | int() >= 0 | infinity

worker, % Type = worker | supervisor

[] % Modules = [Module] | dynamic

}

]

}

}.

%%———————————————————————-

%% Internal functions

%%———————————————————————-

get_app_env(Opt, Default) ->

case application:get_env(application:get_application(), Opt) of

{ok, Val} -> Val;

_ ->

case init:get_argument(Opt) of

[[Val | _]] -> Val;

error -> Default

end

end.

TCP Listener Process (tcp_listener.erl)

-module(tcp_listener).

-author(‘saleyn@gmail.com’).

-behaviour(gen_server).

%% External API

-export([start_link/2]).

%% gen_server callbacks

-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,

code_change/3]).

-record(state, {

listener, % Listening socket

acceptor, % Asynchronous acceptor’s internal reference

module % FSM handling module

}).

%%——————————————————————–

%% @spec (Port::integer(), Module) -> {ok, Pid} | {error, Reason}

%

%% @doc Called by a supervisor to start the listening process.

%% @end

%%———————————————————————-

start_link(Port, Module) when is_integer(Port), is_atom(Module) ->

gen_server:start_link({local, ?MODULE}, ?MODULE, [Port, Module], []).

%%%————————————————————————

%%% Callback functions from gen_server

%%%————————————————————————

%%———————————————————————-

%% @spec (Port::integer()) -> {ok, State} |

%% {ok, State, Timeout} |

%% ignore |

%% {stop, Reason}

%%

%% @doc Called by gen_server framework at process startup.

%% Create listening socket.

%% @end

%%———————————————————————-

init([Port, Module]) ->

process_flag(trap_exit, true),

Opts = [binary, {packet, 2}, {reuseaddr, true},

{keepalive, true}, {backlog, 30}, {active, false}],

case gen_tcp:listen(Port, Opts) of

{ok, Listen_socket} ->

%%Create first accepting process

{ok, Ref} = prim_inet:async_accept(Listen_socket, -1),

{ok, #state{listener = Listen_socket,

acceptor = Ref,

module = Module}};

{error, Reason} ->

{stop, Reason}

end.

%%————————————————————————-

%% @spec (Request, From, State) -> {reply, Reply, State} |

%% {reply, Reply, State, Timeout} |

%% {noreply, State} |

%% {noreply, State, Timeout} |

%% {stop, Reason, Reply, State} |

%% {stop, Reason, State}

%% @doc Callback for synchronous server calls. If `{stop, …}’ tuple

%% is returned, the server is stopped and `terminate/2′ is called.

%% @end

%% @private

%%————————————————————————-

handle_call(Request, _From, State) ->

{stop, {unknown_call, Request}, State}.

%%————————————————————————-

%% @spec (Msg, State) ->{noreply, State} |

%% {noreply, State, Timeout} |

%% {stop, Reason, State}

%% @doc Callback for asyncrous server calls. If `{stop, …}’ tuple

%% is returned, the server is stopped and `terminate/2′ is called.

%% @end

%% @private

%%————————————————————————-

handle_cast(_Msg, State) ->

{noreply, State}.

%%————————————————————————-

%% @spec (Msg, State) ->{noreply, State} |

%% {noreply, State, Timeout} |

%% {stop, Reason, State}

%% @doc Callback for messages sent directly to server’s mailbox.

%% If `{stop, …}’ tuple is returned, the server is stopped and

%% `terminate/2′ is called.

%% @end

%% @private

%%————————————————————————-

handle_info({inet_async, ListSock, Ref, {ok, CliSocket}},

#state{listener=ListSock, acceptor=Ref, module=Module} = State) ->

try

case set_sockopt(ListSock, CliSocket) of

ok -> ok;

{error, Reason} -> exit({set_sockopt, Reason})

end,

%% New client connected – spawn a new process using the simple_one_for_one

%% supervisor.

{ok, Pid} = tcp_server_app:start_client(),

gen_tcp:controlling_process(CliSocket, Pid),

%% Instruct the new FSM that it owns the socket.

Module:set_socket(Pid, CliSocket),

%% Signal the network driver that we are ready to accept another connection

case prim_inet:async_accept(ListSock, -1) of

{ok, NewRef} -> ok;

{error, NewRef} -> exit({async_accept, inet:format_error(NewRef)})

end,

{noreply, State#state{acceptor=NewRef}}

catch exit:Why ->

error_logger:error_msg(“Error in async accept: ~p.\n”, [Why]),

{stop, Why, State}

end;

handle_info({inet_async, ListSock, Ref, Error}, #state{listener=ListSock, acceptor=Ref} = State) ->

error_logger:error_msg(“Error in socket acceptor: ~p.\n”, [Error]),

{stop, Error, State};

handle_info(_Info, State) ->

{noreply, State}.

%%————————————————————————-

%% @spec (Reason, State) -> any

%% @doc Callback executed on server shutdown. It is only invoked if

%% `process_flag(trap_exit, true)’ is set by the server process.

%% The return value is ignored.

%% @end

%% @private

%%————————————————————————-

terminate(_Reason, State) ->

gen_tcp:close(State#state.listener),

ok.

%%————————————————————————-

%% @spec (OldVsn, State, Extra) -> {ok, NewState}

%% @doc Convert process state when code is changed.

%% @end

%% @private

%%————————————————————————-

code_change(_OldVsn, State, _Extra) ->

{ok, State}.

%%%————————————————————————

%%% Internal functions

%%%————————————————————————

%% Taken from prim_inet. We are merely copying some socket options from the

%% listening socket to the new client socket.

set_sockopt(ListSock, CliSocket) ->

true = inet_db:register_socket(CliSocket, inet_tcp),

case prim_inet:getopts(ListSock, [active, nodelay, keepalive, delay_send, priority, tos]) of

{ok, Opts} ->

case prim_inet:setopts(CliSocket, Opts) of

ok -> ok;

Error -> gen_tcp:close(CliSocket), Error

end;

Error ->

gen_tcp:close(CliSocket), Error

end.

TCP Client Socket Handling FSM (tcp_echo_fsm.erl)

-module(tcp_echo_fsm).

-author(‘saleyn@gmail.com’).

-behaviour(gen_fsm).

-export([start_link/0, set_socket/2]).

%% gen_fsm callbacks

-export([init/1, handle_event/3,

handle_sync_event/4, handle_info/3, terminate/3, code_change/4]).

%% FSM States

-export([

‘WAIT_FOR_SOCKET‘/2,

‘WAIT_FOR_DATA‘/2

]).

-record(state, {

socket, % client socket

addr % client address

}).

-define(TIMEOUT, 120000).

%%%————————————————————————

%%% API

%%%————————————————————————

%%————————————————————————-

%% @spec (Socket) -> {ok,Pid} | ignore | {error,Error}

%% @doc To be called by the supervisor in order to start the server.

%% If init/1 fails with Reason, the function returns {error,Reason}.

%% If init/1 returns {stop,Reason} or ignore, the process is

%% terminated and the function returns {error,Reason} or ignore,

%% respectively.

%% @end

%%————————————————————————-

start_link() ->

gen_fsm:start_link(?MODULE, [], []).

set_socket(Pid, Socket) when is_pid(Pid), is_port(Socket) ->

gen_fsm:send_event(Pid, {socket_ready, Socket}).

%%%————————————————————————

%%% Callback functions from gen_server

%%%————————————————————————

%%————————————————————————-

%% Func: init/1

%% Returns: {ok, StateName, StateData} |

%% {ok, StateName, StateData, Timeout} |

%% ignore |

%% {stop, StopReason}

%% @private

%%————————————————————————-

init([]) ->

process_flag(trap_exit, true),

{ok, ‘WAIT_FOR_SOCKET’, #state{}}.

%%————————————————————————-

%% Func: StateName/2

%% Returns: {next_state, NextStateName, NextStateData} |

%% {next_state, NextStateName, NextStateData, Timeout} |

%% {stop, Reason, NewStateData}

%% @private

%%————————————————————————-

‘WAIT_FOR_SOCKET’({socket_ready, Socket}, State) when is_port(Socket) ->

% Now we own the socket

inet:setopts(Socket, [{active, once}, {packet, 2}, binary]),

{ok, {IP, _Port}} = inet:peername(Socket),

{next_state, ‘WAIT_FOR_DATA’, State#state{socket=Socket, addr=IP}, ?TIMEOUT};

‘WAIT_FOR_SOCKET’(Other, State) ->

error_logger:error_msg(“State: ‘WAIT_FOR_SOCKET’. Unexpected message: ~p\n”, [Other]),

%% Allow to receive async messages

{next_state, ‘WAIT_FOR_SOCKET’, State}.

%% Notification event coming from client

‘WAIT_FOR_DATA’({data, Data}, #state{socket=S} = State) ->

ok = gen_tcp:send(S, Data),

{next_state, ‘WAIT_FOR_DATA’, State, ?TIMEOUT};

‘WAIT_FOR_DATA’(timeout, State) ->

error_logger:error_msg(“~p Client connection timeout – closing.\n”, [self()]),

{stop, normal, State};

‘WAIT_FOR_DATA’(Data, State) ->

io:format(“~p Ignoring data: ~p\n”, [self(), Data]),

{next_state, ‘WAIT_FOR_DATA’, State, ?TIMEOUT}.

%%————————————————————————-

%% Func: handle_event/3

%% Returns: {next_state, NextStateName, NextStateData} |

%% {next_state, NextStateName, NextStateData, Timeout} |

%% {stop, Reason, NewStateData}

%% @private

%%————————————————————————-

handle_event(Event, StateName, StateData) ->

{stop, {StateName, undefined_event, Event}, StateData}.

%%————————————————————————-

%% Func: handle_sync_event/4

%% Returns: {next_state, NextStateName, NextStateData} |

%% {next_state, NextStateName, NextStateData, Timeout} |

%% {reply, Reply, NextStateName, NextStateData} |

%% {reply, Reply, NextStateName, NextStateData, Timeout} |

%% {stop, Reason, NewStateData} |

%% {stop, Reason, Reply, NewStateData}

%% @private

%%————————————————————————-

handle_sync_event(Event, _From, StateName, StateData) ->

{stop, {StateName, undefined_event, Event}, StateData}.

%%————————————————————————-

%% Func: handle_info/3

%% Returns: {next_state, NextStateName, NextStateData} |

%% {next_state, NextStateName, NextStateData, Timeout} |

%% {stop, Reason, NewStateData}

%% @private

%%————————————————————————-

handle_info({tcp, Socket, Bin}, StateName, #state{socket=Socket} = StateData) ->

% Flow control: enable forwarding of next TCP message

inet:setopts(Socket, [{active, once}]),

?MODULE:StateName({data, Bin}, StateData);

handle_info({tcp_closed, Socket}, _StateName,

#state{socket=Socket, addr=Addr} = StateData) ->

error_logger:info_msg(“~p Client ~p disconnected.\n”, [self(), Addr]),

{stop, normal, StateData};

handle_info(_Info, StateName, StateData) ->

{noreply, StateName, StateData}.

%%————————————————————————-

%% Func: terminate/3

%% Purpose: Shutdown the fsm

%% Returns: any

%% @private

%%————————————————————————-

terminate(_Reason, _StateName, #state{socket=Socket}) ->

(catch gen_tcp:close(Socket)),

ok.

%%————————————————————————-

%% Func: code_change/4

%% Purpose: Convert process state when code is changed

%% Returns: {ok, NewState, NewStateData}

%% @private

%%————————————————————————-

code_change(_OldVsn, StateName, StateData, _Extra) ->

{ok, StateName, StateData}.

Application File (tcp_server.app)

{application, tcp_server,

[

{description, "Demo TCP server"},

{vsn, "1.0"},

{id, "tcp_server"},

{modules, [tcp_listener, tcp_echo_fsm]},

{registered, [tcp_server_sup, tcp_listener]},

{applications, [kernel, stdlib]},

%%

%% mod: Specify the module name to start the application, plus args

%%

{mod, {tcp_server_app, []}},

{env, []}

]

}.