24
1
mirror of https://github.com/processone/ejabberd.git synced 2024-06-08 21:43:07 +02:00

* src/ejabberd_listener.erl: Enable keepalive option

* src/xml_stream.erl: Added API for managing xml streams without
creating process
* src/ejabberd_receiver.erl: Use this API, now 2 processes are
created per C2S connection

SVN Revision: 288
This commit is contained in:
Alexey Shchepin 2004-12-03 22:54:02 +00:00
parent fb977729a9
commit cc19cb1785
4 changed files with 61 additions and 24 deletions

View File

@ -1,3 +1,12 @@
2004-12-03 Alexey Shchepin <alexey@sevcom.net>
* src/ejabberd_listener.erl: Enable keepalive option
* src/xml_stream.erl: Added API for managing xml streams without
creating process
* src/ejabberd_receiver.erl: Use this API, now 2 processes are
created per C2S connection
2004-12-01 Alexey Shchepin <alexey@sevcom.net> 2004-12-01 Alexey Shchepin <alexey@sevcom.net>
* src/expat_erl.c: Now uses port control instead of port output * src/expat_erl.c: Now uses port control instead of port output

View File

@ -70,7 +70,8 @@ init(Port, Module, Opts) ->
{packet, 0}, {packet, 0},
{active, false}, {active, false},
{reuseaddr, true}, {reuseaddr, true},
{nodelay, true} | {nodelay, true},
{keepalive, true} |
SockOpts]), SockOpts]),
case Res of case Res of
{ok, ListenSocket} -> {ok, ListenSocket} ->

View File

@ -24,7 +24,7 @@ start(Socket, SockMod, Shaper) ->
receiver(Socket, SockMod, Shaper, C2SPid) -> receiver(Socket, SockMod, Shaper, C2SPid) ->
XMLStreamPid = xml_stream:start(self(), C2SPid), XMLStreamState = xml_stream:new(C2SPid),
ShaperState = shaper:new(Shaper), ShaperState = shaper:new(Shaper),
Timeout = case SockMod of Timeout = case SockMod of
ssl -> ssl ->
@ -32,32 +32,32 @@ receiver(Socket, SockMod, Shaper, C2SPid) ->
_ -> _ ->
infinity infinity
end, end,
receiver(Socket, SockMod, ShaperState, C2SPid, XMLStreamPid, Timeout). receiver(Socket, SockMod, ShaperState, C2SPid, XMLStreamState, Timeout).
receiver(Socket, SockMod, ShaperState, C2SPid, XMLStreamPid, Timeout) -> receiver(Socket, SockMod, ShaperState, C2SPid, XMLStreamState, Timeout) ->
Res = (catch SockMod:recv(Socket, 0, Timeout)), Res = (catch SockMod:recv(Socket, 0, Timeout)),
case Res of case Res of
{ok, Data} -> {ok, Data} ->
receive receive
{starttls, TLSSocket} -> {starttls, TLSSocket} ->
exit(XMLStreamPid, closed), xml_stream:close(XMLStreamState),
XMLStreamPid1 = xml_stream:start(self(), C2SPid), XMLStreamState1 = xml_stream:new(C2SPid),
TLSRes = tls:recv_data(TLSSocket, Data), TLSRes = tls:recv_data(TLSSocket, Data),
receiver1(TLSSocket, tls, receiver1(TLSSocket, tls,
ShaperState, C2SPid, XMLStreamPid1, Timeout, ShaperState, C2SPid, XMLStreamState1, Timeout,
TLSRes) TLSRes)
after 0 -> after 0 ->
receiver1(Socket, SockMod, receiver1(Socket, SockMod,
ShaperState, C2SPid, XMLStreamPid, Timeout, ShaperState, C2SPid, XMLStreamState, Timeout,
Res) Res)
end; end;
_ -> _ ->
receiver1(Socket, SockMod, receiver1(Socket, SockMod,
ShaperState, C2SPid, XMLStreamPid, Timeout, Res) ShaperState, C2SPid, XMLStreamState, Timeout, Res)
end. end.
receiver1(Socket, SockMod, ShaperState, C2SPid, XMLStreamPid, Timeout, Res) -> receiver1(Socket, SockMod, ShaperState, C2SPid, XMLStreamState, Timeout, Res) ->
case Res of case Res of
{ok, Text} -> {ok, Text} ->
ShaperSt1 = receive ShaperSt1 = receive
@ -67,27 +67,27 @@ receiver1(Socket, SockMod, ShaperState, C2SPid, XMLStreamPid, Timeout, Res) ->
ShaperState ShaperState
end, end,
NewShaperState = shaper:update(ShaperSt1, size(Text)), NewShaperState = shaper:update(ShaperSt1, size(Text)),
XMLStreamPid1 = receive XMLStreamState1 = receive
reset_stream -> reset_stream ->
exit(XMLStreamPid, closed), xml_stream:close(XMLStreamState),
xml_stream:start(self(), C2SPid) xml_stream:new(C2SPid)
after 0 -> after 0 ->
XMLStreamPid XMLStreamState
end, end,
xml_stream:send_text(XMLStreamPid1, Text), XMLStreamState2 = xml_stream:parse(XMLStreamState1, Text),
receiver(Socket, SockMod, NewShaperState, C2SPid, XMLStreamPid1, receiver(Socket, SockMod, NewShaperState, C2SPid, XMLStreamState2,
Timeout); Timeout);
{error, timeout} -> {error, timeout} ->
receiver(Socket, SockMod, ShaperState, C2SPid, XMLStreamPid, receiver(Socket, SockMod, ShaperState, C2SPid, XMLStreamState,
Timeout); Timeout);
{error, Reason} -> {error, Reason} ->
exit(XMLStreamPid, closed), xml_stream:close(XMLStreamState),
gen_fsm:send_event(C2SPid, closed), gen_fsm:send_event(C2SPid, closed),
ok; ok;
{'EXIT', Reason} -> {'EXIT', Reason} ->
?ERROR_MSG("(~w) abnormal ~w:recv termination:~n\t~p~n", ?ERROR_MSG("(~w) abnormal ~w:recv termination:~n\t~p~n",
[Socket, SockMod, Reason]), [Socket, SockMod, Reason]),
exit(XMLStreamPid, closed), xml_stream:close(XMLStreamState),
gen_fsm:send_event(C2SPid, closed), gen_fsm:send_event(C2SPid, closed),
ok ok
end. end.

View File

@ -1,7 +1,7 @@
%%%---------------------------------------------------------------------- %%%----------------------------------------------------------------------
%%% File : xml_stream.erl %%% File : xml_stream.erl
%%% Author : Alexey Shchepin <alexey@sevcom.net> %%% Author : Alexey Shchepin <alexey@sevcom.net>
%%% Purpose : %%% Purpose : Parse XML streams
%%% Created : 17 Nov 2002 by Alexey Shchepin <alexey@sevcom.net> %%% Created : 17 Nov 2002 by Alexey Shchepin <alexey@sevcom.net>
%%% Id : $Id$ %%% Id : $Id$
%%%---------------------------------------------------------------------- %%%----------------------------------------------------------------------
@ -10,7 +10,12 @@
-author('alexey@sevcom.net'). -author('alexey@sevcom.net').
-vsn('$Revision$ '). -vsn('$Revision$ ').
-export([start/1, start/2, init/1, init/2, send_text/2]). -export([start/1, start/2,
init/1, init/2,
send_text/2,
new/1,
parse/2,
close/1]).
-define(XML_START, 0). -define(XML_START, 0).
-define(XML_END, 1). -define(XML_END, 1).
@ -19,6 +24,8 @@
-define(PARSE_COMMAND, 0). -define(PARSE_COMMAND, 0).
-record(xml_stream_state, {callback_pid, port, stack}).
start(CallbackPid) -> start(CallbackPid) ->
spawn(?MODULE, init, [CallbackPid]). spawn(?MODULE, init, [CallbackPid]).
@ -96,3 +103,23 @@ process_data(CallbackPid, Stack, Data) ->
send_text(Pid, Text) -> send_text(Pid, Text) ->
Pid ! {self(), {send, Text}}. Pid ! {self(), {send, Text}}.
new(CallbackPid) ->
Port = open_port({spawn, expat_erl}, [binary]),
#xml_stream_state{callback_pid = CallbackPid,
port = Port,
stack = []}.
parse(#xml_stream_state{callback_pid = CallbackPid,
port = Port,
stack = Stack} = State, Str) ->
Res = port_control(Port, ?PARSE_COMMAND, Str),
NewStack = lists:foldl(
fun(Data, St) ->
process_data(CallbackPid, St, Data)
end, Stack, binary_to_term(Res)),
State#xml_stream_state{stack = NewStack}.
close(#xml_stream_state{port = Port}) ->
port_close(Port).