Chapril/xmpp.chapril.org-ejabberd
25
1
Fork 0
mirror of https://github.com/processone/ejabberd.git synced 2024-12-24 17:29:28 +01:00
Code Releases Activity

merge from ekhramtsov-ejabberd, which is latest 2.1.x with consistent hash

Browse Source
This commit is contained in:
Christophe Romain 2010-07-13 12:04:30 +02:00
parent b7e02cc42e
commit 628571f8cf
47 changed files with 2028 additions and 683 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
contrib/ejabberd-modules.repo
View File

@ -1,5 +0,0 @@
% List of ejabberd-modules to add for ejabberd packaging (source archive and installer)
%
% HTTP-binding:
%https://svn.process-one.net/ejabberd-modules/http_bind/trunk
%https://svn.process-one.net/ejabberd-modules/mod_http_fileserver/trunk

14
src/Makefile.in
View File

@ -30,16 +30,15 @@ else
INIT_USER=$(INSTALLUSER)
endif
EFLAGS += @ERLANG_SSL39@ -pa .
EFLAGS += @ERLANG_SSLVER@ -pa .
# make debug=true to compile Erlang module with debug informations.
ifdef debug
EFLAGS+=+debug_info +export_all
endif
DEBUGTOOLS = ejabberd_debug.erl
ifdef ejabberd_debug
EFLAGS+=-Dejabberd_debug
DEBUGTOOLS = p1_prof.erl
ifdef debugtools
SOURCES+=$(DEBUGTOOLS)
endif
@ -55,6 +54,11 @@ ifeq (@full_xml@, true)
EFLAGS+=-DFULL_XML_SUPPORT
endif
ifeq (@nif@, true)
EFLAGS+=-DNIF
ERLSHLIBS=xml.so
endif
ifeq (@transient_supervisors@, false)
EFLAGS+=-DNO_TRANSIENT_SUPERVISORS
endif
@ -68,7 +72,7 @@ prefix = @prefix@
exec_prefix = @exec_prefix@
SUBDIRS = @mod_irc@ @mod_pubsub@ @mod_muc@ @mod_proxy65@ @eldap@ @pam@ @web@ stringprep stun @tls@ @odbc@ @ejabberd_zlib@
ERLSHLIBS = expat_erl.so
ERLSHLIBS += expat_erl.so
ERLBEHAVS = cyrsasl.erl gen_mod.erl p1_fsm.erl
SOURCES_ALL = $(wildcard *.erl)
SOURCES_MISC = $(ERLBEHAVS) $(DEBUGTOOLS)

17
src/aclocal.m4 vendored
View File

@ -121,7 +121,6 @@ AC_DEFUN(AM_WITH_ERLANG,
-author('alexey@sevcom.net').
-export([[start/0]]).
-include_lib("ssl/include/ssl_pkix.hrl").
start() ->
EIDirS = code:lib_dir("erl_interface") ++ "\n",
@ -130,11 +129,13 @@ start() ->
file:write_file("conftest.out", list_to_binary(EIDirS ++ EILibS ++ ssldef() ++ RootDirS)),
halt().
-[ifdef]('id-pkix').
ssldef() -> "-DSSL39\n".
-else.
ssldef() -> "\n".
-endif.
ssldef() ->
OTP = (catch erlang:system_info(otp_release)),
if
OTP >= "R14" -> "-DSSL40\n";
OTP >= "R12" -> "-DSSL39\n";
true -> ""
end.
%% return physical architecture based on OS/Processor
archname() ->
@ -184,7 +185,7 @@ _EOF
# Second line
ERLANG_EI_LIB=`cat conftest.out | head -n 2 | tail -n 1`
# Third line
ERLANG_SSL39=`cat conftest.out | head -n 3 | tail -n 1`
ERLANG_SSLVER=`cat conftest.out | head -n 3 | tail -n 1`
# End line
ERLANG_DIR=`cat conftest.out | tail -n 1`
@ -193,7 +194,7 @@ _EOF
AC_SUBST(ERLANG_CFLAGS)
AC_SUBST(ERLANG_LIBS)
AC_SUBST(ERLANG_SSL39)
AC_SUBST(ERLANG_SSLVER)
AC_SUBST(ERLC)
AC_SUBST(ERL)
])

36
src/configure vendored
View File

@ -611,6 +611,7 @@ build
INSTALLUSER
SSL_CFLAGS
SSL_LIBS
nif
full_xml
transient_supervisors
db_type
@ -647,7 +648,7 @@ EGREP
GREP
CPP
LIBICONV
ERLANG_SSL39
ERLANG_SSLVER
ERLANG_LIBS
ERLANG_CFLAGS
ERL
@ -721,6 +722,7 @@ enable_roster_gateway_workaround
enable_mssql
enable_transient_supervisors
enable_full_xml
enable_nif
with_openssl
enable_user
'
@ -1372,6 +1374,8 @@ Optional Features:
(default: yes)
--enable-full-xml use XML features in XMPP stream (ex: CDATA)
(default: no, requires XML compliant clients)
--enable-nif replace some functions with C equivalents. Requires
Erlang R13B04 or higher (default: no)
--enable-user[[[=USER]]]
allow this system user to start ejabberd (default:
no)
@ -3207,7 +3211,6 @@ fi
-author('alexey@sevcom.net').
-export([start/0]).
-include_lib("ssl/include/ssl_pkix.hrl").
start() ->
EIDirS = code:lib_dir("erl_interface") ++ "\n",
@ -3216,11 +3219,13 @@ start() ->
file:write_file("conftest.out", list_to_binary(EIDirS ++ EILibS ++ ssldef() ++ RootDirS)),
halt().
-ifdef('id-pkix').
ssldef() -> "-DSSL39\n".
-else.
ssldef() -> "\n".
-endif.
ssldef() ->
OTP = (catch erlang:system_info(otp_release)),
if
OTP >= "R14" -> "-DSSL40\n";
OTP >= "R12" -> "-DSSL39\n";
true -> ""
end.
%% return physical architecture based on OS/Processor
archname() ->
@ -3270,7 +3275,7 @@ _EOF
# Second line
ERLANG_EI_LIB=`cat conftest.out | head -n 2 | tail -n 1`
# Third line
ERLANG_SSL39=`cat conftest.out | head -n 3 | tail -n 1`
ERLANG_SSLVER=`cat conftest.out | head -n 3 | tail -n 1`
# End line
ERLANG_DIR=`cat conftest.out | tail -n 1`
@ -4650,7 +4655,7 @@ if test "${enable_transient_supervisors+set}" = set; then :
enableval=$enable_transient_supervisors; case "${enableval}" in
yes) transient_supervisors=true ;;
no) transient_supervisors=false ;;
*) as_fn_error "bad value ${enableval} for --enable-full-xml" "$LINENO" 5 ;;
*) as_fn_error "bad value ${enableval} for --enable-transient_supervisors" "$LINENO" 5 ;;
esac
else
transient_supervisors=true
@ -4671,6 +4676,19 @@ fi
# Check whether --enable-nif was given.
if test "${enable_nif+set}" = set; then :
enableval=$enable_nif; case "${enableval}" in
yes) nif=true ;;
no) nif=false ;;
*) as_fn_error "bad value ${enableval} for --enable-nif" "$LINENO" 5 ;;
esac
else
nif=false
fi
ac_config_files="$ac_config_files Makefile $make_mod_irc $make_mod_muc $make_mod_pubsub $make_mod_proxy65 $make_eldap $make_pam $make_web stringprep/Makefile stun/Makefile $make_tls $make_odbc $make_ejabberd_zlib"
#openssl

11
src/configure.ac
View File

@ -81,7 +81,7 @@ AC_ARG_ENABLE(transient_supervisors,
[case "${enableval}" in
yes) transient_supervisors=true ;;
no) transient_supervisors=false ;;
*) AC_MSG_ERROR(bad value ${enableval} for --enable-full-xml) ;;
*) AC_MSG_ERROR(bad value ${enableval} for --enable-transient_supervisors) ;;
esac],[transient_supervisors=true])
AC_SUBST(transient_supervisors)
@ -94,6 +94,15 @@ AC_ARG_ENABLE(full_xml,
esac],[full_xml=false])
AC_SUBST(full_xml)
AC_ARG_ENABLE(nif,
[AC_HELP_STRING([--enable-nif], [replace some functions with C equivalents. Requires Erlang R13B04 or higher (default: no)])],
[case "${enableval}" in
yes) nif=true ;;
no) nif=false ;;
*) AC_MSG_ERROR(bad value ${enableval} for --enable-nif) ;;
esac],[nif=false])
AC_SUBST(nif)
AC_CONFIG_FILES([Makefile
$make_mod_irc
$make_mod_muc

1
src/ejabberd.app
View File

@ -115,6 +115,7 @@
nodetree_virtual,
p1_fsm,
p1_mnesia,
p1_prof,
randoms,
sha,
shaper,

3
src/ejabberd_app.erl
View File

@ -46,6 +46,7 @@ start(normal, _Args) ->
db_init(),
sha:start(),
stringprep_sup:start_link(),
xml:start(),
start(),
translate:start(),
acl:start(),
@ -65,6 +66,8 @@ start(normal, _Args) ->
%ejabberd_debug:fprof_start(),
maybe_add_nameservers(),
start_modules(),
ejabberd_cluster:announce(),
ejabberd_node_groups:start(),
ejabberd_listener:start_listeners(),
?INFO_MSG("ejabberd ~s is started in the node ~p", [?VERSION, node()]),
Sup;

16
src/ejabberd_auth_anonymous.erl
View File

@ -61,9 +61,11 @@
%% Register to login / logout events
start(Host) ->
%% TODO: Check cluster mode
update_tables(),
mnesia:create_table(anonymous, [{ram_copies, [node()]},
{type, bag},
{type, bag}, {local_content, true},
{attributes, record_info(fields, anonymous)}]),
mnesia:add_table_copy(anonymous, node(), ram_copies),
%% The hooks are needed to add / remove users from the anonymous tables
ejabberd_hooks:add(sm_register_connection_hook, Host,
?MODULE, register_connection, 100),
@ -136,7 +138,7 @@ remove_connection(SID, LUser, LServer) ->
F = fun() ->
mnesia:delete_object({anonymous, US, SID})
end,
mnesia:transaction(F).
mnesia:async_dirty(F).
%% Register connection
register_connection(SID, #jid{luser = LUser, lserver = LServer}, Info) ->
@ -144,7 +146,7 @@ register_connection(SID, #jid{luser = LUser, lserver = LServer}, Info) ->
case AuthModule == ?MODULE of
true ->
US = {LUser, LServer},
mnesia:sync_dirty(
mnesia:async_dirty(
fun() -> mnesia:write(#anonymous{us = US, sid=SID})
end);
false ->
@ -246,3 +248,11 @@ remove_user(_User, _Server, _Password) ->
plain_password_required() ->
false.
update_tables() ->
case catch mnesia:table_info(anonymous, local_content) of
false ->
mnesia:delete_table(anonymous);
_ ->
ok
end.

178
src/ejabberd_c2s.erl
View File

@ -35,7 +35,7 @@
%% External exports
-export([start/2,
stop/1,
start_link/2,
start_link/3,
send_text/2,
send_element/2,
socket_type/0,
@ -56,8 +56,9 @@
code_change/4,
handle_info/3,
terminate/3,
print_state/1
]).
print_state/1,
migrate/3
]).
-include("ejabberd.hrl").
-include("jlib.hrl").
@ -96,6 +97,7 @@
conn = unknown,
auth_module = unknown,
ip,
fsm_limit_opts,
lang}).
%-define(DBGFSM, true).
@ -108,11 +110,12 @@
%% Module start with or without supervisor:
-ifdef(NO_TRANSIENT_SUPERVISORS).
-define(SUPERVISOR_START, ?GEN_FSM:start(ejabberd_c2s, [SockData, Opts],
fsm_limit_opts(Opts) ++ ?FSMOPTS)).
-define(SUPERVISOR_START, ?GEN_FSM:start(ejabberd_c2s,
[SockData, Opts, FSMLimitOpts],
FSMLimitOpts ++ ?FSMOPTS)).
-else.
-define(SUPERVISOR_START, supervisor:start_child(ejabberd_c2s_sup,
[SockData, Opts])).
[SockData, Opts, FSMLimitOpts])).
-endif.
%% This is the timeout to apply between event when starting a new
@ -140,12 +143,17 @@
%%%----------------------------------------------------------------------
%%% API
%%%----------------------------------------------------------------------
start(StateName, #state{fsm_limit_opts = Opts} = State) ->
start(StateName, State, Opts);
start(SockData, Opts) ->
start(SockData, Opts, fsm_limit_opts(Opts)).
start(SockData, Opts, FSMLimitOpts) ->
?SUPERVISOR_START.
start_link(SockData, Opts) ->
?GEN_FSM:start_link(ejabberd_c2s, [SockData, Opts],
fsm_limit_opts(Opts) ++ ?FSMOPTS).
start_link(SockData, Opts, FSMLimitOpts) ->
?GEN_FSM:start_link(ejabberd_c2s, [SockData, Opts, FSMLimitOpts],
FSMLimitOpts ++ ?FSMOPTS).
socket_type() ->
xml_stream.
@ -157,6 +165,9 @@ get_presence(FsmRef) ->
stop(FsmRef) ->
?GEN_FSM:send_event(FsmRef, closed).
migrate(FsmRef, Node, After) ->
?GEN_FSM:send_all_state_event(FsmRef, {migrate, Node, After}).
%%%----------------------------------------------------------------------
%%% Callback functions from gen_fsm
%%%----------------------------------------------------------------------
@ -168,7 +179,7 @@ stop(FsmRef) ->
%% ignore |
%% {stop, StopReason}
%%----------------------------------------------------------------------
init([{SockMod, Socket}, Opts]) ->
init([{SockMod, Socket}, Opts, FSMLimitOpts]) ->
Access = case lists:keysearch(access, 1, Opts) of
{value, {_, A}} -> A;
_ -> all
@ -192,7 +203,12 @@ init([{SockMod, Socket}, Opts]) ->
(_) -> false
end, Opts),
TLSOpts = [verify_none | TLSOpts1],
IP = peerip(SockMod, Socket),
IP = case lists:keysearch(frontend_ip, 1, Opts) of
{value, {_, IP1}} ->
IP1;
_ ->
peerip(SockMod, Socket)
end,
%% Check if IP is blacklisted:
case is_ip_blacklisted(IP) of
true ->
@ -220,8 +236,35 @@ init([{SockMod, Socket}, Opts]) ->
streamid = new_id(),
access = Access,
shaper = Shaper,
ip = IP},
ip = IP,
fsm_limit_opts = FSMLimitOpts},
?C2S_OPEN_TIMEOUT}
end;
init([StateName, StateData, _FSMLimitOpts]) ->
MRef = (StateData#state.sockmod):monitor(StateData#state.socket),
if StateName == session_established ->
Conn = get_conn_type(StateData),
Info = [{ip, StateData#state.ip}, {conn, Conn},
{auth_module, StateData#state.auth_module}],
{Time, _} = StateData#state.sid,
SID = {Time, self()},
Priority = case StateData#state.pres_last of
undefined ->
undefined;
El ->
get_priority_from_presence(El)
end,
ejabberd_sm:open_session(
SID,
StateData#state.user,
StateData#state.server,
StateData#state.resource,
Priority,
Info),
NewStateData = StateData#state{sid = SID, socket_monitor = MRef},
{ok, StateName, NewStateData};
true ->
{ok, StateName, StateData#state{socket_monitor = MRef}}
end.
%% Return list of all available resources of contacts,
@ -469,13 +512,13 @@ wait_for_auth({xmlstreamelement, El}, StateData) ->
jlib:jid_to_string(JID), AuthModule]),
SID = {now(), self()},
Conn = get_conn_type(StateData),
Info = [{ip, StateData#state.ip}, {conn, Conn},
{auth_module, AuthModule}],
%% Info = [{ip, StateData#state.ip}, {conn, Conn},
%% {auth_module, AuthModule}],
Res1 = jlib:make_result_iq_reply(El),
Res = setelement(4, Res1, []),
send_element(StateData, Res),
ejabberd_sm:open_session(
SID, U, StateData#state.server, R, Info),
%% ejabberd_sm:open_session(
%% SID, U, StateData#state.server, R, Info),
change_shaper(StateData, JID),
{Fs, Ts} = ejabberd_hooks:run_fold(
roster_get_subscription_lists,
@ -502,8 +545,7 @@ wait_for_auth({xmlstreamelement, El}, StateData) ->
pres_f = ?SETS:from_list(Fs1),
pres_t = ?SETS:from_list(Ts1),
privacy_list = PrivList},
fsm_next_state_pack(session_established,
NewStateData);
maybe_migrate(session_established, NewStateData);
_ ->
?INFO_MSG(
"(~w) Failed legacy authentication for ~s",
@ -627,7 +669,7 @@ wait_for_feature_request({xmlstreamelement, El}, StateData) ->
Socket = StateData#state.socket,
TLSSocket = (StateData#state.sockmod):starttls(
Socket, TLSOpts,
xml:element_to_string(
xml:element_to_binary(
{xmlelement, "proceed", [{"xmlns", ?NS_TLS}], []})),
fsm_next_state(wait_for_stream,
StateData#state{socket = TLSSocket,
@ -650,7 +692,7 @@ wait_for_feature_request({xmlstreamelement, El}, StateData) ->
Socket = StateData#state.socket,
ZlibSocket = (StateData#state.sockmod):compress(
Socket,
xml:element_to_string(
xml:element_to_binary(
{xmlelement, "compressed",
[{"xmlns", ?NS_COMPRESS}], []})),
fsm_next_state(wait_for_stream,
@ -831,7 +873,7 @@ wait_for_session({xmlstreamelement, El}, StateData) ->
case jlib:iq_query_info(El) of
#iq{type = set, xmlns = ?NS_SESSION} ->
U = StateData#state.user,
R = StateData#state.resource,
%%R = StateData#state.resource,
JID = StateData#state.jid,
case acl:match_rule(StateData#state.server,
StateData#state.access, JID) of
@ -857,10 +899,10 @@ wait_for_session({xmlstreamelement, El}, StateData) ->
[U, StateData#state.server]),
SID = {now(), self()},
Conn = get_conn_type(StateData),
Info = [{ip, StateData#state.ip}, {conn, Conn},
{auth_module, StateData#state.auth_module}],
ejabberd_sm:open_session(
SID, U, StateData#state.server, R, Info),
%% Info = [{ip, StateData#state.ip}, {conn, Conn},
%% {auth_module, StateData#state.auth_module}],
%% ejabberd_sm:open_session(
%% SID, U, StateData#state.server, R, Info),
NewStateData =
StateData#state{
sid = SID,
@ -868,8 +910,7 @@ wait_for_session({xmlstreamelement, El}, StateData) ->
pres_f = ?SETS:from_list(Fs1),
pres_t = ?SETS:from_list(Ts1),
privacy_list = PrivList},
fsm_next_state_pack(session_established,
NewStateData);
maybe_migrate(session_established, NewStateData);
_ ->
ejabberd_hooks:run(forbidden_session_hook,
StateData#state.server, [JID]),
@ -1047,6 +1088,8 @@ session_established2(El, StateData) ->
%% {next_state, NextStateName, NextStateData, Timeout} |
%% {stop, Reason, NewStateData}
%%----------------------------------------------------------------------
handle_event({migrate, Node, After}, StateName, StateData) when Node /= node() ->
fsm_migrate(StateName, StateData, Node, After * 2);
handle_event(_Event, StateName, StateData) ->
fsm_next_state(StateName, StateData).
@ -1382,6 +1425,22 @@ print_state(State = #state{pres_t = T, pres_f = F, pres_a = A, pres_i = I}) ->
%% Purpose: Shutdown the fsm
%% Returns: any
%%----------------------------------------------------------------------
terminate({migrated, ClonePid}, StateName, StateData) ->
if StateName == session_established ->
?INFO_MSG("(~w) Migrating ~s to ~p on node ~p",
[StateData#state.socket,
jlib:jid_to_string(StateData#state.jid),
ClonePid, node(ClonePid)]),
ejabberd_sm:close_session(StateData#state.sid,
StateData#state.user,
StateData#state.server,
StateData#state.resource);
true ->
ok
end,
(StateData#state.sockmod):change_controller(
StateData#state.socket, ClonePid),
ok;
terminate(_Reason, StateName, StateData) ->
case StateName of
session_established ->
@ -1453,14 +1512,14 @@ change_shaper(StateData, JID) ->
(StateData#state.sockmod):change_shaper(StateData#state.socket, Shaper).
send_text(StateData, Text) ->
?DEBUG("Send XML on stream = ~p", [lists:flatten(Text)]),
?DEBUG("Send XML on stream = ~p", [Text]),
(StateData#state.sockmod):send(StateData#state.socket, Text).
send_element(StateData, El) when StateData#state.xml_socket ->
(StateData#state.sockmod):send_xml(StateData#state.socket,
{xmlstreamelement, El});
send_element(StateData, El) ->
send_text(StateData, xml:element_to_string(El)).
send_text(StateData, xml:element_to_binary(El)).
send_header(StateData, Server, Version, Lang)
when StateData#state.xml_socket ->
@ -1550,16 +1609,21 @@ get_auth_tags([], U, P, D, R) ->
get_conn_type(StateData) ->
case (StateData#state.sockmod):get_sockmod(StateData#state.socket) of
gen_tcp -> c2s;
tls -> c2s_tls;
ejabberd_zlib ->
case ejabberd_zlib:get_sockmod((StateData#state.socket)#socket_state.socket) of
gen_tcp -> c2s_compressed;
tls -> c2s_compressed_tls
end;
ejabberd_http_poll -> http_poll;
ejabberd_http_bind -> http_bind;
_ -> unknown
gen_tcp -> c2s;
tls -> c2s_tls;
ejabberd_zlib ->
if is_pid(StateData#state.socket) ->
unknown;
true ->
case ejabberd_zlib:get_sockmod(
(StateData#state.socket)#socket_state.socket) of
gen_tcp -> c2s_compressed;
tls -> c2s_compressed_tls
end
end;
ejabberd_http_poll -> http_poll;
ejabberd_http_bind -> http_bind;
_ -> unknown
end.
process_presence_probe(From, To, StateData) ->
@ -2158,16 +2222,28 @@ peerip(SockMod, Socket) ->
_ -> undefined
end.
%% fsm_next_state_pack: Pack the StateData structure to improve
%% sharing.
fsm_next_state_pack(StateName, StateData) ->
fsm_next_state_gc(StateName, pack(StateData)).
%% fsm_next_state_gc: Garbage collect the process heap to make use of
%% the newly packed StateData structure.
fsm_next_state_gc(StateName, PackedStateData) ->
erlang:garbage_collect(),
fsm_next_state(StateName, PackedStateData).
maybe_migrate(StateName, StateData) ->
PackedStateData = pack(StateData),
case ejabberd_cluster:get_node({StateData#state.user,
StateData#state.server}) of
Node when Node == node() ->
Conn = get_conn_type(StateData),
Info = [{ip, StateData#state.ip}, {conn, Conn},
{auth_module, StateData#state.auth_module}],
#state{user = U, server = S, resource = R, sid = SID} = StateData,
ejabberd_sm:open_session(SID, U, S, R, Info),
erlang:garbage_collect(),
case ejabberd_cluster:get_node_new({U, S}) of
Node ->
ok;
NewNode ->
After = ejabberd_cluster:rehash_timeout(),
migrate(self(), NewNode, After)
end,
fsm_next_state(StateName, PackedStateData);
Node ->
fsm_migrate(StateName, PackedStateData, Node, 0)
end.
%% fsm_next_state: Generate the next_state FSM tuple with different
%% timeout, depending on the future state
@ -2176,6 +2252,10 @@ fsm_next_state(session_established, StateData) ->
fsm_next_state(StateName, StateData) ->
{next_state, StateName, StateData, ?C2S_OPEN_TIMEOUT}.
fsm_migrate(StateName, StateData, Node, Timeout) ->
{migrate, StateData,
{Node, ?MODULE, start, [StateName, StateData]}, Timeout}.
%% fsm_reply: Generate the reply FSM tuple with different timeout,
%% depending on the future state
fsm_reply(Reply, session_established, StateData) ->

151
src/ejabberd_captcha.erl
View File

@ -35,7 +35,7 @@
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-export([create_captcha/6, build_captcha_html/2, check_captcha/2,
-export([create_captcha/5, build_captcha_html/2, check_captcha/2,
process_reply/1, process/2, is_feature_available/0]).
-include("jlib.hrl").
@ -48,19 +48,11 @@
-define(CAPTCHA_TEXT(Lang), translate:translate(Lang, "Enter the text you see")).
-define(CAPTCHA_LIFETIME, 120000). % two minutes
-define(RPC_TIMEOUT, 5000).
-record(state, {}).
-record(captcha, {id, pid, key, tref, args}).
-define(T(S),
case catch mnesia:transaction(fun() -> S end) of
{atomic, Res} ->
Res;
{_, Reason} ->
?ERROR_MSG("mnesia transaction failed: ~p", [Reason]),
{error, Reason}
end).
%%====================================================================
%% API
%%====================================================================
@ -71,11 +63,12 @@
start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
create_captcha(Id, SID, From, To, Lang, Args)
when is_list(Id), is_list(Lang), is_list(SID),
create_captcha(SID, From, To, Lang, Args)
when is_list(Lang), is_list(SID),
is_record(From, jid), is_record(To, jid) ->
case create_image() of
{ok, Type, Key, Image} ->
Id = randoms:get_string() ++ "-" ++ ejabberd_cluster:node_id(),
B64Image = jlib:encode_base64(binary_to_list(Image)),
JID = jlib:jid_to_string(From),
CID = "sha1+" ++ sha:sha(Image) ++ "@bob.xmpp.org",
@ -101,13 +94,9 @@ create_captcha(Id, SID, From, To, Lang, Args)
OOB = {xmlelement, "x", [{"xmlns", ?NS_OOB}],
[{xmlelement, "url", [], [{xmlcdata, get_url(Id)}]}]},
Tref = erlang:send_after(?CAPTCHA_LIFETIME, ?MODULE, {remove_id, Id}),
case ?T(mnesia:write(#captcha{id=Id, pid=self(), key=Key,
tref=Tref, args=Args})) of
ok ->
{ok, [Body, OOB, Captcha, Data]};
_Err ->
error
end;
ets:insert(captcha, #captcha{id=Id, pid=self(), key=Key,
tref=Tref, args=Args}),
{ok, Id, [Body, OOB, Captcha, Data]};
_Err ->
error
end.
@ -119,8 +108,8 @@ create_captcha(Id, SID, From, To, Lang, Args)
%% IdEl = xmlelement()
%% KeyEl = xmlelement()
build_captcha_html(Id, Lang) ->
case mnesia:dirty_read(captcha, Id) of
[#captcha{}] ->
case lookup_captcha(Id) of
{ok, _} ->
ImgEl = {xmlelement, "img", [{"src", get_url(Id ++ "/image")}], []},
TextEl = {xmlcdata, ?CAPTCHA_TEXT(Lang)},
IdEl = {xmlelement, "input", [{"type", "hidden"},
@ -150,21 +139,25 @@ build_captcha_html(Id, Lang) ->
%% @spec (Id::string(), ProvidedKey::string()) -> captcha_valid | captcha_non_valid | captcha_not_found
check_captcha(Id, ProvidedKey) ->
?T(case mnesia:read(captcha, Id, write) of
[#captcha{pid=Pid, args=Args, key=StoredKey, tref=Tref}] ->
mnesia:delete({captcha, Id}),
erlang:cancel_timer(Tref),
if StoredKey == ProvidedKey ->
Pid ! {captcha_succeed, Args},
captcha_valid;
true ->
Pid ! {captcha_failed, Args},
captcha_non_valid
end;
_ ->
captcha_not_found
end).
case string:tokens(Id, "-") of
[_, NodeID] ->
case ejabberd_cluster:get_node_by_id(NodeID) of
Node when Node == node() ->
do_check_captcha(Id, ProvidedKey);
Node ->
case catch rpc:call(Node, ?MODULE, check_captcha,
[Id, ProvidedKey], ?RPC_TIMEOUT) of
{'EXIT', _} ->
captcha_not_found;
{badrpc, _} ->
captcha_not_found;
Res ->
Res
end
end;
_ ->
captcha_not_found
end.
process_reply({xmlelement, "captcha", _, _} = El) ->
case xml:get_subtag(El, "x") of
@ -175,20 +168,14 @@ process_reply({xmlelement, "captcha", _, _} = El) ->
case {proplists:get_value("challenge", Fields),
proplists:get_value("ocr", Fields)} of
{[Id|_], [OCR|_]} ->
?T(case mnesia:read(captcha, Id, write) of
[#captcha{pid=Pid, args=Args, key=Key, tref=Tref}] ->
mnesia:delete({captcha, Id}),
erlang:cancel_timer(Tref),
if OCR == Key ->
Pid ! {captcha_succeed, Args},
ok;
true ->
Pid ! {captcha_failed, Args},
{error, bad_match}
end;
_ ->
{error, not_found}
end);
case check_captcha(Id, OCR) of
captcha_valid ->
ok;
captcha_non_valid ->
{error, bad_match};
captcha_not_found ->
{error, not_found}
end;
_ ->
{error, malformed}
end
@ -209,8 +196,8 @@ process(_Handlers, #request{method='GET', lang=Lang, path=[_, Id]}) ->
end;
process(_Handlers, #request{method='GET', path=[_, Id, "image"]}) ->
case mnesia:dirty_read(captcha, Id) of
[#captcha{key=Key}] ->
case lookup_captcha(Id) of
{ok, #captcha{key=Key}} ->
case create_image(Key) of
{ok, Type, _, Img} ->
{200,
@ -249,10 +236,8 @@ process(_Handlers, _Request) ->
%% gen_server callbacks
%%====================================================================
init([]) ->
mnesia:create_table(captcha,
[{ram_copies, [node()]},
{attributes, record_info(fields, captcha)}]),
mnesia:add_table_copy(captcha, node(), ram_copies),
mnesia:delete_table(captcha),
ets:new(captcha, [named_table, public, {keypos, #captcha.id}]),
check_captcha_setup(),
{ok, #state{}}.
@ -264,13 +249,13 @@ handle_cast(_Msg, State) ->
handle_info({remove_id, Id}, State) ->
?DEBUG("captcha ~p timed out", [Id]),
_ = ?T(case mnesia:read(captcha, Id, write) of
[#captcha{args=Args, pid=Pid}] ->
Pid ! {captcha_failed, Args},
mnesia:delete({captcha, Id});
_ ->
ok
end),
case ets:lookup(captcha, Id) of
[#captcha{args=Args, pid=Pid}] ->
Pid ! {captcha_failed, Args},
ets:delete(captcha, Id);
_ ->
ok
end,
{noreply, State};
handle_info(_Info, State) ->
@ -414,3 +399,43 @@ check_captcha_setup() ->
false ->
ok
end.
lookup_captcha(Id) ->
case string:tokens(Id, "-") of
[_, NodeID] ->
case ejabberd_cluster:get_node_by_id(NodeID) of
Node when Node == node() ->
case ets:lookup(captcha, Id) of
[C] ->
{ok, C};
_ ->
{error, enoent}
end;
Node ->
case catch rpc:call(Node, ets, lookup,
[captcha, Id], ?RPC_TIMEOUT) of
[C] ->
{ok, C};
_ ->
{error, enoent}
end
end;
_ ->
{error, enoent}
end.
do_check_captcha(Id, ProvidedKey) ->
case ets:lookup(captcha, Id) of
[#captcha{pid = Pid, args = Args, key = ValidKey, tref = Tref}] ->
ets:delete(captcha, Id),
erlang:cancel_timer(Tref),
if ValidKey == ProvidedKey ->
Pid ! {captcha_succeed, Args},
captcha_valid;
true ->
Pid ! {captcha_failed, Args},
captcha_non_valid
end;
_ ->
captcha_not_found
end.

177
src/ejabberd_cluster.erl Normal file
View File

@ -0,0 +1,177 @@
%%%-------------------------------------------------------------------
%%% File : ejabberd_cluster.erl
%%% Author : Evgeniy Khramtsov <ekhramtsov@process-one.net>
%%% Description :
%%%
%%% Created : 2 Apr 2010 by Evgeniy Khramtsov <ekhramtsov@process-one.net>
%%%-------------------------------------------------------------------
-module(ejabberd_cluster).
-behaviour(gen_server).
%% API
-export([start_link/0, get_node/1, get_node_new/1, announce/0,
node_id/0, get_node_by_id/1, get_nodes/0, rehash_timeout/0]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-include("ejabberd.hrl").
-define(HASHTBL, nodes_hash).
-define(HASHTBL_NEW, nodes_hash_new).
-define(POINTS, 16).
-define(REHASH_TIMEOUT, 5000).
-record(state, {}).
%%====================================================================
%% API
%%====================================================================
start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
get_node(Key) ->
Hash = erlang:phash2(Key),
get_node_by_hash(?HASHTBL, Hash).
get_node_new(Key) ->
Hash = erlang:phash2(Key),
get_node_by_hash(?HASHTBL_NEW, Hash).
get_nodes() ->
%% TODO
mnesia:system_info(running_db_nodes).
announce() ->
gen_server:call(?MODULE, announce, infinity).
node_id() ->
integer_to_list(erlang:phash2(node())).
rehash_timeout() ->
?REHASH_TIMEOUT.
get_node_by_id(NodeID) when is_list(NodeID) ->
case catch list_to_existing_atom(NodeID) of
{'EXIT', _} ->
node();
Res ->
get_node_by_id(Res)
end;
get_node_by_id(NodeID) ->
case global:whereis_name(NodeID) of
Pid when is_pid(Pid) ->
node(Pid);
_ ->
node()
end.
%%====================================================================
%% gen_server callbacks
%%====================================================================
init([]) ->
net_kernel:monitor_nodes(true, [{node_type, visible}]),
ets:new(?HASHTBL, [named_table, public, ordered_set]),
ets:new(?HASHTBL_NEW, [named_table, public, ordered_set]),
register_node(),
AllNodes = mnesia:system_info(running_db_nodes),
OtherNodes = case AllNodes of
[_] ->
AllNodes;
_ ->
AllNodes -- [node()]
end,
append_nodes(?HASHTBL, OtherNodes),
append_nodes(?HASHTBL_NEW, AllNodes),
{ok, #state{}}.
handle_call(announce, _From, State) ->
case mnesia:system_info(running_db_nodes) of
[_MyNode] ->
ok;
Nodes ->
OtherNodes = Nodes -- [node()],
lists:foreach(
fun(Node) ->
{?MODULE, Node} ! {node_ready, node()}
end, OtherNodes),
?INFO_MSG("waiting for migration from nodes: ~w",
[OtherNodes]),
timer:sleep(?REHASH_TIMEOUT),
append_node(?HASHTBL, node())
end,
{reply, ok, State};
handle_call(_Request, _From, State) ->
Reply = ok,
{reply, Reply, State}.
handle_cast(_Msg, State) ->
{noreply, State}.
handle_info({node_ready, Node}, State) ->
?INFO_MSG("node ~p is ready, starting migration", [Node]),
append_node(?HASHTBL_NEW, Node),
ejabberd_hooks:run(node_hash_update, [?REHASH_TIMEOUT]),
timer:sleep(?REHASH_TIMEOUT),
?INFO_MSG("adding node ~p to hash", [Node]),
append_node(?HASHTBL, Node),
{noreply, State};
handle_info({nodedown, Node, _}, State) ->
?INFO_MSG("node ~p goes down", [Node]),
delete_node(?HASHTBL, Node),
delete_node(?HASHTBL_NEW, Node),
{noreply, State};
handle_info(_Info, State) ->
{noreply, State}.
terminate(_Reason, _State) ->
ok.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%%--------------------------------------------------------------------
%% Internal functions
%%--------------------------------------------------------------------
append_nodes(Tab, Nodes) ->
lists:foreach(
fun(Node) ->
append_node(Tab, Node)
end, Nodes).
append_node(Tab, Node) ->
lists:foreach(
fun(I) ->
Hash = erlang:phash2({I, Node}),
ets:insert(Tab, {Hash, Node})
end, lists:seq(1, ?POINTS)).
delete_node(Tab, Node) ->
lists:foreach(
fun(I) ->
Hash = erlang:phash2({I, Node}),
ets:delete(Tab, Hash)
end, lists:seq(1, ?POINTS)).
get_node_by_hash(Tab, Hash) ->
NodeHash = case ets:next(Tab, Hash) of
'$end_of_table' ->
ets:first(Tab);
NH ->
NH
end,
if NodeHash == '$end_of_table' ->
erlang:error(no_running_nodes);
true ->
case ets:lookup(Tab, NodeHash) of
[] ->
get_node_by_hash(Tab, Hash);
[{_, Node}] ->
Node
end
end.
register_node() ->
global:register_name(list_to_atom(node_id()), self()).

3
src/ejabberd_ctl.erl
View File

@ -201,8 +201,7 @@ process(Args) ->
case String of
[] -> ok;
_ ->
io:format(String),
io:format("\n")
io:format("~s~n", [String])
end,
Code.

97
src/ejabberd_debug.erl
View File

@ -1,97 +0,0 @@
%%%----------------------------------------------------------------------
%%% File : ejabberd_debug.erl
%%% Author : Mickael Remond
%%% Purpose : ejabberd's application callback module
%%% Created : 6 may 2009 by Mickael Remond <mremond@process-one.net>
%%%
%%%
%%% ejabberd, Copyright (C) 2002-2010 ProcessOne
%%%
%%% This program is free software; you can redistribute it and/or
%%% modify it under the terms of the GNU General Public License as
%%% published by the Free Software Foundation; either version 2 of the
%%% License, or (at your option) any later version.
%%%
%%% This program is distributed in the hope that it will be useful,
%%% but WITHOUT ANY WARRANTY; without even the implied warranty of
%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
%%% General Public License for more details.
%%%
%%% You should have received a copy of the GNU General Public License
%%% along with this program; if not, write to the Free Software
%%% Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
%%% 02111-1307 USA
%%%
%%%----------------------------------------------------------------------
-module(ejabberd_debug).
-export([eprof_start/0, fprof_start/0, stop/0]).
-export([pids/0]).
eprof_start() ->
eprof:start(),
eprof:profile(pids()).
fprof_start() ->
fprof:trace([start, {file, "/tmp/fprof"}, {procs, pids()}]).
%% Stop all profilers
stop() ->
catch eprof:stop(),
catch fprof:stop(),
ok.
pids() ->
lists:zf(
fun(Pid) ->
case process_info(Pid) of
ProcessInfo when is_list(ProcessInfo) ->
CurrentFunction = current_function(ProcessInfo),
InitialCall = initial_call(ProcessInfo),
RegisteredName = registered_name(ProcessInfo),
Ancestor = ancestor(ProcessInfo),
filter_pid(Pid, CurrentFunction, InitialCall, RegisteredName, Ancestor);
_ ->
false
end
end,
processes()).
current_function(ProcessInfo) ->
{value, {_, {CurrentFunction, _,_}}} =
lists:keysearch(current_function, 1, ProcessInfo),
atom_to_list(CurrentFunction).
initial_call(ProcessInfo) ->
{value, {_, {InitialCall, _,_}}} =
lists:keysearch(initial_call, 1, ProcessInfo),
atom_to_list(InitialCall).
registered_name(ProcessInfo) ->
case lists:keysearch(registered_name, 1, ProcessInfo) of
{value, {_, Name}} when is_atom(Name) -> atom_to_list(Name);
_ -> ""
end.
ancestor(ProcessInfo) ->
{value, {_, Dictionary}} = lists:keysearch(dictionary, 1, ProcessInfo),
case lists:keysearch('$ancestors', 1, Dictionary) of
{value, {_, [Ancestor|_T]}} when is_atom(Ancestor) ->
atom_to_list(Ancestor);
_ ->
""
end.
filter_pid(Pid, "ejabberd" ++ _, _InitialCall, _RegisteredName, _Ancestor) ->
{true, Pid};
filter_pid(Pid, _CurrentFunction, "ejabberd" ++ _, _RegisteredName, _Ancestor) ->
{true, Pid};
filter_pid(Pid, _CurrentFunction, _InitialCall, "ejabberd"++_, _Ancestor) ->
{true, Pid};
filter_pid(Pid, _CurrentFunction, _InitialCall, "stringprep"++_, _Ancestor) ->
{true, Pid};
filter_pid(Pid, _CurrentFunction, _InitialCall, _RegisteredName, "ejabberd"++_) ->
{true, Pid};
filter_pid(_Pid, _CurrentFunction, _InitialCall, _RegisteredName, _Ancestor) ->
false.

87
src/ejabberd_frontend_socket.erl
View File

@ -45,6 +45,8 @@
get_peer_certificate/1,
get_verify_result/1,
close/1,
setopts/2,
change_controller/2,
sockname/1, peername/1]).
%% gen_server callbacks
@ -94,18 +96,15 @@ start(Module, SockMod, Socket, Opts) ->
todo
end.
starttls(FsmRef, _TLSOpts) ->
%% TODO: Frontend improvements planned by Aleksey
%%gen_server:call(FsmRef, {starttls, TLSOpts}),
FsmRef.
starttls(FsmRef, TLSOpts) ->
starttls(FsmRef, TLSOpts, undefined).
starttls(FsmRef, TLSOpts, Data) ->
gen_server:call(FsmRef, {starttls, TLSOpts, Data}),
FsmRef.
compress(FsmRef) ->
gen_server:call(FsmRef, compress),
FsmRef.
compress(FsmRef, undefined).
compress(FsmRef, Data) ->
gen_server:call(FsmRef, {compress, Data}),
@ -138,11 +137,14 @@ close(FsmRef) ->
sockname(FsmRef) ->
gen_server:call(FsmRef, sockname).
peername(_FsmRef) ->
%% TODO: Frontend improvements planned by Aleksey
%%gen_server:call(FsmRef, peername).
{ok, {{0, 0, 0, 0}, 0}}.
setopts(FsmRef, Opts) ->
gen_server:call(FsmRef, {setopts, Opts}).
change_controller(FsmRef, C2SPid) ->
gen_server:call(FsmRef, {change_controller, C2SPid}).
peername(FsmRef) ->
gen_server:call(FsmRef, peername).
%%====================================================================
%% gen_server callbacks
@ -158,9 +160,16 @@ peername(_FsmRef) ->
init([Module, SockMod, Socket, Opts, Receiver]) ->
%% TODO: monitor the receiver
Node = ejabberd_node_groups:get_closest_node(backend),
IP = case peername(SockMod, Socket) of
{ok, IP1} ->
IP1;
_ ->
undefined
end,
{SockMod2, Socket2} = check_starttls(SockMod, Socket, Receiver, Opts),
{ok, Pid} =
rpc:call(Node, Module, start, [{?MODULE, self()}, Opts]),
rpc:call(Node, Module, start,
[{?MODULE, self()}, [{frontend_ip, IP} | Opts]]),
ejabberd_receiver:become_controller(Receiver, Pid),
{ok, #state{sockmod = SockMod2,
socket = Socket2,
@ -175,38 +184,16 @@ init([Module, SockMod, Socket, Opts, Receiver]) ->
%% {stop, Reason, State}
%% Description: Handling call messages
%%--------------------------------------------------------------------
handle_call({starttls, TLSOpts}, _From, State) ->
{ok, TLSSocket} = tls:tcp_to_tls(State#state.socket, TLSOpts),
ejabberd_receiver:starttls(State#state.receiver, TLSSocket),
Reply = ok,
{reply, Reply, State#state{socket = TLSSocket, sockmod = tls},
?HIBERNATE_TIMEOUT};
handle_call({starttls, TLSOpts, Data}, _From, State) ->
{ok, TLSSocket} = tls:tcp_to_tls(State#state.socket, TLSOpts),
ejabberd_receiver:starttls(State#state.receiver, TLSSocket),
catch (State#state.sockmod):send(
State#state.socket, Data),
{ok, TLSSocket} = ejabberd_receiver:starttls(
State#state.receiver, TLSOpts, Data),
Reply = ok,
{reply, Reply, State#state{socket = TLSSocket, sockmod = tls},
?HIBERNATE_TIMEOUT};
handle_call(compress, _From, State) ->
{ok, ZlibSocket} = ejabberd_zlib:enable_zlib(
State#state.sockmod,
State#state.socket),
ejabberd_receiver:compress(State#state.receiver, ZlibSocket),
Reply = ok,
{reply, Reply, State#state{socket = ZlibSocket, sockmod = ejabberd_zlib},
?HIBERNATE_TIMEOUT};
handle_call({compress, Data}, _From, State) ->
{ok, ZlibSocket} = ejabberd_zlib:enable_zlib(
State#state.sockmod,
State#state.socket),
ejabberd_receiver:compress(State#state.receiver, ZlibSocket),
catch (State#state.sockmod):send(
State#state.socket, Data),
{ok, ZlibSocket} = ejabberd_receiver:compress(
State#state.receiver, Data),
Reply = ok,
{reply, Reply, State#state{socket = ZlibSocket, sockmod = ejabberd_zlib},
?HIBERNATE_TIMEOUT};
@ -246,13 +233,7 @@ handle_call(close, _From, State) ->
handle_call(sockname, _From, State) ->
#state{sockmod = SockMod, socket = Socket} = State,
Reply =
case SockMod of
gen_tcp ->
inet:sockname(Socket);
_ ->
SockMod:sockname(Socket)
end,
Reply = peername(SockMod, Socket),
{reply, Reply, State, ?HIBERNATE_TIMEOUT};
handle_call(peername, _From, State) ->
@ -266,6 +247,14 @@ handle_call(peername, _From, State) ->
end,
{reply, Reply, State, ?HIBERNATE_TIMEOUT};
handle_call({setopts, Opts}, _From, State) ->
ejabberd_receiver:setopts(State#state.receiver, Opts),
{reply, ok, State, ?HIBERNATE_TIMEOUT};
handle_call({change_controller, Pid}, _From, State) ->
ejabberd_receiver:change_controller(State#state.receiver, Pid),
{reply, ok, State, ?HIBERNATE_TIMEOUT};
handle_call(_Request, _From, State) ->
Reply = ok,
{reply, Reply, State, ?HIBERNATE_TIMEOUT}.
@ -318,10 +307,16 @@ check_starttls(SockMod, Socket, Receiver, Opts) ->
end, Opts),
if
TLSEnabled ->
{ok, TLSSocket} = tls:tcp_to_tls(Socket, TLSOpts),
ejabberd_receiver:starttls(Receiver, TLSSocket),
{ok, TLSSocket} = ejabberd_receiver:starttls(Receiver, TLSOpts),
{tls, TLSSocket};
true ->
{SockMod, Socket}
end.
peername(SockMod, Socket) ->
case SockMod of
gen_tcp ->
inet:peername(Socket);
_ ->
SockMod:peername(Socket)
end.

50
src/ejabberd_local.erl
View File

@ -125,7 +125,7 @@ route(From, To, Packet) ->
route_iq(From, To, #iq{type = Type} = IQ, F) when is_function(F) ->
Packet = if Type == set; Type == get ->
ID = randoms:get_string(),
ID = ejabberd_router:make_id(),
Host = From#jid.lserver,
register_iq_response_handler(Host, ID, undefined, F),
jlib:iq_to_xml(IQ#iq{id = ID});
@ -136,10 +136,10 @@ route_iq(From, To, #iq{type = Type} = IQ, F) when is_function(F) ->
register_iq_response_handler(_Host, ID, Module, Function) ->
TRef = erlang:start_timer(?IQ_TIMEOUT, ejabberd_local, ID),
mnesia:dirty_write(#iq_response{id = ID,
module = Module,
function = Function,
timer = TRef}).
ets:insert(iq_response, #iq_response{id = ID,
module = Module,
function = Function,
timer = TRef}).
register_iq_handler(Host, XMLNS, Module, Fun) ->
ejabberd_local ! {register_iq_handler, Host, XMLNS, Module, Fun}.
@ -181,11 +181,9 @@ init([]) ->
?MODULE, bounce_resource_packet, 100)
end, ?MYHOSTS),
catch ets:new(?IQTABLE, [named_table, public]),
update_table(),
mnesia:create_table(iq_response,
[{ram_copies, [node()]},
{attributes, record_info(fields, iq_response)}]),
mnesia:add_table_copy(iq_response, node(), ram_copies),
mnesia:delete_table(iq_response),
catch ets:new(iq_response, [named_table, public,
{keypos, #iq_response.id}]),
{ok, #state{}}.
%%--------------------------------------------------------------------
@ -257,7 +255,7 @@ handle_info(refresh_iq_handlers, State) ->
end, ets:tab2list(?IQTABLE)),
{noreply, State};
handle_info({timeout, _TRef, ID}, State) ->
process_iq_timeout(ID),
spawn(fun() -> process_iq_timeout(ID) end),
{noreply, State};
handle_info(_Info, State) ->
{noreply, State}.
@ -312,40 +310,22 @@ do_route(From, To, Packet) ->
end
end.
update_table() ->
case catch mnesia:table_info(iq_response, attributes) of
[id, module, function] ->
mnesia:delete_table(iq_response);
[id, module, function, timer] ->
ok;
{'EXIT', _} ->
ok
end.
get_iq_callback(ID) ->
case mnesia:dirty_read(iq_response, ID) of
case ets:lookup(iq_response, ID) of
[#iq_response{module = Module, timer = TRef,
function = Function}] ->
cancel_timer(TRef),
mnesia:dirty_delete(iq_response, ID),
ets:delete(iq_response, ID),
{ok, Module, Function};
_ ->
error
end.
process_iq_timeout(ID) ->
spawn(fun process_iq_timeout/0) ! ID.
process_iq_timeout() ->
receive
ID ->
case get_iq_callback(ID) of
{ok, undefined, Function} ->
Function(timeout);
_ ->
ok
end
after 5000 ->
case get_iq_callback(ID) of
{ok, undefined, Function} ->
Function(timeout);
_ ->
ok
end.

40
src/ejabberd_node_groups.erl
View File

@ -31,6 +31,7 @@
%% API
-export([start_link/0,
start/0,
join/1,
leave/1,
get_members/1,
@ -40,7 +41,7 @@
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-record(state, {}).
-record(state, {groups = []}).
%%====================================================================
%% API
@ -49,6 +50,15 @@
%% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
%% Description: Starts the server
%%--------------------------------------------------------------------
start() ->
ChildSpec = {?MODULE,
{?MODULE, start_link, []},
permanent,
brutal_kill,
worker,
[?MODULE]},
supervisor:start_child(ejabberd_sup, ChildSpec).
start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
@ -81,30 +91,19 @@ get_closest_node(Name) ->
%% Description: Initiates the server
%%--------------------------------------------------------------------
init([]) ->
{FE, BE} =
Groups =
case ejabberd_config:get_local_option(node_type) of
frontend ->
{true, false};
[frontend];
backend ->
{false, true};
[backend];
generic ->
{true, true};
[frontend, backend];
undefined ->
{true, true}
[frontend, backend]
end,
if
FE ->
join(frontend);
true ->
ok
end,
if
BE ->
join(backend);
true ->
ok
end,
{ok, #state{}}.
lists:foreach(fun join/1, Groups),
{ok, #state{groups = Groups}}.
%%--------------------------------------------------------------------
%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
@ -144,7 +143,8 @@ handle_info(_Info, State) ->
%% cleaning up. When it returns, the gen_server terminates with Reason.
%% The return value is ignored.
%%--------------------------------------------------------------------
terminate(_Reason, _State) ->
terminate(_Reason, #state{groups = Groups}) ->
lists:foreach(fun leave/1, Groups),
ok.
%%--------------------------------------------------------------------

125
src/ejabberd_receiver.erl
View File

@ -36,8 +36,12 @@
change_shaper/2,
reset_stream/1,
starttls/2,
starttls/3,
compress/2,
send/2,
become_controller/2,
change_controller/2,
setopts/2,
close/1]).
%% gen_server callbacks
@ -52,6 +56,7 @@
c2s_pid,
max_stanza_size,
xml_stream_state,
tref,
timeout}).
-define(HIBERNATE_TIMEOUT, 90000).
@ -86,15 +91,32 @@ change_shaper(Pid, Shaper) ->
reset_stream(Pid) ->
gen_server:call(Pid, reset_stream).
starttls(Pid, TLSSocket) ->
gen_server:call(Pid, {starttls, TLSSocket}).
starttls(Pid, TLSOpts) ->
starttls(Pid, TLSOpts, undefined).
compress(Pid, ZlibSocket) ->
gen_server:call(Pid, {compress, ZlibSocket}).
starttls(Pid, TLSOpts, Data) ->
gen_server:call(Pid, {starttls, TLSOpts, Data}).
compress(Pid, Data) ->
gen_server:call(Pid, {compress, Data}).
become_controller(Pid, C2SPid) ->
gen_server:call(Pid, {become_controller, C2SPid}).
change_controller(Pid, C2SPid) ->
gen_server:call(Pid, {change_controller, C2SPid}).
setopts(Pid, Opts) ->
case lists:member({active, false}, Opts) of
true ->
gen_server:call(Pid, deactivate_socket);
false ->
ok
end.
send(Pid, Data) ->
gen_server:call(Pid, {send, Data}).
close(Pid) ->
gen_server:cast(Pid, close).
@ -132,10 +154,17 @@ init([Socket, SockMod, Shaper, MaxStanzaSize]) ->
%% {stop, Reason, State}
%% Description: Handling call messages
%%--------------------------------------------------------------------
handle_call({starttls, TLSSocket}, _From,
handle_call({starttls, TLSOpts, Data}, _From,
#state{xml_stream_state = XMLStreamState,
c2s_pid = C2SPid,
socket = Socket,
max_stanza_size = MaxStanzaSize} = State) ->
{ok, TLSSocket} = tls:tcp_to_tls(Socket, TLSOpts),
if Data /= undefined ->
do_send(State, Data);
true ->
ok
end,
close_stream(XMLStreamState),
NewXMLStreamState = xml_stream:new(C2SPid, MaxStanzaSize),
NewState = State#state{socket = TLSSocket,
@ -143,14 +172,23 @@ handle_call({starttls, TLSSocket}, _From,
xml_stream_state = NewXMLStreamState},
case tls:recv_data(TLSSocket, "") of
{ok, TLSData} ->
{reply, ok, process_data(TLSData, NewState), ?HIBERNATE_TIMEOUT};
{reply, {ok, TLSSocket},
process_data(TLSData, NewState), ?HIBERNATE_TIMEOUT};
{error, _Reason} ->
{stop, normal, ok, NewState}
end;
handle_call({compress, ZlibSocket}, _From,
handle_call({compress, Data}, _From,
#state{xml_stream_state = XMLStreamState,
c2s_pid = C2SPid,
socket = Socket,
sock_mod = SockMod,
max_stanza_size = MaxStanzaSize} = State) ->
{ok, ZlibSocket} = ejabberd_zlib:enable_zlib(SockMod, Socket),
if Data /= undefined ->
do_send(State, Data);
true ->
ok
end,
close_stream(XMLStreamState),
NewXMLStreamState = xml_stream:new(C2SPid, MaxStanzaSize),
NewState = State#state{socket = ZlibSocket,
@ -158,7 +196,8 @@ handle_call({compress, ZlibSocket}, _From,
xml_stream_state = NewXMLStreamState},
case ejabberd_zlib:recv_data(ZlibSocket, "") of
{ok, ZlibData} ->
{reply, ok, process_data(ZlibData, NewState), ?HIBERNATE_TIMEOUT};
{reply, {ok, ZlibSocket},
process_data(ZlibData, NewState), ?HIBERNATE_TIMEOUT};
{error, _Reason} ->
{stop, normal, ok, NewState}
end;
@ -172,12 +211,31 @@ handle_call(reset_stream, _From,
{reply, Reply, State#state{xml_stream_state = NewXMLStreamState},
?HIBERNATE_TIMEOUT};
handle_call({become_controller, C2SPid}, _From, State) ->
erlang:monitor(process, C2SPid),
XMLStreamState = xml_stream:new(C2SPid, State#state.max_stanza_size),
NewState = State#state{c2s_pid = C2SPid,
xml_stream_state = XMLStreamState},
activate_socket(NewState),
Reply = ok,
{reply, Reply, NewState, ?HIBERNATE_TIMEOUT};
handle_call({change_controller, C2SPid}, _From, State) ->
erlang:monitor(process, C2SPid),
NewXMLStreamState = xml_stream:change_callback_pid(
State#state.xml_stream_state, C2SPid),
NewState = State#state{c2s_pid = C2SPid,
xml_stream_state = NewXMLStreamState},
activate_socket(NewState),
{reply, ok, NewState, ?HIBERNATE_TIMEOUT};
handle_call({send, Data}, _From, State) ->
case do_send(State, Data) of
ok ->
{reply, ok, State, ?HIBERNATE_TIMEOUT};
{error, _Reason} = Err ->
{stop, normal, Err, State}
end;
handle_call(deactivate_socket, _From, State) ->
deactivate_socket(State),
{reply, ok, State, ?HIBERNATE_TIMEOUT};
handle_call(_Request, _From, State) ->
Reply = ok,
{reply, Reply, State, ?HIBERNATE_TIMEOUT}.
@ -237,6 +295,9 @@ handle_info({Tag, _TCPSocket, Reason}, State)
_ ->
{stop, normal, State}
end;
handle_info({'DOWN', _MRef, process, C2SPid, _},
#state{c2s_pid = C2SPid} = State) ->
{stop, normal, State};
handle_info({timeout, _Ref, activate}, State) ->
activate_socket(State),
{noreply, State, ?HIBERNATE_TIMEOUT};
@ -294,6 +355,17 @@ activate_socket(#state{socket = Socket,
ok
end.
deactivate_socket(#state{socket = Socket,
tref = TRef,
sock_mod = SockMod}) ->
cancel_timer(TRef),
case SockMod of
gen_tcp ->
inet:setopts(Socket, [{active, false}]);
_ ->
SockMod:setopts(Socket, [{active, false}])
end.
%% Data processing for connectors directly generating xmlelement in
%% Erlang data structure.
%% WARNING: Shaper does not work with Erlang data structure.
@ -315,20 +387,23 @@ process_data([Element|Els], #state{c2s_pid = C2SPid} = State)
%% Data processing for connectors receivind data as string.
process_data(Data,
#state{xml_stream_state = XMLStreamState,
tref = TRef,
shaper_state = ShaperState,
c2s_pid = C2SPid} = State) ->
?DEBUG("Received XML on stream = ~p", [binary_to_list(Data)]),
XMLStreamState1 = xml_stream:parse(XMLStreamState, Data),
{NewShaperState, Pause} = shaper:update(ShaperState, size(Data)),
if
C2SPid == undefined ->
ok;
Pause > 0 ->
erlang:start_timer(Pause, self(), activate);
true ->
activate_socket(State)
end,
NewTRef = if
C2SPid == undefined ->
TRef;
Pause > 0 ->
erlang:start_timer(Pause, self(), activate);
true ->
activate_socket(State),
TRef
end,
State#state{xml_stream_state = XMLStreamState1,
tref = NewTRef,
shaper_state = NewShaperState}.
%% Element coming from XML parser are wrapped inside xmlstreamelement
@ -345,3 +420,21 @@ close_stream(undefined) ->
ok;
close_stream(XMLStreamState) ->
xml_stream:close(XMLStreamState).
do_send(State, Data) ->
(State#state.sock_mod):send(State#state.socket, Data).
cancel_timer(TRef) when is_reference(TRef) ->
case erlang:cancel_timer(TRef) of
false ->
receive
{timeout, TRef, _} ->
ok
after 0 ->
ok
end;
_ ->
ok
end;
cancel_timer(_) ->
ok.

37
src/ejabberd_router.erl
View File

@ -38,7 +38,8 @@
unregister_route/1,
unregister_routes/1,
dirty_get_all_routes/0,
dirty_get_all_domains/0
dirty_get_all_domains/0,
make_id/0
]).
-export([start_link/0]).
@ -53,6 +54,9 @@
-record(route, {domain, pid, local_hint}).
-record(state, {}).
%% "rr" stands for Record-Route.
-define(ROUTE_PREFIX, "rr-").
%%====================================================================
%% API
%%====================================================================
@ -65,7 +69,7 @@ start_link() ->
route(From, To, Packet) ->
case catch do_route(From, To, Packet) of
case catch route_check_id(From, To, Packet) of
{'EXIT', Reason} ->
?ERROR_MSG("~p~nwhen processing: ~p",
[Reason, {From, To, Packet}]);
@ -192,6 +196,8 @@ dirty_get_all_routes() ->
dirty_get_all_domains() ->
lists:usort(mnesia:dirty_all_keys(route)).
make_id() ->
?ROUTE_PREFIX ++ randoms:get_string() ++ "-" ++ ejabberd_cluster:node_id().
%%====================================================================
%% gen_server callbacks
@ -309,6 +315,32 @@ code_change(_OldVsn, State, _Extra) ->
%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------
route_check_id(From, To, {xmlelement, "iq", Attrs, _} = Packet) ->
case xml:get_attr_s("id", Attrs) of
?ROUTE_PREFIX ++ Rest ->
Type = xml:get_attr_s("type", Attrs),
if Type == "error"; Type == "result" ->
case string:tokens(Rest, "-") of
[_, NodeID] ->
case ejabberd_cluster:get_node_by_id(NodeID) of
Node when Node == node() ->
do_route(From, To, Packet);
Node ->
{ejabberd_router, Node} !
{route, From, To, Packet}
end;
_ ->
do_route(From, To, Packet)
end;
true ->
do_route(From, To, Packet)
end;
_ ->
do_route(From, To, Packet)
end;
route_check_id(From, To, Packet) ->
do_route(From, To, Packet).
do_route(OrigFrom, OrigTo, OrigPacket) ->
?DEBUG("route~n\tfrom ~p~n\tto ~p~n\tpacket ~p~n",
[OrigFrom, OrigTo, OrigPacket]),
@ -413,4 +445,3 @@ update_tables() ->
false ->
ok
end.

137
src/ejabberd_s2s.erl
View File

@ -40,7 +40,8 @@
dirty_get_connections/0,
allow_host/2,
incoming_s2s_number/0,
outgoing_s2s_number/0
outgoing_s2s_number/0,
migrate/1
]).
%% gen_server callbacks
@ -94,30 +95,63 @@ remove_connection(FromTo, Pid, Key) ->
end.
have_connection(FromTo) ->
case catch mnesia:dirty_read(s2s, FromTo) of
[_] ->
true;
_ ->
false
case ejabberd_cluster:get_node(FromTo) of
Node when Node == node() ->
case mnesia:dirty_read(s2s, FromTo) of
[_] ->
true;
_ ->
false
end;
Node ->
case catch rpc:call(Node, mnesia, dirty_read,
[s2s, FromTo], 5000) of
[_] ->
true;
_ ->
false
end
end.
has_key(FromTo, Key) ->
case mnesia:dirty_select(s2s,
[{#s2s{fromto = FromTo, key = Key, _ = '_'},
[],
['$_']}]) of
[] ->
false;
_ ->
true
Query = [{#s2s{fromto = FromTo, key = Key, _ = '_'},
[],
['$_']}],
case ejabberd_cluster:get_node(FromTo) of
Node when Node == node() ->
case mnesia:dirty_select(s2s, Query) of
[] ->
false;
_ ->
true
end;
Node ->
case catch rpc:call(Node, mnesia, dirty_select,
[s2s, Query], 5000) of
[_|_] ->
true;
_ ->
false
end
end.
get_connections_pids(FromTo) ->
case catch mnesia:dirty_read(s2s, FromTo) of
L when is_list(L) ->
[Connection#s2s.pid || Connection <- L];
_ ->
[]
case ejabberd_cluster:get_node(FromTo) of
Node when Node == node() ->
case catch mnesia:dirty_read(s2s, FromTo) of
L when is_list(L) ->
[Connection#s2s.pid || Connection <- L];
_ ->
[]
end;
Node ->
case catch rpc:call(Node, mnesia, dirty_read,
[s2s, FromTo], 5000) of
L when is_list(L) ->
[Connection#s2s.pid || Connection <- L];
_ ->
[]
end
end.
try_register(FromTo) ->
@ -148,7 +182,33 @@ try_register(FromTo) ->
end.
dirty_get_connections() ->
mnesia:dirty_all_keys(s2s).
lists:flatmap(
fun(Node) when Node == node() ->
mnesia:dirty_all_keys(s2s);
(Node) ->
case catch rpc:call(Node, mnesia, dirty_all_keys, [s2s], 5000) of
L when is_list(L) ->
L;
_ ->
[]
end
end, ejabberd_cluster:get_nodes()).
migrate(After) ->
Ss = mnesia:dirty_select(
s2s,
[{#s2s{fromto = '$1', pid = '$2', _ = '_'},
[],
['$$']}]),
lists:foreach(
fun([FromTo, Pid]) ->
case ejabberd_cluster:get_node_new(FromTo) of
Node when Node /= node() ->
ejabberd_s2s_out:stop_connection(Pid, After * 2);
_ ->
ok
end
end, Ss).
%%====================================================================
%% gen_server callbacks
@ -163,10 +223,11 @@ dirty_get_connections() ->
%%--------------------------------------------------------------------
init([]) ->
update_tables(),
mnesia:create_table(s2s, [{ram_copies, [node()]}, {type, bag},
mnesia:create_table(s2s, [{ram_copies, [node()]},
{type, bag}, {local_content, true},
{attributes, record_info(fields, s2s)}]),
mnesia:add_table_copy(s2s, node(), ram_copies),
mnesia:subscribe(system),
ejabberd_hooks:add(node_hash_update, ?MODULE, migrate, 100),
ejabberd_commands:register_commands(commands()),
{ok, #state{}}.
@ -198,9 +259,6 @@ handle_cast(_Msg, State) ->
%% {stop, Reason, State}
%% Description: Handling all non call/cast messages
%%--------------------------------------------------------------------
handle_info({mnesia_system_event, {mnesia_down, Node}}, State) ->
clean_table_from_bad_node(Node),
{noreply, State};
handle_info({route, From, To, Packet}, State) ->
case catch do_route(From, To, Packet) of
{'EXIT', Reason} ->
@ -221,6 +279,7 @@ handle_info(_Info, State) ->
%% The return value is ignored.
%%--------------------------------------------------------------------
terminate(_Reason, _State) ->
ejabberd_hooks:delete(node_hash_update, ?MODULE, migrate, 100),
ejabberd_commands:unregister_commands(commands()),
ok.
@ -234,22 +293,18 @@ code_change(_OldVsn, State, _Extra) ->
%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------
clean_table_from_bad_node(Node) ->
F = fun() ->
Es = mnesia:select(
s2s,
[{#s2s{pid = '$1', _ = '_'},
[{'==', {node, '$1'}, Node}],
['$_']}]),
lists:foreach(fun(E) ->
mnesia:delete_object(E)
end, Es)
end,
mnesia:async_dirty(F).
do_route(From, To, Packet) ->
?DEBUG("s2s manager~n\tfrom ~p~n\tto ~p~n\tpacket ~P~n",
[From, To, Packet, 8]),
FromTo = {From#jid.lserver, To#jid.lserver},
case ejabberd_cluster:get_node(FromTo) of
Node when Node == node() ->
do_route1(From, To, Packet);
Node ->
{?MODULE, Node} ! {route, From, To, Packet}
end.
do_route1(From, To, Packet) ->
case find_connection(From, To) of
{atomic, Pid} when is_pid(Pid) ->
?DEBUG("sending to process ~p~n", [Pid]),
@ -482,6 +537,12 @@ update_tables() ->
mnesia:delete_table(local_s2s);
false ->
ok
end,
case catch mnesia:table_info(s2s, local_content) of
false ->
mnesia:delete_table(s2s);
_ ->
ok
end.
%% Check if host is in blacklist or white list

10
src/ejabberd_s2s_in.erl
View File

@ -48,6 +48,11 @@
-include("ejabberd.hrl").
-include("jlib.hrl").
-ifdef(SSL40).
-include_lib("public_key/include/public_key.hrl").
-define(PKIXEXPLICIT, 'OTP-PUB-KEY').
-define(PKIXIMPLICIT, 'OTP-PUB-KEY').
-else.
-ifdef(SSL39).
-include_lib("ssl/include/ssl_pkix.hrl").
-define(PKIXEXPLICIT, 'OTP-PKIX').
@ -58,6 +63,7 @@
-define(PKIXEXPLICIT, 'PKIX1Explicit88').
-define(PKIXIMPLICIT, 'PKIX1Implicit88').
-endif.
-endif.
-include("XmppAddr.hrl").
-define(DICT, dict).
@ -263,7 +269,7 @@ wait_for_feature_request({xmlstreamelement, El}, StateData) ->
TLSOpts = StateData#state.tls_options,
TLSSocket = (StateData#state.sockmod):starttls(
Socket, TLSOpts,
xml:element_to_string(
xml:element_to_binary(
{xmlelement, "proceed", [{"xmlns", ?NS_TLS}], []})),
{next_state, wait_for_stream,
StateData#state{socket = TLSSocket,
@ -618,7 +624,7 @@ send_text(StateData, Text) ->
(StateData#state.sockmod):send(StateData#state.socket, Text).
send_element(StateData, El) ->
send_text(StateData, xml:element_to_string(El)).
send_text(StateData, xml:element_to_binary(El)).
change_shaper(StateData, Host, JID) ->

25
src/ejabberd_s2s_out.erl
View File

@ -34,7 +34,8 @@
start_link/3,
start_connection/1,
terminate_if_waiting_delay/2,
stop_connection/1]).
stop_connection/1,
stop_connection/2]).
%% p1_fsm callbacks (same as gen_fsm)
-export([init/1,
@ -54,6 +55,7 @@
print_state/1,
code_change/4,
test_get_addr_port/1,
print_state/1,
get_addr_port/1]).
-include("ejabberd.hrl").
@ -84,10 +86,11 @@
%% Module start with or without supervisor:
-ifdef(NO_TRANSIENT_SUPERVISORS).
-define(SUPERVISOR_START, p1_fsm:start(ejabberd_s2s_out, [From, Host, Type],
fsm_limit_opts() ++ ?FSMOPTS)).
-define(SUPERVISOR_START, rpc:call(Node, p1_fsm, start,
[ejabberd_s2s_out, [From, Host, Type],
fsm_limit_opts() ++ ?FSMOPTS])).
-else.
-define(SUPERVISOR_START, supervisor:start_child(ejabberd_s2s_out_sup,
-define(SUPERVISOR_START, supervisor:start_child({ejabberd_s2s_out_sup, Node},
[From, Host, Type])).
-endif.
@ -126,6 +129,7 @@
%%% API
%%%----------------------------------------------------------------------
start(From, Host, Type) ->
Node = ejabberd_cluster:get_node({From, Host}),
?SUPERVISOR_START.
start_link(From, Host, Type) ->
@ -136,7 +140,10 @@ start_connection(Pid) ->
p1_fsm:send_event(Pid, init).
stop_connection(Pid) ->
p1_fsm:send_event(Pid, stop).
p1_fsm:send_event(Pid, closed).
stop_connection(Pid, Timeout) ->
p1_fsm:send_all_state_event(Pid, {closed, Timeout}).
%%%----------------------------------------------------------------------
%%% Callback functions from p1_fsm
@ -746,9 +753,15 @@ stream_established(closed, StateData) ->
%% {next_state, NextStateName, NextStateData, Timeout} |
%% {stop, Reason, NewStateData}
%%----------------------------------------------------------------------
handle_event({closed, Timeout}, StateName, StateData) ->
p1_fsm:send_event_after(Timeout, closed),
{next_state, StateName, StateData};
handle_event(_Event, StateName, StateData) ->
{next_state, StateName, StateData, get_timeout_interval(StateName)}.
print_state(StateData) ->
StateData.
%%----------------------------------------------------------------------
%% Func: handle_sync_event/4
%% Returns: The associated StateData for this connection
@ -892,7 +905,7 @@ send_text(StateData, Text) ->
ejabberd_socket:send(StateData#state.socket, Text).
send_element(StateData, El) ->
send_text(StateData, xml:element_to_string(El)).
send_text(StateData, xml:element_to_binary(El)).
send_queue(StateData, Q) ->
case queue:out(Q) of

4
src/ejabberd_service.erl
View File

@ -347,7 +347,7 @@ handle_info({route, From, To, Packet}, StateName, StateData) ->
Attrs2 = jlib:replace_from_to_attrs(jlib:jid_to_string(From),
jlib:jid_to_string(To),
Attrs),
Text = xml:element_to_string({xmlelement, Name, Attrs2, Els}),
Text = xml:element_to_binary({xmlelement, Name, Attrs2, Els}),
send_text(StateData, Text);
deny ->
Err = jlib:make_error_reply(Packet, ?ERR_NOT_ALLOWED),
@ -391,7 +391,7 @@ send_text(StateData, Text) ->
(StateData#state.sockmod):send(StateData#state.socket, Text).
send_element(StateData, El) ->
send_text(StateData, xml:element_to_string(El)).
send_text(StateData, xml:element_to_binary(El)).
new_id() ->
randoms:get_string().

204
src/ejabberd_sm.erl
View File

@ -32,7 +32,10 @@
%% API
-export([start_link/0,
route/3,
open_session/5, close_session/4,
set_session/6,
open_session/5,
open_session/6,
close_session/4,
check_in_subscription/6,
bounce_offline_message/3,
disconnect_removed_user/2,
@ -43,6 +46,7 @@
dirty_get_sessions_list/0,
dirty_get_my_sessions_list/0,
get_vh_session_list/1,
get_vh_my_session_list/1,
get_vh_session_number/1,
register_iq_handler/4,
register_iq_handler/5,
@ -53,7 +57,8 @@
user_resources/2,
get_session_pid/3,
get_user_info/3,
get_user_ip/3
get_user_ip/3,
migrate/1
]).
%% gen_server callbacks
@ -66,7 +71,6 @@
-include("mod_privacy.hrl").
-record(session, {sid, usr, us, priority, info}).
-record(session_counter, {vhost, count}).
-record(state, {}).
%% default value for the maximum number of user connections
@ -92,9 +96,10 @@ route(From, To, Packet) ->
end.
open_session(SID, User, Server, Resource, Info) ->
set_session(SID, User, Server, Resource, undefined, Info),
mnesia:dirty_update_counter(session_counter,
jlib:nameprep(Server), 1),
open_session(SID, User, Server, Resource, undefined, Info).
open_session(SID, User, Server, Resource, Priority, Info) ->
set_session(SID, User, Server, Resource, Priority, Info),
check_for_sessions_to_replace(User, Server, Resource),
JID = jlib:make_jid(User, Server, Resource),
ejabberd_hooks:run(sm_register_connection_hook, JID#jid.lserver,
@ -106,9 +111,7 @@ close_session(SID, User, Server, Resource) ->
[#session{info=I}] -> I
end,
F = fun() ->
mnesia:delete({session, SID}),
mnesia:dirty_update_counter(session_counter,
jlib:nameprep(Server), -1)
mnesia:delete({session, SID})
end,
mnesia:sync_dirty(F),
JID = jlib:make_jid(User, Server, Resource),
@ -138,11 +141,17 @@ get_user_resources(User, Server) ->
LUser = jlib:nodeprep(User),
LServer = jlib:nameprep(Server),
US = {LUser, LServer},
case catch mnesia:dirty_index_read(session, US, #session.us) of
{'EXIT', _Reason} ->
[];
Ss ->
[element(3, S#session.usr) || S <- clean_session_list(Ss)]
Ss = case ejabberd_cluster:get_node({LUser, LServer}) of
Node when Node == node() ->
catch mnesia:dirty_index_read(session, US, #session.us);
Node ->
catch rpc:call(Node, mnesia, dirty_index_read,
[session, US, #session.us], 5000)
end,
if is_list(Ss) ->
[element(3, S#session.usr) || S <- clean_session_list(Ss)];
true ->
[]
end.
get_user_ip(User, Server, Resource) ->
@ -150,12 +159,18 @@ get_user_ip(User, Server, Resource) ->
LServer = jlib:nameprep(Server),
LResource = jlib:resourceprep(Resource),
USR = {LUser, LServer, LResource},
case mnesia:dirty_index_read(session, USR, #session.usr) of
[] ->
undefined;
Ss ->
Ss = case ejabberd_cluster:get_node({LUser, LServer}) of
Node when Node == node() ->
mnesia:dirty_index_read(session, USR, #session.usr);
Node ->
catch rpc:call(Node, mnesia, dirty_index_read,
[session, USR, #session.usr], 5000)
end,
if is_list(Ss), Ss /= [] ->
Session = lists:max(Ss),
proplists:get_value(ip, Session#session.info)
proplists:get_value(ip, Session#session.info);
true ->
undefined
end.
get_user_info(User, Server, Resource) ->
@ -163,15 +178,21 @@ get_user_info(User, Server, Resource) ->
LServer = jlib:nameprep(Server),
LResource = jlib:resourceprep(Resource),
USR = {LUser, LServer, LResource},
case mnesia:dirty_index_read(session, USR, #session.usr) of
[] ->
offline;
Ss ->
Ss = case ejabberd_cluster:get_node({LUser, LServer}) of
Node when Node == node() ->
mnesia:dirty_index_read(session, USR, #session.usr);
Node ->
catch rpc:call(Node, mnesia, dirty_index_read,
[session, USR, #session.usr], 5000)
end,
if is_list(Ss), Ss /= [] ->
Session = lists:max(Ss),
Node = node(element(2, Session#session.sid)),
N = node(element(2, Session#session.sid)),
Conn = proplists:get_value(conn, Session#session.info),
IP = proplists:get_value(ip, Session#session.info),
[{node, Node}, {conn, Conn}, {ip, IP}]
[{node, N}, {conn, Conn}, {ip, IP}];
true ->
offline
end.
set_presence(SID, User, Server, Resource, Priority, Presence, Info) ->
@ -194,26 +215,37 @@ get_session_pid(User, Server, Resource) ->
LServer = jlib:nameprep(Server),
LResource = jlib:resourceprep(Resource),
USR = {LUser, LServer, LResource},
case catch mnesia:dirty_index_read(session, USR, #session.usr) of
Res = case ejabberd_cluster:get_node({LUser, LServer}) of
Node when Node == node() ->
mnesia:dirty_index_read(session, USR, #session.usr);
Node ->
catch rpc:call(Node, mnesia, dirty_index_read,
[session, USR, #session.usr], 5000)
end,
case Res of
[#session{sid = {_, Pid}}] -> Pid;
_ -> none
end.
dirty_get_sessions_list() ->
mnesia:dirty_select(
session,
[{#session{usr = '$1', _ = '_'},
[],
['$1']}]).
Match = [{#session{usr = '$1', _ = '_'}, [], ['$1']}],
lists:flatmap(
fun(Node) when Node == node() ->
mnesia:dirty_select(session, Match);
(Node) ->
case catch rpc:call(Node, mnesia, dirty_select,
[session, Match], 5000) of
Ss when is_list(Ss) ->
Ss;
_ ->
[]
end
end, ejabberd_cluster:get_nodes()).
dirty_get_my_sessions_list() ->
mnesia:dirty_select(
session,
[{#session{sid = {'_', '$1'}, _ = '_'},
[{'==', {node, '$1'}, node()}],
['$_']}]).
mnesia:dirty_match_object(#session{_ = '_'}).
get_vh_session_list(Server) ->
get_vh_my_session_list(Server) ->
LServer = jlib:nameprep(Server),
mnesia:dirty_select(
session,
@ -221,19 +253,24 @@ get_vh_session_list(Server) ->
[{'==', {element, 2, '$1'}, LServer}],
['$1']}]).
get_vh_session_list(Server) ->
lists:flatmap(
fun(Node) when Node == node() ->
get_vh_my_session_list(Server);
(Node) ->
case catch rpc:call(Node, ?MODULE, get_vh_my_session_list,
[Server], 5000) of
Ss when is_list(Ss) ->
Ss;
_ ->
[]
end
end, ejabberd_cluster:get_nodes()).
get_vh_session_number(Server) ->
LServer = jlib:nameprep(Server),
Query = mnesia:dirty_select(
session_counter,
[{#session_counter{vhost = LServer, count = '$1'},
[],
['$1']}]),
case Query of
[Count] ->
Count;
_ -> 0
end.
%% TODO
length(get_vh_session_list(Server)).
register_iq_handler(Host, XMLNS, Module, Fun) ->
ejabberd_sm ! {register_iq_handler, Host, XMLNS, Module, Fun}.
@ -243,6 +280,21 @@ register_iq_handler(Host, XMLNS, Module, Fun, Opts) ->
unregister_iq_handler(Host, XMLNS) ->
ejabberd_sm ! {unregister_iq_handler, Host, XMLNS}.
migrate(After) ->
Ss = mnesia:dirty_select(
session,
[{#session{us = '$1', sid = {'_', '$2'}, _ = '_'},
[],
['$$']}]),
lists:foreach(
fun([US, Pid]) ->
case ejabberd_cluster:get_node_new(US) of
Node when Node /= node() ->
ejabberd_c2s:migrate(Pid, Node, After);
_ ->
ok
end
end, Ss).
%%====================================================================
%% gen_server callbacks
@ -259,16 +311,13 @@ init([]) ->
update_tables(),
mnesia:create_table(session,
[{ram_copies, [node()]},
{local_content, true},
{attributes, record_info(fields, session)}]),
mnesia:create_table(session_counter,
[{ram_copies, [node()]},
{attributes, record_info(fields, session_counter)}]),
mnesia:add_table_index(session, usr),
mnesia:add_table_index(session, us),
mnesia:add_table_copy(session, node(), ram_copies),
mnesia:add_table_copy(session_counter, node(), ram_copies),
mnesia:subscribe(system),
ets:new(sm_iqtable, [named_table]),
ejabberd_hooks:add(node_hash_update, ?MODULE, migrate, 100),
lists:foreach(
fun(Host) ->
ejabberd_hooks:add(roster_in_subscription, Host,
@ -319,9 +368,6 @@ handle_info({route, From, To, Packet}, State) ->
ok
end,
{noreply, State};
handle_info({mnesia_system_event, {mnesia_down, Node}}, State) ->
recount_session_table(Node),
{noreply, State};
handle_info({register_iq_handler, Host, XMLNS, Module, Function}, State) ->
ets:insert(sm_iqtable, {{XMLNS, Host}, Module, Function}),
{noreply, State};
@ -348,6 +394,7 @@ handle_info(_Info, State) ->
%% The return value is ignored.
%%--------------------------------------------------------------------
terminate(_Reason, _State) ->
ejabberd_hooks:delete(node_hash_update, ?MODULE, migrate, 100),
ejabberd_commands:unregister_commands(commands()),
ok.
@ -377,38 +424,20 @@ set_session(SID, User, Server, Resource, Priority, Info) ->
end,
mnesia:sync_dirty(F).
%% Recalculates alive sessions when Node goes down
%% and updates session and session_counter tables
recount_session_table(Node) ->
F = fun() ->
Es = mnesia:select(
session,
[{#session{sid = {'_', '$1'}, _ = '_'},
[{'==', {node, '$1'}, Node}],
['$_']}]),
lists:foreach(fun(E) ->
mnesia:delete({session, E#session.sid})
end, Es),
%% reset session_counter table with active sessions
mnesia:clear_table(session_counter),
lists:foreach(fun(Server) ->
LServer = jlib:nameprep(Server),
Hs = mnesia:select(session,
[{#session{usr = '$1', _ = '_'},
[{'==', {element, 2, '$1'}, LServer}],
['$1']}]),
mnesia:write(
#session_counter{vhost = LServer,
count = length(Hs)})
end, ?MYHOSTS)
end,
mnesia:async_dirty(F).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
do_route(From, To, Packet) ->
?DEBUG("session manager~n\tfrom ~p~n\tto ~p~n\tpacket ~P~n",
[From, To, Packet, 8]),
{U, S, _} = jlib:jid_tolower(To),
case ejabberd_cluster:get_node({U, S}) of
Node when Node /= node() ->
{?MODULE, Node} ! {route, From, To, Packet};
_ ->
do_route1(From, To, Packet)
end.
do_route1(From, To, Packet) ->
#jid{user = User, server = Server,
luser = LUser, lserver = LServer, lresource = LResource} = To,
{xmlelement, Name, Attrs, _Els} = Packet,
@ -795,4 +824,11 @@ update_tables() ->
mnesia:delete_table(local_session);
false ->
ok
end,
mnesia:delete_table(session_counter),
case catch mnesia:table_info(session, local_content) of
false ->
mnesia:delete_table(session);
_ ->
ok
end.

60
src/ejabberd_socket.erl
View File

@ -44,6 +44,7 @@
get_peer_certificate/1,
get_verify_result/1,
close/1,
change_controller/2,
sockname/1, peername/1]).
-include("ejabberd.hrl").
@ -129,29 +130,19 @@ connect(Addr, Port, Opts, Timeout) ->
end.
starttls(SocketData, TLSOpts) ->
{ok, TLSSocket} = tls:tcp_to_tls(SocketData#socket_state.socket, TLSOpts),
ejabberd_receiver:starttls(SocketData#socket_state.receiver, TLSSocket),
SocketData#socket_state{socket = TLSSocket, sockmod = tls}.
starttls(SocketData, TLSOpts, undefined).
starttls(SocketData, TLSOpts, Data) ->
{ok, TLSSocket} = tls:tcp_to_tls(SocketData#socket_state.socket, TLSOpts),
ejabberd_receiver:starttls(SocketData#socket_state.receiver, TLSSocket),
send(SocketData, Data),
{ok, TLSSocket} = ejabberd_receiver:starttls(
SocketData#socket_state.receiver, TLSOpts, Data),
SocketData#socket_state{socket = TLSSocket, sockmod = tls}.
compress(SocketData) ->
{ok, ZlibSocket} = ejabberd_zlib:enable_zlib(
SocketData#socket_state.sockmod,
SocketData#socket_state.socket),
ejabberd_receiver:compress(SocketData#socket_state.receiver, ZlibSocket),
SocketData#socket_state{socket = ZlibSocket, sockmod = ejabberd_zlib}.
compress(SocketData, undefined).
compress(SocketData, Data) ->
{ok, ZlibSocket} = ejabberd_zlib:enable_zlib(
SocketData#socket_state.sockmod,
SocketData#socket_state.socket),
ejabberd_receiver:compress(SocketData#socket_state.receiver, ZlibSocket),
send(SocketData, Data),
{ok, ZlibSocket} = ejabberd_receiver:compress(
SocketData#socket_state.receiver, Data),
SocketData#socket_state{socket = ZlibSocket, sockmod = ejabberd_zlib}.
reset_stream(SocketData) when is_pid(SocketData#socket_state.receiver) ->
@ -160,10 +151,25 @@ reset_stream(SocketData) when is_atom(SocketData#socket_state.receiver) ->
(SocketData#socket_state.receiver):reset_stream(
SocketData#socket_state.socket).
change_controller(#socket_state{receiver = Recv}, Pid) when is_pid(Recv) ->
ejabberd_receiver:setopts(Recv, [{active, false}]),
sync_events(Pid),
ejabberd_receiver:change_controller(Recv, Pid);
change_controller(#socket_state{socket = Socket, receiver = Mod}, Pid) ->
Mod:setopts(Socket, [{active, false}]),
sync_events(Pid),
Mod:change_controller(Socket, Pid).
%% sockmod=gen_tcp|tls|ejabberd_zlib
send(SocketData, Data) ->
case catch (SocketData#socket_state.sockmod):send(
SocketData#socket_state.socket, Data) of
Res = if node(SocketData#socket_state.receiver) == node() ->
catch (SocketData#socket_state.sockmod):send(
SocketData#socket_state.socket, Data);
true ->
catch ejabberd_receiver:send(
SocketData#socket_state.receiver, Data)
end,
case Res of
ok -> ok;
{error, timeout} ->
?INFO_MSG("Timeout on ~p:send",[SocketData#socket_state.sockmod]),
@ -225,3 +231,21 @@ peername(#socket_state{sockmod = SockMod, socket = Socket}) ->
%%====================================================================
%% Internal functions
%%====================================================================
%% dirty hack to relay queued messages from
%% old owner to new owner. The idea is based
%% on code of gen_tcp:controlling_process/2.
sync_events(C2SPid) ->
receive
{'$gen_event', El} = Event when element(1, El) == xmlelement;
element(1, El) == xmlstreamstart;
element(1, El) == xmlstreamelement;
element(1, El) == xmlstreamend;
element(1, El) == xmlstreamerror ->
C2SPid ! Event,
sync_events(C2SPid);
closed ->
C2SPid ! closed,
sync_events(C2SPid)
after 0 ->
ok
end.

16
src/ejabberd_sup.erl
View File

@ -42,13 +42,6 @@ init([]) ->
brutal_kill,
worker,
[ejabberd_hooks]},
NodeGroups =
{ejabberd_node_groups,
{ejabberd_node_groups, start_link, []},
permanent,
brutal_kill,
worker,
[ejabberd_node_groups]},
SystemMonitor =
{ejabberd_system_monitor,
{ejabberd_system_monitor, start_link, []},
@ -177,9 +170,16 @@ init([]) ->
infinity,
supervisor,
[ejabberd_tmp_sup]},
Cluster =
{ejabberd_cluster,
{ejabberd_cluster, start_link, []},
permanent,
brutal_kill,
worker,
[ejabberd_cluster]},
{ok, {{one_for_one, 10, 1},
[Hooks,
NodeGroups,
Cluster,
SystemMonitor,
Router,
SM,

8
src/ejabberdctl.template
View File

@ -82,7 +82,7 @@ else
KERNEL_OPTS="-kernel inet_dist_listen_min ${FIREWALL_WINDOW%-*} inet_dist_listen_max ${FIREWALL_WINDOW#*-}"
fi
ERLANG_OPTS="+K $POLL -smp $SMP +P $ERL_PROCESSES $ERL_OPTIONS $KERNEL_OPTS"
ERLANG_OPTS="+K $POLL -smp $SMP +P $ERL_PROCESSES $ERL_OPTIONS"
# define additional environment variables
if [ "$EJABBERDDIR" = "" ]; then
@ -141,6 +141,7 @@ start ()
-noinput -detached \
-pa $EJABBERD_EBIN_PATH \
-mnesia dir \"\\\"$SPOOLDIR\\\"\" \
$KERNEL_OPTS \
-s ejabberd \
-sasl sasl_error_logger \\{file,\\\"$SASL_LOG_PATH\\\"\\} \
$ERLANG_OPTS $ARGS \"$@\""
@ -174,6 +175,7 @@ debug ()
$NAME debug-${TTY}-${ERLANG_NODE} \
-remsh $ERLANG_NODE \
-hidden \
$KERNEL_OPTS \
$ERLANG_OPTS $ARGS \"$@\""
}
@ -203,6 +205,7 @@ live ()
$NAME $ERLANG_NODE \
-pa $EJABBERD_EBIN_PATH \
-mnesia dir \"\\\"$SPOOLDIR\\\"\" \
$KERNEL_OPTS \
-s ejabberd \
$ERLANG_OPTS $ARGS \"$@\""
}
@ -236,7 +239,7 @@ ctl ()
MAXCONNID=100
CONNLOCKDIR=@LOCALSTATEDIR@/lock/ejabberdctl
FLOCK='/usr/bin/flock'
if [ ! -x "$FLOCK" ] ; then
if [ ! -x "$FLOCK" ] || [ ! -d "$CONNLOCKDIR" ] ; then
JOT='/usr/bin/jot'
if [ ! -x "$JOT" ] ; then
# no flock or jot, simply invoke ctlexec()
@ -310,6 +313,7 @@ ctlexec ()
-noinput \
-hidden \
-pa $EJABBERD_EBIN_PATH \
$KERNEL_OPTS \
-s ejabberd_ctl -extra $ERLANG_NODE $COMMAND"
}

2
src/mod_last.erl
View File

@ -127,7 +127,7 @@ process_sm_iq(From, To, #iq{type = Type, sub_el = SubEl} = IQ) ->
[User, Server, UserListRecord,
{From, To,
{xmlelement, "presence", [], []}},
out]) of
in]) of
allow ->
get_last(IQ, SubEl, User, Server);
deny ->

2
src/mod_last_odbc.erl
View File

@ -119,7 +119,7 @@ process_sm_iq(From, To, #iq{type = Type, sub_el = SubEl} = IQ) ->
[User, Server, UserListRecord,
{From, To,
{xmlelement, "presence", [], []}},
out]) of
in]) of
allow ->
get_last(IQ, SubEl, User, Server);
deny ->

180
src/mod_muc/mod_muc.erl
View File

@ -41,6 +41,9 @@
create_room/5,
process_iq_disco_items/4,
broadcast_service_message/2,
register_room/3,
migrate/1,
get_vh_rooms/1,
can_use_nick/3]).
%% gen_server callbacks
@ -109,7 +112,9 @@ room_destroyed(Host, Room, Pid, ServerHost) ->
%% Else use the passed options as defined in mod_muc_room.
create_room(Host, Name, From, Nick, Opts) ->
Proc = gen_mod:get_module_proc(Host, ?PROCNAME),
gen_server:call(Proc, {create, Name, From, Nick, Opts}).
RoomHost = gen_mod:get_module_opt_host(Host, ?MODULE, "conference.@HOST@"),
Node = ejabberd_cluster:get_node({Name, RoomHost}),
gen_server:call({Proc, Node}, {create, Name, From, Nick, Opts}).
store_room(Host, Name, Opts) ->
F = fun() ->
@ -162,6 +167,22 @@ can_use_nick(Host, JID, Nick) ->
U == LUS
end.
migrate(After) ->
Rs = mnesia:dirty_select(
muc_online_room,
[{#muc_online_room{name_host = '$1', pid = '$2', _ = '_'},
[],
['$$']}]),
lists:foreach(
fun([NameHost, Pid]) ->
case ejabberd_cluster:get_node_new(NameHost) of
Node when Node /= node() ->
mod_muc_room:migrate(Pid, Node, After);
_ ->
ok
end
end, Rs).
%%====================================================================
%% gen_server callbacks
%%====================================================================
@ -174,6 +195,7 @@ can_use_nick(Host, JID, Nick) ->
%% Description: Initiates the server
%%--------------------------------------------------------------------
init([Host, Opts]) ->
update_muc_online_table(),
mnesia:create_table(muc_room,
[{disc_copies, [node()]},
{attributes, record_info(fields, muc_room)}]),
@ -182,14 +204,13 @@ init([Host, Opts]) ->
{attributes, record_info(fields, muc_registered)}]),
mnesia:create_table(muc_online_room,
[{ram_copies, [node()]},
{local_content, true},
{attributes, record_info(fields, muc_online_room)}]),
mnesia:add_table_copy(muc_online_room, node(), ram_copies),
catch ets:new(muc_online_users, [bag, named_table, public, {keypos, 2}]),
MyHost = gen_mod:get_opt_host(Host, Opts, "conference.@HOST@"),
update_tables(MyHost),
clean_table_from_bad_node(node(), MyHost),
mnesia:add_table_index(muc_registered, nick),
mnesia:subscribe(system),
Access = gen_mod:get_opt(access, Opts, all),
AccessCreate = gen_mod:get_opt(access_create, Opts, all),
AccessAdmin = gen_mod:get_opt(access_admin, Opts, none),
@ -198,6 +219,7 @@ init([Host, Opts]) ->
DefRoomOpts = gen_mod:get_opt(default_room_options, Opts, []),
RoomShaper = gen_mod:get_opt(room_shaper, Opts, none),
ejabberd_router:register_route(MyHost),
ejabberd_hooks:add(node_hash_update, ?MODULE, migrate, 100),
load_permanent_rooms(MyHost, Host,
{Access, AccessCreate, AccessAdmin, AccessPersistent},
HistorySize,
@ -264,12 +286,19 @@ handle_info({route, From, To, Packet},
default_room_opts = DefRoomOpts,
history_size = HistorySize,
room_shaper = RoomShaper} = State) ->
case catch do_route(Host, ServerHost, Access, HistorySize, RoomShaper,
From, To, Packet, DefRoomOpts) of
{'EXIT', Reason} ->
?ERROR_MSG("~p", [Reason]);
_ ->
ok
{U, S, _} = jlib:jid_tolower(To),
case ejabberd_cluster:get_node({U, S}) of
Node when Node == node() ->
case catch do_route(Host, ServerHost, Access, HistorySize,
RoomShaper, From, To, Packet, DefRoomOpts) of
{'EXIT', Reason} ->
?ERROR_MSG("~p", [Reason]);
_ ->
ok
end;
Node ->
Proc = gen_mod:get_module_proc(ServerHost, ?PROCNAME),
{Proc, Node} ! {route, From, To, Packet}
end,
{noreply, State};
handle_info({room_destroyed, RoomHost, Pid}, State) ->
@ -277,10 +306,7 @@ handle_info({room_destroyed, RoomHost, Pid}, State) ->
mnesia:delete_object(#muc_online_room{name_host = RoomHost,
pid = Pid})
end,
mnesia:transaction(F),
{noreply, State};
handle_info({mnesia_system_event, {mnesia_down, Node}}, State) ->
clean_table_from_bad_node(Node),
mnesia:async_dirty(F),
{noreply, State};
handle_info(_Info, State) ->
{noreply, State}.
@ -293,6 +319,7 @@ handle_info(_Info, State) ->
%% The return value is ignored.
%%--------------------------------------------------------------------
terminate(_Reason, State) ->
ejabberd_hooks:delete(node_hash_update, ?MODULE, migrate, 100),
ejabberd_router:unregister_route(State#state.host),
ok.
@ -532,17 +559,22 @@ load_permanent_rooms(Host, ServerHost, Access, HistorySize, RoomShaper) ->
lists:foreach(
fun(R) ->
{Room, Host} = R#muc_room.name_host,
case mnesia:dirty_read(muc_online_room, {Room, Host}) of
[] ->
{ok, Pid} = mod_muc_room:start(
Host,
ServerHost,
Access,
Room,
HistorySize,
RoomShaper,
R#muc_room.opts),
register_room(Host, Room, Pid);
case ejabberd_cluster:get_node({Room, Host}) of
Node when Node == node() ->
case mnesia:dirty_read(muc_online_room, {Room, Host}) of
[] ->
{ok, Pid} = mod_muc_room:start(
Host,
ServerHost,
Access,
Room,
HistorySize,
RoomShaper,
R#muc_room.opts),
register_room(Host, Room, Pid);
_ ->
ok
end;
_ ->
ok
end
@ -568,11 +600,10 @@ start_new_room(Host, ServerHost, Access, Room,
register_room(Host, Room, Pid) ->
F = fun() ->
mnesia:write(#muc_online_room{name_host = {Room, Host},
pid = Pid})
end,
mnesia:transaction(F).
mnesia:write(#muc_online_room{name_host = {Room, Host},
pid = Pid})
end,
mnesia:async_dirty(F).
iq_disco_info(Lang) ->
[{xmlelement, "identity",
@ -601,7 +632,7 @@ iq_disco_items(Host, From, Lang, none) ->
_ ->
false
end
end, get_vh_rooms(Host));
end, get_vh_rooms_all_nodes(Host));
iq_disco_items(Host, From, Lang, Rsm) ->
{Rooms, RsmO} = get_vh_rooms(Host, Rsm),
@ -621,19 +652,9 @@ iq_disco_items(Host, From, Lang, Rsm) ->
end, Rooms) ++ RsmOut.
get_vh_rooms(Host, #rsm_in{max=M, direction=Direction, id=I, index=Index})->
AllRooms = lists:sort(get_vh_rooms(Host)),
AllRooms = get_vh_rooms_all_nodes(Host),
Count = erlang:length(AllRooms),
Guard = case Direction of
_ when Index =/= undefined -> [{'==', {element, 2, '$1'}, Host}];
aft -> [{'==', {element, 2, '$1'}, Host}, {'>=',{element, 1, '$1'} ,I}];
before when I =/= []-> [{'==', {element, 2, '$1'}, Host}, {'=<',{element, 1, '$1'} ,I}];
_ -> [{'==', {element, 2, '$1'}, Host}]
end,
L = lists:sort(
mnesia:dirty_select(muc_online_room,
[{#muc_online_room{name_host = '$1', _ = '_'},
Guard,
['$_']}])),
L = get_vh_rooms_direction(Direction, I, Index, AllRooms),
L2 = if
Index == undefined andalso Direction == before ->
lists:reverse(lists:sublist(lists:reverse(L), 1, M));
@ -656,6 +677,27 @@ get_vh_rooms(Host, #rsm_in{max=M, direction=Direction, id=I, index=Index})->
{L2, #rsm_out{first=F, last=Last, count=Count, index=NewIndex}}
end.
get_vh_rooms_direction(_Direction, _I, Index, AllRooms) when Index =/= undefined ->
AllRooms;
get_vh_rooms_direction(aft, I, _Index, AllRooms) ->
{_Before, After} =
lists:splitwith(
fun(#muc_online_room{name_host = {Na, _}}) ->
Na < I end, AllRooms),
case After of
[] -> [];
[#muc_online_room{name_host = {I, _Host}} | AfterTail] -> AfterTail;
_ -> After
end;
get_vh_rooms_direction(before, I, _Index, AllRooms) when I =/= []->
{Before, _} =
lists:splitwith(
fun(#muc_online_room{name_host = {Na, _}}) ->
Na < I end, AllRooms),
Before;
get_vh_rooms_direction(_Direction, _I, _Index, AllRooms) ->
AllRooms.
%% @doc Return the position of desired room in the list of rooms.
%% The room must exist in the list. The count starts in 0.
%% @spec (Desired::muc_online_room(), Rooms::[muc_online_room()]) -> integer()
@ -813,7 +855,22 @@ broadcast_service_message(Host, Msg) ->
fun(#muc_online_room{pid = Pid}) ->
gen_fsm:send_all_state_event(
Pid, {service_message, Msg})
end, get_vh_rooms(Host)).
end, get_vh_rooms_all_nodes(Host)).
get_vh_rooms_all_nodes(Host) ->
Rooms = lists:foldl(
fun(Node, Acc) when Node == node() ->
get_vh_rooms(Host) ++ Acc;
(Node, Acc) ->
case catch rpc:call(Node, ?MODULE, get_vh_rooms,
[Host], 5000) of
Res when is_list(Res) ->
Res ++ Acc;
_ ->
Acc
end
end, [], ejabberd_cluster:get_nodes()),
lists:ukeysort(#muc_online_room.name_host, Rooms).
get_vh_rooms(Host) ->
mnesia:dirty_select(muc_online_room,
@ -821,39 +878,18 @@ get_vh_rooms(Host) ->
[{'==', {element, 2, '$1'}, Host}],
['$_']}]).
clean_table_from_bad_node(Node) ->
F = fun() ->
Es = mnesia:select(
muc_online_room,
[{#muc_online_room{pid = '$1', _ = '_'},
[{'==', {node, '$1'}, Node}],
['$_']}]),
lists:foreach(fun(E) ->
mnesia:delete_object(E)
end, Es)
end,
mnesia:async_dirty(F).
clean_table_from_bad_node(Node, Host) ->
F = fun() ->
Es = mnesia:select(
muc_online_room,
[{#muc_online_room{pid = '$1',
name_host = {'_', Host},
_ = '_'},
[{'==', {node, '$1'}, Node}],
['$_']}]),
lists:foreach(fun(E) ->
mnesia:delete_object(E)
end, Es)
end,
mnesia:async_dirty(F).
update_tables(Host) ->
update_muc_room_table(Host),
update_muc_registered_table(Host).
update_muc_online_table() ->
case catch mnesia:table_info(muc_online_room, local_content) of
false ->
mnesia:delete_table(muc_online_room);
_ ->
ok
end.
update_muc_room_table(Host) ->
Fields = record_info(fields, muc_room),
case mnesia:table_info(muc_room, attributes) of

24
src/mod_muc/mod_muc_log.erl
View File

@ -73,11 +73,11 @@
%% Description: Starts the server
%%--------------------------------------------------------------------
start_link(Host, Opts) ->
Proc = gen_mod:get_module_proc(Host, ?PROCNAME),
gen_server:start_link({local, Proc}, ?MODULE, [Host, Opts], []).
Proc = get_proc_name(Host),
gen_server:start_link(Proc, ?MODULE, [Host, Opts], []).
start(Host, Opts) ->
Proc = gen_mod:get_module_proc(Host, ?PROCNAME),
Proc = get_proc_name(Host),
ChildSpec =
{Proc,
{?MODULE, start_link, [Host, Opts]},
@ -88,7 +88,7 @@ start(Host, Opts) ->
supervisor:start_child(ejabberd_sup, ChildSpec).
stop(Host) ->
Proc = gen_mod:get_module_proc(Host, ?PROCNAME),
Proc = get_proc_name(Host),
gen_server:call(Proc, stop),
supervisor:delete_child(ejabberd_sup, Proc).
@ -860,7 +860,9 @@ roomconfig_to_string(Options, Lang, FileFormat) ->
T ->
case Opt of
password -> "<div class=\"rcoe\">" ++ OptText ++ "</div>";
title -> "<div class=\"rcot\">" ++ ?T("Room title") ++ ": \"" ++ htmlize(T, FileFormat) ++ "\"</div>";
max_users -> "<div class=\"rcot\">" ++ OptText ++ ": \"" ++ htmlize(integer_to_list(T), FileFormat) ++ "\"</div>";
title -> "<div class=\"rcot\">" ++ OptText ++ ": \"" ++ htmlize(T, FileFormat) ++ "\"</div>";
description -> "<div class=\"rcot\">" ++ OptText ++ ": \"" ++ htmlize(T, FileFormat) ++ "\"</div>";
_ -> "\"" ++ T ++ "\""
end
end,
@ -876,7 +878,7 @@ get_roomconfig_text(public) -> "Make room public searchable";
get_roomconfig_text(public_list) -> "Make participants list public";
get_roomconfig_text(password_protected) -> "Make room password protected";
get_roomconfig_text(password) -> "Password";
get_roomconfig_text(anonymous) -> "Make room semianonymous";
get_roomconfig_text(anonymous) -> "This room is not anonymous";
get_roomconfig_text(members_only) -> "Make room members-only";
get_roomconfig_text(moderated) -> "Make room moderated";
get_roomconfig_text(members_by_default) -> "Default users as participants";
@ -885,6 +887,13 @@ get_roomconfig_text(allow_private_messages) -> "Allow users to send private mess
get_roomconfig_text(allow_query_users) -> "Allow users to query other users";
get_roomconfig_text(allow_user_invites) -> "Allow users to send invites";
get_roomconfig_text(logging) -> "Enable logging";
get_roomconfig_text(allow_visitor_nickchange) -> "Allow visitors to change nickname";
get_roomconfig_text(allow_visitor_status) -> "Allow visitors to send status text in presence updates";
get_roomconfig_text(captcha_protected) -> "Make room captcha protected";
get_roomconfig_text(description) -> "Room description";
%% get_roomconfig_text(subject) -> "Subject";
%% get_roomconfig_text(subject_author) -> "Subject author";
get_roomconfig_text(max_users) -> "Maximum Number of Occupants";
get_roomconfig_text(_) -> undefined.
%% Users = [{JID, Nick, Role}]
@ -941,7 +950,8 @@ get_room_state(RoomPid) ->
{ok, R} = gen_fsm:sync_send_all_state_event(RoomPid, get_state),
R.
get_proc_name(Host) -> gen_mod:get_module_proc(Host, ?PROCNAME).
get_proc_name(Host) ->
{global, gen_mod:get_module_proc(Host, ?PROCNAME)}.
calc_hour_offset(TimeHere) ->
TimeZero = calendar:now_to_universal_time(now()),

82
src/mod_muc/mod_muc_room.erl
View File

@ -27,14 +27,19 @@
-module(mod_muc_room).
-author('alexey@process-one.net').
-behaviour(gen_fsm).
-define(GEN_FSM, p1_fsm).
-behaviour(?GEN_FSM).
%% External exports
-export([start_link/9,
start_link/7,
start_link/2,
start/9,
start/7,
start/2,
migrate/3,
route/4]).
%% gen_fsm callbacks
@ -44,6 +49,7 @@
handle_sync_event/4,
handle_info/3,
terminate/3,
print_state/1,
code_change/4]).
-include("ejabberd.hrl").
@ -63,16 +69,12 @@
%% Module start with or without supervisor:
-ifdef(NO_TRANSIENT_SUPERVISORS).
-define(SUPERVISOR_START,
gen_fsm:start(?MODULE, [Host, ServerHost, Access, Room, HistorySize,
RoomShaper, Creator, Nick, DefRoomOpts],
?FSMOPTS)).
-define(SUPERVISOR_START(Args),
?GEN_FSM:start(?MODULE, Args, ?FSMOPTS)).
-else.
-define(SUPERVISOR_START,
-define(SUPERVISOR_START(Args),
Supervisor = gen_mod:get_module_proc(ServerHost, ejabberd_mod_muc_sup),
supervisor:start_child(
Supervisor, [Host, ServerHost, Access, Room, HistorySize, RoomShaper,
Creator, Nick, DefRoomOpts])).
supervisor:start_child(Supervisor, Args)).
-endif.
%%%----------------------------------------------------------------------
@ -80,7 +82,8 @@
%%%----------------------------------------------------------------------
start(Host, ServerHost, Access, Room, HistorySize, RoomShaper,
Creator, Nick, DefRoomOpts) ->
?SUPERVISOR_START.
?SUPERVISOR_START([Host, ServerHost, Access, Room, HistorySize,
RoomShaper, Creator, Nick, DefRoomOpts]).
start(Host, ServerHost, Access, Room, HistorySize, RoomShaper, Opts) ->
Supervisor = gen_mod:get_module_proc(ServerHost, ejabberd_mod_muc_sup),
@ -88,16 +91,26 @@ start(Host, ServerHost, Access, Room, HistorySize, RoomShaper, Opts) ->
Supervisor, [Host, ServerHost, Access, Room, HistorySize, RoomShaper,
Opts]).
start(StateName, StateData) ->
ServerHost = StateData#state.server_host,
?SUPERVISOR_START([StateName, StateData]).
start_link(Host, ServerHost, Access, Room, HistorySize, RoomShaper,
Creator, Nick, DefRoomOpts) ->
gen_fsm:start_link(?MODULE, [Host, ServerHost, Access, Room, HistorySize,
RoomShaper, Creator, Nick, DefRoomOpts],
?FSMOPTS).
?GEN_FSM:start_link(?MODULE, [Host, ServerHost, Access, Room, HistorySize,
RoomShaper, Creator, Nick, DefRoomOpts],
?FSMOPTS).
start_link(Host, ServerHost, Access, Room, HistorySize, RoomShaper, Opts) ->
gen_fsm:start_link(?MODULE, [Host, ServerHost, Access, Room, HistorySize,
RoomShaper, Opts],
?FSMOPTS).
?GEN_FSM:start_link(?MODULE, [Host, ServerHost, Access, Room, HistorySize,
RoomShaper, Opts],
?FSMOPTS).
start_link(StateName, StateData) ->
?GEN_FSM:start_link(?MODULE, [StateName, StateData], ?FSMOPTS).
migrate(FsmRef, Node, After) ->
?GEN_FSM:send_all_state_event(FsmRef, {migrate, Node, After}).
%%%----------------------------------------------------------------------
%%% Callback functions from gen_fsm
@ -139,7 +152,11 @@ init([Host, ServerHost, Access, Room, HistorySize, RoomShaper, Opts]) ->
jid = jlib:make_jid(Room, Host, ""),
room_shaper = Shaper}),
add_to_log(room_existence, started, State),
{ok, normal_state, State}.
{ok, normal_state, State};
init([StateName, #state{room = Room, host = Host} = StateData]) ->
process_flag(trap_exit, true),
mod_muc:register_room(Host, Room, self()),
{ok, StateName, StateData}.
%%----------------------------------------------------------------------
%% Func: StateName/2
@ -162,7 +179,7 @@ normal_state({route, From, "",
trunc(gen_mod:get_module_opt(
StateData#state.server_host,
mod_muc, min_message_interval, 0) * 1000000),
Size = iolist_size(xml:element_to_string(Packet)),
Size = element_size(Packet),
{MessageShaper, MessageShaperInterval} =
shaper:update(Activity#activity.message_shaper, Size),
if
@ -580,6 +597,9 @@ handle_event(destroy, StateName, StateData) ->
handle_event({set_affiliations, Affiliations}, StateName, StateData) ->
{next_state, StateName, StateData#state{affiliations = Affiliations}};
handle_event({migrate, Node, After}, StateName, StateData) when Node /= node() ->
{migrate, StateData,
{Node, ?MODULE, start, [StateName, StateData]}, After * 2};
handle_event(_Event, StateName, StateData) ->
{next_state, StateName, StateData}.
@ -612,6 +632,9 @@ handle_sync_event(_Event, _From, StateName, StateData) ->
code_change(_OldVsn, StateName, StateData, _Extra) ->
{ok, StateName, StateData}.
print_state(StateData) ->
StateData.
%%----------------------------------------------------------------------
%% Func: handle_info/3
%% Returns: {next_state, NextStateName, NextStateData} |
@ -701,6 +724,13 @@ handle_info(_Info, StateName, StateData) ->
%% Purpose: Shutdown the fsm
%% Returns: any
%%----------------------------------------------------------------------
terminate({migrated, Clone}, _StateName, StateData) ->
?INFO_MSG("Migrating room ~s@~s to ~p on node ~p",
[StateData#state.room, StateData#state.host,
Clone, node(Clone)]),
mod_muc:room_destroyed(StateData#state.host, StateData#state.room,
self(), StateData#state.server_host),
ok;
terminate(Reason, _StateName, StateData) ->
?INFO_MSG("Stopping MUC room ~s@~s",
[StateData#state.room, StateData#state.host]),
@ -739,7 +769,7 @@ terminate(Reason, _StateName, StateData) ->
%%%----------------------------------------------------------------------
route(Pid, From, ToNick, Packet) ->
gen_fsm:send_event(Pid, {route, From, ToNick, Packet}).
?GEN_FSM:send_event(Pid, {route, From, ToNick, Packet}).
process_groupchat_message(From, {xmlelement, "message", Attrs, _Els} = Packet,
StateData) ->
@ -1394,7 +1424,7 @@ prepare_room_queue(StateData) ->
{{value, {message, From}}, _RoomQueue} ->
Activity = get_user_activity(From, StateData),
Packet = Activity#activity.message,
Size = iolist_size(xml:element_to_string(Packet)),
Size = element_size(Packet),
{RoomShaper, RoomShaperInterval} =
shaper:update(StateData#state.room_shaper, Size),
erlang:send_after(
@ -1405,7 +1435,7 @@ prepare_room_queue(StateData) ->
{{value, {presence, From}}, _RoomQueue} ->
Activity = get_user_activity(From, StateData),
{_Nick, Packet} = Activity#activity.presence,
Size = iolist_size(xml:element_to_string(Packet)),
Size = element_size(Packet),
{RoomShaper, RoomShaperInterval} =
shaper:update(StateData#state.room_shaper, Size),
erlang:send_after(
@ -1625,13 +1655,12 @@ add_new_user(From, Nick, {xmlelement, _, Attrs, Els} = Packet, StateData) ->
From, Err),
StateData;
captcha_required ->
ID = randoms:get_string(),
SID = xml:get_attr_s("id", Attrs),
RoomJID = StateData#state.jid,
To = jlib:jid_replace_resource(RoomJID, Nick),
case ejabberd_captcha:create_captcha(
ID, SID, RoomJID, To, Lang, From) of
{ok, CaptchaEls} ->
SID, RoomJID, To, Lang, From) of
{ok, ID, CaptchaEls} ->
MsgPkt = {xmlelement, "message", [{"id", ID}], CaptchaEls},
Robots = ?DICT:store(From,
{Nick, Packet}, StateData#state.robots),
@ -2068,7 +2097,7 @@ add_message_to_history(FromNick, FromJID, Packet, StateData) ->
jlib:jid_replace_resource(StateData#state.jid, FromNick),
StateData#state.jid,
TSPacket),
Size = iolist_size(xml:element_to_string(SPacket)),
Size = element_size(SPacket),
Q1 = lqueue_in({FromNick, TSPacket, HaveSubject, TimeStamp, Size},
StateData#state.history),
add_to_log(text, {FromNick, Packet}, StateData),
@ -3582,3 +3611,6 @@ tab_count_user(JID) ->
_ ->
0
end.
element_size(El) ->
size(xml:element_to_binary(El)).

3
src/mod_offline_odbc.erl
View File

@ -123,8 +123,7 @@ loop(Host, AccessMaxOfflineMsgs) ->
M#offline_msg.timestamp))]},
XML =
ejabberd_odbc:escape(
lists:flatten(
xml:element_to_string(Packet))),
xml:element_to_binary(Packet)),
odbc_queries:add_spool_sql(Username, XML)
end, Msgs),
case catch odbc_queries:add_spool(Host, Query) of

10
src/mod_proxy65/mod_proxy65_sm.erl
View File

@ -71,7 +71,9 @@ start_link(Host, Opts) ->
gen_server:start_link({local, Proc}, ?MODULE, [Opts], []).
init([Opts]) ->
update_tables(),
mnesia:create_table(bytestream, [{ram_copies, [node()]},
{local_content, true},
{attributes, record_info(fields, bytestream)}]),
mnesia:add_table_copy(bytestream, node(), ram_copies),
MaxConnections = gen_mod:get_opt(max_connections, Opts, infinity),
@ -179,3 +181,11 @@ activate_stream(SHA1, IJid, TJid, Host) when is_list(SHA1) ->
_ ->
error
end.
update_tables() ->
case catch mnesia:table_info(bytestream, local_content) of
false ->
mnesia:delete_table(bytestream);
_ ->
ok
end.

2
src/odbc/ejabberd_odbc.erl
View File

@ -178,6 +178,7 @@ init([Host, StartInterval]) ->
end,
[DBType | _] = db_opts(Host),
?GEN_FSM:send_event(self(), connect),
ejabberd_odbc_sup:add_pid(Host, self()),
{ok, connecting, #state{db_type = DBType,
host = Host,
max_pending_requests_len = max_fsm_queue(),
@ -274,6 +275,7 @@ handle_info(Info, StateName, State) ->
{next_state, StateName, State}.
terminate(_Reason, _StateName, State) ->
ejabberd_odbc_sup:remove_pid(State#state.host, self()),
case State#state.db_type of
mysql ->
%% old versions of mysql driver don't have the stop function

41
src/odbc/ejabberd_odbc_sup.erl
View File

@ -30,6 +30,8 @@
%% API
-export([start_link/1,
init/1,
add_pid/2,
remove_pid/2,
get_pids/1,
get_random_pid/1
]).
@ -44,7 +46,19 @@
-define(CONNECT_TIMEOUT, 500). % milliseconds
-record(sql_pool, {host, pid}).
start_link(Host) ->
mnesia:create_table(sql_pool,
[{ram_copies, [node()]},
{type, bag},
{local_content, true},
{attributes, record_info(fields, sql_pool)}]),
mnesia:add_table_copy(local_config, node(), ram_copies),
F = fun() ->
mnesia:delete({sql_pool, Host})
end,
mnesia:ets(F),
supervisor:start_link({local, gen_mod:get_module_proc(Host, ?MODULE)},
?MODULE, [Host]).
@ -86,16 +100,25 @@ init([Host]) ->
end, lists:seq(1, PoolSize))}}.
get_pids(Host) ->
Proc = gen_mod:get_module_proc(Host, ?MODULE),
% throw an exception if supervisor is not ready (i.e. if it cannot
% start its children, if the database is down for example)
sys:get_status(Proc, ?CONNECT_TIMEOUT),
[Child ||
{_Id, Child, _Type, _Modules} <- supervisor:which_children(Proc),
Child /= undefined].
Rs = mnesia:dirty_read(sql_pool, Host),
[R#sql_pool.pid || R <- Rs].
get_random_pid(Host) ->
Pids = get_pids(Host),
lists:nth(erlang:phash(now(), length(Pids)), Pids).
add_pid(Host, Pid) ->
F = fun() ->
mnesia:write(
#sql_pool{host = Host,
pid = Pid})
end,
mnesia:ets(F).
remove_pid(Host, Pid) ->
F = fun() ->
mnesia:delete_object(
#sql_pool{host = Host,
pid = Pid})
end,
mnesia:ets(F).

70
src/p1_fsm.erl
View File

@ -517,6 +517,25 @@ print_event(Dev, return, {Name, StateName}) ->
io:format(Dev, "*DBG* ~p switched to state ~w~n",
[Name, StateName]).
relay_messages(MRef, TRef, Clone, Queue) ->
lists:foreach(
fun(Msg) -> Clone ! Msg end,
queue:to_list(Queue)),
relay_messages(MRef, TRef, Clone).
relay_messages(MRef, TRef, Clone) ->
receive
{'DOWN', MRef, process, Clone, Reason} ->
Reason;
{'EXIT', _Parent, _Reason} ->
{migrated, Clone};
{timeout, TRef, timeout} ->
{migrated, Clone};
Msg ->
Clone ! Msg,
relay_messages(MRef, TRef, Clone)
end.
handle_msg(Msg, Parent, Name, StateName, StateData, Mod, _Time,
Limits, Queue, QueueLen) -> %No debug here
From = from(Msg),
@ -535,6 +554,23 @@ handle_msg(Msg, Parent, Name, StateName, StateData, Mod, _Time,
reply(From, Reply),
loop(Parent, Name, NStateName, NStateData, Mod, Time1, [],
Limits, Queue, QueueLen);
{migrate, NStateData, {Node, M, F, A}, Time1} ->
Reason = case catch rpc:call(Node, M, F, A, 5000) of
{badrpc, _} = Err ->
{migration_error, Err};
{'EXIT', _} = Err ->
{migration_error, Err};
{error, _} = Err ->
{migration_error, Err};
{ok, Clone} ->
process_flag(trap_exit, true),
MRef = erlang:monitor(process, Clone),
TRef = erlang:start_timer(Time1, self(), timeout),
relay_messages(MRef, TRef, Clone, Queue);
Reply ->
{migration_error, {bad_reply, Reply}}
end,
terminate(Reason, Name, Msg, Mod, StateName, NStateData, []);
{stop, Reason, NStateData} ->
terminate(Reason, Name, Msg, Mod, StateName, NStateData, []);
{stop, Reason, Reply, NStateData} when From =/= undefined ->
@ -571,6 +607,23 @@ handle_msg(Msg, Parent, Name, StateName, StateData,
Debug1 = reply(Name, From, Reply, Debug, NStateName),
loop(Parent, Name, NStateName, NStateData,
Mod, Time1, Debug1, Limits, Queue, QueueLen);
{migrate, NStateData, {Node, M, F, A}, Time1} ->
Reason = case catch rpc:call(Node, M, F, A, Time1) of
{badrpc, R} ->
{migration_error, R};
{'EXIT', R} ->
{migration_error, R};
{error, R} ->
{migration_error, R};
{ok, Clone} ->
process_flag(trap_exit, true),
MRef = erlang:monitor(process, Clone),
TRef = erlang:start_timer(Time1, self(), timeout),
relay_messages(MRef, TRef, Clone, Queue);
Reply ->
{migration_error, {bad_reply, Reply}}
end,
terminate(Reason, Name, Msg, Mod, StateName, NStateData, Debug);
{stop, Reason, NStateData} ->
terminate(Reason, Name, Msg, Mod, StateName, NStateData, Debug);
{stop, Reason, Reply, NStateData} when From =/= undefined ->
@ -633,12 +686,10 @@ terminate(Reason, Name, Msg, Mod, StateName, StateData, Debug) ->
%% Priority shutdown should be considered as
%% shutdown by SASL
exit(shutdown);
{process_limit, Limit} ->
%% Priority shutdown should be considered as
%% shutdown by SASL
error_logger:error_msg("FSM limit reached (~p): ~p~n",
[self(), Limit]),
exit(shutdown);
{process_limit, _Limit} ->
exit(Reason);
{migrated, _Clone} ->
exit(normal);
_ ->
error_info(Mod, Reason, Name, Msg, StateName, StateData, Debug),
exit(Reason)
@ -705,7 +756,12 @@ get_msg(Msg) -> Msg.
format_status(Opt, StatusData) ->
[PDict, SysState, Parent, Debug, [Name, StateName, StateData, Mod, _Time]] =
StatusData,
Header = lists:concat(["Status for state machine ", Name]),
NameTag = if is_pid(Name) ->
pid_to_list(Name);
is_atom(Name) ->
Name
end,
Header = lists:concat(["Status for state machine ", NameTag]),
Log = sys:get_debug(log, Debug, []),
Specfic =
case erlang:function_exported(Mod, format_status, 2) of

337
src/p1_prof.erl Normal file
View File

@ -0,0 +1,337 @@
%%%-------------------------------------------------------------------
%%% File : p1_prof.erl
%%% Author : Evgeniy Khramtsov <ekhramtsov@process-one.net>
%%% Description : Handy wrapper around eprof and fprof
%%%
%%% Created : 23 Jan 2010 by Evgeniy Khramtsov <ekhramtsov@process-one.net>
%%%
%%%
%%% ejabberd, Copyright (C) 2002-2010 ProcessOne
%%%
%%% This program is free software; you can redistribute it and/or
%%% modify it under the terms of the GNU General Public License as
%%% published by the Free Software Foundation; either version 2 of the
%%% License, or (at your option) any later version.
%%%
%%% This program is distributed in the hope that it will be useful,
%%% but WITHOUT ANY WARRANTY; without even the implied warranty of
%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
%%% General Public License for more details.
%%%
%%% You should have received a copy of the GNU General Public License
%%% along with this program; if not, write to the Free Software
%%% Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
%%% 02111-1307 USA
%%%
%%%-------------------------------------------------------------------
-module(p1_prof).
%% API
-export([eprof_start/0, eprof_stop/0,
fprof_start/0, fprof_start/1,
fprof_stop/0, fprof_analyze/0,
queue/0, queue/1, memory/0, memory/1,
reds/0, reds/1, trace/1, help/0,
q/0, m/0, r/0, q/1, m/1, r/1]).
-define(APPS, [ejabberd, mnesia]).
%%====================================================================
%% API
%%====================================================================
eprof_start() ->
eprof:start(),
case lists:keyfind(running, 1, application:info()) of
{_, Apps} ->
case get_procs(?APPS, Apps) of
[] ->
{error, no_procs_found};
Procs ->
eprof:start_profiling(Procs)
end;
_ ->
{error, no_app_info}
end.
fprof_start() ->
fprof_start(0).
fprof_start(Duration) ->
case lists:keyfind(running, 1, application:info()) of
{_, Apps} ->
case get_procs(?APPS, Apps) of
[] ->
{error, no_procs_found};
Procs ->
fprof:trace([start, {procs, Procs}]),
io:format("Profiling started~n"),
if Duration > 0 ->
timer:sleep(Duration*1000),
fprof:trace([stop]),
fprof:stop();
true->
ok
end
end;
_ ->
{error, no_app_info}
end.
fprof_stop() ->
fprof:trace([stop]),
fprof:profile(),
fprof:analyse([totals, no_details, {sort, own},
no_callers, {dest, "fprof.analysis"}]),
fprof:stop(),
format_fprof_analyze().
fprof_analyze() ->
fprof_stop().
eprof_stop() ->
eprof:stop_profiling(),
case erlang:function_exported(eprof, analyse, 0) of
true ->
eprof:analyse();
false ->
eprof:analyze()
end.
help() ->
M = ?MODULE,
io:format("Brief help:~n"
"~p:queue(N) - show top N pids sorted by queue length~n"
"~p:queue() - shorthand for ~p:queue(10)~n"
"~p:memory(N) - show top N pids sorted by memory usage~n"
"~p:memory() - shorthand for ~p:memory(10)~n"
"~p:reds(N) - show top N pids sorted by reductions~n"
"~p:reds() - shorthand for ~p:reds(10)~n"
"~p:q(N)|~p:q() - same as ~p:queue(N)|~p:queue()~n"
"~p:m(N)|~p:m() - same as ~p:memory(N)|~p:memory()~n"
"~p:r(N)|~p:r() - same as ~p:reds(N)|~p:reds()~n"
"~p:trace(Pid) - trace Pid; to stop tracing close "
"Erlang shell with Ctrl+C~n"
"~p:eprof_start() - start eprof on all available pids; "
"DO NOT use on production system!~n"
"~p:eprof_stop() - stop eprof and print result~n"
"~p:fprof_start() - start fprof on all available pids; "
"DO NOT use on production system!~n"
"~p:fprof_stop() - stop eprof and print formatted result~n"
"~p:fprof_start(N) - start and run fprof for N seconds; "
"use ~p:fprof_analyze() to analyze collected statistics and "
"print formatted result; use on production system with CARE~n"
"~p:fprof_analyze() - analyze previously collected statistics "
"using ~p:fprof_start(N) and print formatted result~n"
"~p:help() - print this help~n",
lists:duplicate(31, M)).
q() ->
queue().
q(N) ->
queue(N).
m() ->
memory().
m(N) ->
memory(N).
r() ->
reds().
r(N) ->
reds(N).
queue() ->
queue(10).
memory() ->
memory(10).
reds() ->
reds(10).
queue(N) ->
dump(N, lists:reverse(lists:ukeysort(1, all_pids(queue)))).
memory(N) ->
dump(N, lists:reverse(lists:ukeysort(2, all_pids(memory)))).
reds(N) ->
dump(N, lists:reverse(lists:ukeysort(3, all_pids(reductions)))).
trace(Pid) ->
erlang:trace(Pid, true, [send, 'receive']),
trace_loop().
trace_loop() ->
receive
M ->
io:format("~p~n", [M]),
trace_loop()
end.
%%====================================================================
%% Internal functions
%%====================================================================
get_procs(Apps, AppList) ->
io:format("Searching for processes to profile...~n", []),
Procs = lists:flatmap(
fun({App, Leader}) when is_pid(Leader) ->
case lists:member(App, Apps) of
true ->
get_procs(Leader);
false ->
[]
end;
(_) ->
[]
end, AppList),
io:format("Found ~p processes~n", [length(Procs)]),
Procs.
get_procs(Leader) ->
lists:filter(
fun(Pid) ->
case process_info(Pid, group_leader) of
{_, Leader} ->
true;
_ ->
false
end
end, processes()).
format_fprof_analyze() ->
case file:consult("fprof.analysis") of
{ok, [_, [{totals, _, _, TotalOWN}] | Rest]} ->
OWNs = lists:flatmap(
fun({MFA, _, _, OWN}) ->
Percent = OWN*100/TotalOWN,
case round(Percent) of
0 ->
[];
_ ->
[{mfa_to_list(MFA), Percent}]
end
end, Rest),
ACCs = collect_accs(Rest),
MaxACC = find_max(ACCs),
MaxOWN = find_max(OWNs),
io:format("=== Sorted by OWN:~n"),
lists:foreach(
fun({MFA, Per}) ->
L = length(MFA),
S = lists:duplicate(MaxOWN - L + 2, $ ),
io:format("~s~s~.2f%~n", [MFA, S, Per])
end, lists:reverse(lists:keysort(2, OWNs))),
io:format("~n=== Sorted by ACC:~n"),
lists:foreach(
fun({MFA, Per}) ->
L = length(MFA),
S = lists:duplicate(MaxACC - L + 2, $ ),
io:format("~s~s~.2f%~n", [MFA, S, Per])
end, lists:reverse(lists:keysort(2, ACCs)));
Err ->
Err
end.
mfa_to_list({M, F, A}) ->
atom_to_list(M) ++ ":" ++ atom_to_list(F) ++ "/" ++ integer_to_list(A);
mfa_to_list(F) when is_atom(F) ->
atom_to_list(F).
find_max(List) ->
find_max(List, 0).
find_max([{V, _}|Tail], Acc) ->
find_max(Tail, lists:max([length(V), Acc]));
find_max([], Acc) ->
Acc.
collect_accs(List) ->
List1 = lists:filter(
fun({MFA, _, _, _}) ->
case MFA of
{sys, _, _} ->
false;
suspend ->
false;
{gen_fsm, _, _} ->
false;
{p1_fsm, _, _} ->
false;
{gen, _, _} ->
false;
{gen_server, _, _} ->
false;
{proc_lib, _, _} ->
false;
_ ->
true
end
end, List),
TotalACC = lists:sum([A || {_, _, A, _} <- List1]),
lists:flatmap(
fun({MFA, _, ACC, _}) ->
Percent = ACC*100/TotalACC,
case round(Percent) of
0 ->
[];
_ ->
[{mfa_to_list(MFA), Percent}]
end
end, List1).
all_pids(Type) ->
lists:foldl(
fun(P, Acc) when P == self() ->
%% exclude ourself from statistics
Acc;
(P, Acc) ->
case catch process_info(
P,
[message_queue_len,
memory,
reductions,
dictionary,
current_function,
registered_name]) of
[{_, Len}, {_, Memory}, {_, Reds},
{_, Dict}, {_, CurFun}, {_, RegName}] ->
if Type == queue andalso Len == 0 ->
Acc;
true ->
[{Len, Memory, Reds, Dict, CurFun, P, RegName}|Acc]
end;
_ ->
Acc
end
end, [], processes()).
dump(N, Rs) ->
lists:foreach(
fun({MsgQLen, Memory, Reds, Dict, CurFun, Pid, RegName}) ->
PidStr = pid_to_list(Pid),
[_, Maj, Min] = string:tokens(
string:substr(
PidStr, 2, length(PidStr) - 2), "."),
io:format("** pid(0,~s,~s)~n"
"** registered name: ~p~n"
"** memory: ~p~n"
"** reductions: ~p~n"
"** message queue len: ~p~n"
"** current_function: ~p~n"
"** dictionary: ~p~n~n",
[Maj, Min, RegName, Memory, Reds, MsgQLen, CurFun, Dict])
end, nthhead(N, Rs)).
nthhead(N, L) ->
lists:reverse(nthhead(N, L, [])).
nthhead(0, _L, Acc) ->
Acc;
nthhead(N, [H|T], Acc) ->
nthhead(N-1, T, [H|Acc]);
nthhead(_N, [], Acc) ->
Acc.

1
src/tls/Makefile.in
View File

@ -21,6 +21,7 @@ ifeq ($(shell uname),SunOs)
DYNAMIC_LIB_CFLAGS = -KPIC -G -z text
endif
EFLAGS += @ERLANG_SSLVER@
EFLAGS += -I ..
EFLAGS += -pz ..

10
src/tls/tls.erl
View File

@ -61,6 +61,13 @@
-define(GET_VERIFY_RESULT, 8).
-define(VERIFY_NONE, 16#10000).
-ifdef(SSL40).
-define(CERT_DECODE, {public_key, pkix_decode_cert, plain}).
-else.
-define(CERT_DECODE, {ssl_pkix, decode_cert, [pkix]}).
-endif.
-record(tlssock, {tcpsock, tlsport}).
start() ->
@ -232,7 +239,8 @@ close(#tlssock{tcpsock = TCPSocket, tlsport = Port}) ->
get_peer_certificate(#tlssock{tlsport = Port}) ->
case port_control(Port, ?GET_PEER_CERTIFICATE, []) of
<<0, BCert/binary>> ->
case catch ssl_pkix:decode_cert(BCert, [pkix]) of
{CertMod, CertFun, CertSecondArg} = ?CERT_DECODE,
case catch apply(CertMod, CertFun, [BCert, CertSecondArg]) of
{ok, Cert} ->
{ok, Cert};
_ ->

2
src/web/Makefile.in
View File

@ -9,7 +9,7 @@ LIBS = @LIBS@
ERLANG_CFLAGS = @ERLANG_CFLAGS@
ERLANG_LIBS = @ERLANG_LIBS@
EFLAGS += @ERLANG_SSL39@
EFLAGS += @ERLANG_SSLVER@
EFLAGS += -I ..
EFLAGS += -pz ..

110
src/web/ejabberd_http_bind.erl
View File

@ -27,6 +27,7 @@
setopts/2,
controlling_process/2,
become_controller/2,
change_controller/2,
custom_receiver/1,
reset_stream/1,
change_shaper/2,
@ -121,9 +122,19 @@
start(XMPPDomain, Sid, Key, IP) ->
?DEBUG("Starting session", []),
case catch supervisor:start_child(ejabberd_http_bind_sup, [Sid, Key, IP]) of
{ok, Pid} -> {ok, Pid};
_ -> check_bind_module(XMPPDomain),
{error, "Cannot start HTTP bind session"}
{ok, Pid} ->
{ok, Pid};
{error, _} = Err ->
case check_bind_module(XMPPDomain) of
false ->
{error, "Cannot start HTTP bind session"};
true ->
?ERROR_MSG("Cannot start HTTP bind session: ~p", [Err]),
Err
end;
Exit ->
?ERROR_MSG("Cannot start HTTP bind session: ~p", [Exit]),
{error, Exit}
end.
start_link(Sid, Key, IP) ->
@ -140,7 +151,13 @@ setopts({http_bind, FsmRef, _IP}, Opts) ->
true ->
gen_fsm:send_all_state_event(FsmRef, {activate, self()});
_ ->
ok
case lists:member({active, false}, Opts) of
true ->
gen_fsm:sync_send_all_state_event(
FsmRef, deactivate_socket);
_ ->
ok
end
end.
controlling_process(_Socket, _Pid) ->
@ -152,6 +169,9 @@ custom_receiver({http_bind, FsmRef, _IP}) ->
become_controller(FsmRef, C2SPid) ->
gen_fsm:send_all_state_event(FsmRef, {become_controller, C2SPid}).
change_controller({http_bind, FsmRef, _IP}, C2SPid) ->
become_controller(FsmRef, C2SPid).
reset_stream({http_bind, _FsmRef, _IP}) ->
ok.
@ -170,7 +190,6 @@ sockname(_Socket) ->
peername({http_bind, _FsmRef, IP}) ->
{ok, IP}.
%% Entry point for data coming from client through ejabberd HTTP server:
process_request(Data, IP) ->
Opts1 = ejabberd_c2s_config:get_c2s_limits(),
@ -192,12 +211,12 @@ process_request(Data, IP) ->
"xmlns='" ++ ?NS_HTTP_BIND ++ "'/>"};
XmppDomain ->
%% create new session
Sid = sha:sha(term_to_binary({now(), make_ref()})),
Sid = make_sid(),
case start(XmppDomain, Sid, "", IP) of
{error, _} ->
{200, ?HEADER, "<body type='terminate' "
{500, ?HEADER, "<body type='terminate' "
"condition='internal-server-error' "
"xmlns='" ++ ?NS_HTTP_BIND ++ "'>BOSH module not started</body>"};
"xmlns='" ++ ?NS_HTTP_BIND ++ "'>Internal Server Error</body>"};
{ok, Pid} ->
handle_session_start(
Pid, XmppDomain, Sid, Rid, Attrs,
@ -223,10 +242,10 @@ process_request(Data, IP) ->
handle_http_put(Sid, Rid, Attrs, Payload2, PayloadSize,
StreamStart, IP);
{size_limit, Sid} ->
case mnesia:dirty_read({http_bind, Sid}) of
[] ->
case get_session(Sid) of
{error, _} ->
{404, ?HEADER, ""};
[#http_bind{pid = FsmRef}] ->
{ok, #http_bind{pid = FsmRef}} ->
gen_fsm:sync_send_all_state_event(FsmRef, {stop, close}),
{200, ?HEADER, "<body type='terminate' "
"condition='undefined-condition' "
@ -282,7 +301,7 @@ handle_session_start(Pid, XmppDomain, Sid, Rid, Attrs,
end,
XmppVersion = xml:get_attr_s("xmpp:version", Attrs),
?DEBUG("Create session: ~p", [Sid]),
mnesia:transaction(
mnesia:async_dirty(
fun() ->
mnesia:write(
#http_bind{id = Sid,
@ -340,6 +359,7 @@ init([Sid, Key, IP]) ->
%% {stop, Reason, NewStateData}
%%----------------------------------------------------------------------
handle_event({become_controller, C2SPid}, StateName, StateData) ->
erlang:monitor(process, C2SPid),
case StateData#state.input of
cancel ->
{next_state, StateName, StateData#state{
@ -404,6 +424,14 @@ handle_sync_event({stop,close}, _From, _StateName, StateData) ->
handle_sync_event({stop,stream_closed}, _From, _StateName, StateData) ->
Reply = ok,
{stop, normal, Reply, StateData};
handle_sync_event(deactivate_socket, _From, StateName, StateData) ->
%% Input = case StateData#state.input of
%% cancel ->
%% queue:new();
%% Q ->
%% Q
%% end,
{reply, ok, StateName, StateData#state{waiting_input = false}};
handle_sync_event({stop,Reason}, _From, _StateName, StateData) ->
?DEBUG("Closing bind session ~p - Reason: ~p", [StateData#state.id, Reason]),
Reply = ok,
@ -537,6 +565,9 @@ handle_info({timeout, ShaperTimer, _}, StateName,
#state{shaper_timer = ShaperTimer} = StateData) ->
{next_state, StateName, StateData#state{shaper_timer = undefined}};
handle_info({'DOWN', _MRef, process, C2SPid, _}, _StateName,
#state{waiting_input = C2SPid} = StateData) ->
{stop, normal, StateData};
handle_info(_, StateName, StateData) ->
{next_state, StateName, StateData}.
@ -790,7 +821,7 @@ handle_http_put(Sid, Rid, Attrs, Payload, PayloadSize, StreamStart, IP) ->
?DEBUG("Trafic Shaper: Delaying request ~p", [Rid]),
timer:sleep(Pause),
%{200, ?HEADER,
% xml:element_to_string(
% xml:element_to_binary(
% {xmlelement, "body",
% [{"xmlns", ?NS_HTTP_BIND},
% {"type", "error"}], []})};
@ -804,10 +835,10 @@ handle_http_put(Sid, Rid, Attrs, Payload, PayloadSize, StreamStart, IP) ->
http_put(Sid, Rid, Attrs, Payload, PayloadSize, StreamStart, IP) ->
?DEBUG("Looking for session: ~p", [Sid]),
case mnesia:dirty_read({http_bind, Sid}) of
[] ->
case get_session(Sid) of
{error, _} ->
{error, not_exists};
[#http_bind{pid = FsmRef, hold=Hold, to={To, StreamVersion}}=Sess] ->
{ok, #http_bind{pid = FsmRef, hold=Hold, to={To, StreamVersion}}=Sess}->
NewStream =
case StreamStart of
true ->
@ -827,21 +858,21 @@ handle_http_put_error(Reason, #http_bind{pid=FsmRef, version=Version})
case Reason of
not_exists ->
{200, ?HEADER,
xml:element_to_string(
xml:element_to_binary(
{xmlelement, "body",
[{"xmlns", ?NS_HTTP_BIND},
{"type", "terminate"},
{"condition", "item-not-found"}], []})};
bad_key ->
{200, ?HEADER,
xml:element_to_string(
xml:element_to_binary(
{xmlelement, "body",
[{"xmlns", ?NS_HTTP_BIND},
{"type", "terminate"},
{"condition", "item-not-found"}], []})};
polling_too_frequently ->
{200, ?HEADER,
xml:element_to_string(
xml:element_to_binary(
{xmlelement, "body",
[{"xmlns", ?NS_HTTP_BIND},
{"type", "terminate"},
@ -986,7 +1017,7 @@ prepare_outpacket_response(#http_bind{id=Sid, wait=Wait,
MaxInactivity = get_max_inactivity(To, ?MAX_INACTIVITY),
MaxPause = get_max_pause(To),
{200, ?HEADER,
xml:element_to_string(
xml:element_to_binary(
{xmlelement,"body",
[{"xmlns",
?NS_HTTP_BIND},
@ -1041,7 +1072,7 @@ send_outpacket(#http_bind{pid = FsmRef}, OutPacket) ->
true ->
TypedEls = [check_default_xmlns(OEl) ||
{xmlstreamelement, OEl} <- OutPacket],
Body = xml:element_to_string(
Body = xml:element_to_binary(
{xmlelement,"body",
[{"xmlns",
?NS_HTTP_BIND}],
@ -1075,7 +1106,7 @@ send_outpacket(#http_bind{pid = FsmRef}, OutPacket) ->
StreamTail]
end,
{200, ?HEADER,
xml:element_to_string(
xml:element_to_binary(
{xmlelement,"body",
[{"xmlns",
?NS_HTTP_BIND}],
@ -1191,7 +1222,7 @@ set_inactivity_timer(_Pause, MaxInactivity) ->
elements_to_string([]) ->
[];
elements_to_string([El | Els]) ->
xml:element_to_string(El) ++ elements_to_string(Els).
[xml:element_to_binary(El)|elements_to_string(Els)].
%% @spec (To, Default::integer()) -> integer()
%% where To = [] | {Host::string(), Version::string()}
@ -1225,7 +1256,36 @@ check_default_xmlns({xmlelement, Name, Attrs, Els} = El) ->
%% Print a warning in log file if this is not the case.
check_bind_module(XmppDomain) ->
case gen_mod:is_loaded(XmppDomain, mod_http_bind) of
true -> ok;
true -> true;
false -> ?ERROR_MSG("You are trying to use BOSH (HTTP Bind), but the module mod_http_bind is not started.~n"
"Check your 'modules' section in your ejabberd configuration file.",[])
"Check your 'modules' section in your ejabberd configuration file.",[]),
false
end.
make_sid() ->
sha:sha(term_to_binary({now(), make_ref()}))
++ "-" ++ ejabberd_cluster:node_id().
get_session(SID) ->
case string:tokens(SID, "-") of
[_, NodeID] ->
case ejabberd_cluster:get_node_by_id(NodeID) of
Node when Node == node() ->
case mnesia:dirty_read({http_bind, SID}) of
[] ->
{error, enoent};
[Session] ->
{ok, Session}
end;
Node ->
case catch rpc:call(Node, mnesia, dirty_read,
[{http_bind, SID}], 5000) of
[Session] ->
{ok, Session};
_ ->
{error, enoent}
end
end;
_ ->
{error, enoent}
end.

57
src/web/ejabberd_http_poll.erl
View File

@ -77,9 +77,12 @@
%%% API
%%%----------------------------------------------------------------------
start(ID, Key, IP) ->
update_tables(),
mnesia:create_table(http_poll,
[{ram_copies, [node()]},
{local_content, true},
{attributes, record_info(fields, http_poll)}]),
mnesia:add_table_copy(http_poll, node(), ram_copies),
supervisor:start_child(ejabberd_http_poll_sup, [ID, Key, IP]).
start_link(ID, Key, IP) ->
@ -115,9 +118,9 @@ process([], #request{data = Data,
{ok, ID1, Key, NewKey, Packet} ->
ID = if
(ID1 == "0") or (ID1 == "mobile") ->
NewID = sha:sha(term_to_binary({now(), make_ref()})),
NewID = make_sid(),
{ok, Pid} = start(NewID, "", IP),
mnesia:transaction(
mnesia:async_dirty(
fun() ->
mnesia:write(#http_poll{id = NewID,
pid = Pid})
@ -350,7 +353,7 @@ handle_info(_, StateName, StateData) ->
%% Returns: any
%%----------------------------------------------------------------------
terminate(_Reason, _StateName, StateData) ->
mnesia:transaction(
mnesia:async_dirty(
fun() ->
mnesia:delete({http_poll, StateData#state.id})
end),
@ -375,19 +378,19 @@ terminate(_Reason, _StateName, StateData) ->
%%%----------------------------------------------------------------------
http_put(ID, Key, NewKey, Packet) ->
case mnesia:dirty_read({http_poll, ID}) of
[] ->
case get_session(ID) of
{error, _} ->
{error, not_exists};
[#http_poll{pid = FsmRef}] ->
{ok, #http_poll{pid = FsmRef}} ->
gen_fsm:sync_send_all_state_event(
FsmRef, {http_put, Key, NewKey, Packet})
end.
http_get(ID) ->
case mnesia:dirty_read({http_poll, ID}) of
[] ->
case get_session(ID) of
{error, _} ->
{error, not_exists};
[#http_poll{pid = FsmRef}] ->
{ok, #http_poll{pid = FsmRef}} ->
gen_fsm:sync_send_all_state_event(FsmRef, http_get)
end.
@ -446,3 +449,39 @@ get_jid(Type, ParsedPacket) ->
false ->
jlib:make_jid("","","")
end.
update_tables() ->
case catch mnesia:table_info(http_poll, local_content) of
false ->
mnesia:delete_table(http_poll);
_ ->
ok
end.
make_sid() ->
sha:sha(term_to_binary({now(), make_ref()}))
++ "-" ++ ejabberd_cluster:node_id().
get_session(SID) ->
case string:tokens(SID, "-") of
[_, NodeID] ->
case ejabberd_cluster:get_node_by_id(NodeID) of
Node when Node == node() ->
case mnesia:dirty_read({http_poll, SID}) of
[] ->
{error, enoent};
[Session] ->
{ok, Session}
end;
Node ->
case catch rpc:call(Node, mnesia, dirty_read,
[{http_poll, SID}], 5000) of
[Session] ->
{ok, Session};
_ ->
{error, enoent}
end
end;
_ ->
{error, enoent}
end.

10
src/web/mod_http_bind.erl
View File

@ -134,7 +134,9 @@ setup_database() ->
migrate_database(),
mnesia:create_table(http_bind,
[{ram_copies, [node()]},
{attributes, record_info(fields, http_bind)}]).
{local_content, true},
{attributes, record_info(fields, http_bind)}]),
mnesia:add_table_copy(http_bind, node(), ram_copies).
migrate_database() ->
case catch mnesia:table_info(http_bind, attributes) of
@ -144,4 +146,10 @@ migrate_database() ->
%% Since the stored information is not important, instead
%% of actually migrating data, let's just destroy the table
mnesia:delete_table(http_bind)
end,
case catch mnesia:table_info(http_bind, local_content) of
false ->
mnesia:delete_table(http_bind);
_ ->
ok
end.

4
src/web/mod_http_fileserver.erl
View File

@ -66,11 +66,15 @@
headers
}).
-ifdef(SSL40).
-define(STRING2LOWER, string).
-else.
-ifdef(SSL39).
-define(STRING2LOWER, string).
-else.
-define(STRING2LOWER, httpd_util).
-endif.
-endif.
-record(state, {host, docroot, accesslog, accesslogfd, directory_indices,
custom_headers, default_content_type, content_types = []}).

230
src/xml.c Normal file
View File

@ -0,0 +1,230 @@
#include <erl_nif.h>
#include <string.h>
#include <stdio.h>
struct buf {
int limit;
int len;
unsigned char *b;
};
static int make_element(ErlNifEnv* env, struct buf *rbuf, ERL_NIF_TERM el);
static struct buf *init_buf(ErlNifEnv* env)
{
struct buf *rbuf = enif_alloc(env, sizeof(struct buf));
rbuf->limit = 1024;
rbuf->len = 0;
rbuf->b = enif_alloc(env, rbuf->limit);
return rbuf;
}
static void destroy_buf(ErlNifEnv* env, struct buf *rbuf)
{
if (rbuf) {
if (rbuf->b) {
enif_free(env, rbuf->b);
};
enif_free(env, rbuf);
};
}
inline void resize_buf(ErlNifEnv* env, struct buf *rbuf, int len_to_add)
{
int new_len = rbuf->len + len_to_add;
if (new_len >= rbuf->limit) {
rbuf->limit = ((new_len / 1024) + 1) * 1024;
rbuf->b = enif_realloc(env, rbuf->b, rbuf->limit);
};
}
static void buf_add_char(ErlNifEnv* env, struct buf *rbuf, unsigned char c)
{
resize_buf(env, rbuf, 1);
(rbuf->b)[rbuf->len] = c;
rbuf->len += 1;
}
static void buf_add_str(ErlNifEnv* env, struct buf *rbuf, char *data, int len)
{
resize_buf(env, rbuf, len);
memcpy(rbuf->b + rbuf->len, data, len);
rbuf->len += len;
}
inline void crypt(ErlNifEnv* env, struct buf *rbuf, unsigned char *data, int len)
{
int i;
for (i = 0; i < len; i++) {
switch (data[i]) {
case '&':
buf_add_str(env, rbuf, "&amp;", 5);
break;
case '<':
buf_add_str(env, rbuf, "&lt;", 4);
break;
case '>':
buf_add_str(env, rbuf, "&gt;", 4);
break;
case '"':
buf_add_str(env, rbuf, "&quot;", 6);
break;
case '\'':
buf_add_str(env, rbuf, "&apos;", 6);
break;
default:
buf_add_char(env, rbuf, data[i]);
break;
};
};
}
static int make_elements(ErlNifEnv* env, struct buf *rbuf, ERL_NIF_TERM els)
{
ERL_NIF_TERM head, tail;
int ret = 0;
while (enif_get_list_cell(env, els, &head, &tail)) {
ret = make_element(env, rbuf, head);
if (ret) {
els = tail;
} else {
break;
};
};
return ret;
}
static int make_attrs(ErlNifEnv* env, struct buf *rbuf, ERL_NIF_TERM attrs)
{
ErlNifBinary name, data;
ERL_NIF_TERM head, tail;
const ERL_NIF_TERM *tuple;
int arity, ret = 1;
while (enif_get_list_cell(env, attrs, &head, &tail)) {
if (enif_get_tuple(env, head, &arity, &tuple)) {
if (arity == 2) {
if (enif_inspect_iolist_as_binary(env, tuple[0], &name) &&
enif_inspect_iolist_as_binary(env, tuple[1], &data)) {
buf_add_char(env, rbuf, ' ');
buf_add_str(env, rbuf, (char *)name.data, name.size);
buf_add_str(env, rbuf, "='", 2);
crypt(env, rbuf, data.data, data.size);
buf_add_char(env, rbuf, '\'');
attrs = tail;
} else {
ret = 0;
break;
};
} else {
ret = 0;
break;
};
} else {
ret = 0;
break;
};
};
return ret;
}
static int make_element(ErlNifEnv* env, struct buf *rbuf, ERL_NIF_TERM el)
{
ErlNifBinary cdata, name;
const ERL_NIF_TERM *tuple;
int arity, ret = 0;
if (enif_get_tuple(env, el, &arity, &tuple)) {
if (arity == 2) {
if (!enif_compare(env, tuple[0], enif_make_atom(env, "xmlcdata"))) {
if (enif_inspect_iolist_as_binary(env, tuple[1], &cdata)) {
crypt(env, rbuf, cdata.data, cdata.size);
ret = 1;
};
};
};
if (arity == 4) {
if (!enif_compare(env, tuple[0], enif_make_atom(env, "xmlelement"))) {
if (enif_inspect_iolist_as_binary(env, tuple[1], &name)) {
buf_add_char(env, rbuf, '<');
buf_add_str(env, rbuf, (char *)name.data, name.size);
ret = make_attrs(env, rbuf, tuple[2]);
if (ret) {
if (enif_is_empty_list(env, tuple[3])) {
buf_add_str(env, rbuf, "/>", 2);
} else {
buf_add_char(env, rbuf, '>');
ret = make_elements(env, rbuf, tuple[3]);
if (ret) {
buf_add_str(env, rbuf, "</", 2);
buf_add_str(env, rbuf, (char*)name.data, name.size);
buf_add_char(env, rbuf, '>');
};
};
};
};
};
};
};
return ret;
}
static ERL_NIF_TERM element_to(ErlNifEnv* env, int argc,
const ERL_NIF_TERM argv[],
int as_string)
{
ErlNifBinary output;
ERL_NIF_TERM result;
struct buf *rbuf;
if (argc == 1) {
rbuf = init_buf(env);
if (make_element(env, rbuf, argv[0])) {
if (as_string) {
(rbuf->b)[rbuf->len] = 0;
result = enif_make_string(env, (char *) rbuf->b, ERL_NIF_LATIN1);
destroy_buf(env, rbuf);
return result;
} else {
if (enif_alloc_binary(env, rbuf->len, &output)) {
memcpy(output.data, rbuf->b, rbuf->len);
result = enif_make_binary(env, &output);
destroy_buf(env, rbuf);
return result;
};
};
};
destroy_buf(env, rbuf);
};
return enif_make_badarg(env);
}
/* static ERL_NIF_TERM element_to_string(ErlNifEnv* env, int argc, */
/* const ERL_NIF_TERM argv[]) */
/* { */
/* return element_to(env, argc, argv, 1); */
/* } */
static ERL_NIF_TERM element_to_binary(ErlNifEnv* env, int argc,
const ERL_NIF_TERM argv[])
{
return element_to(env, argc, argv, 0);
}
static ErlNifFunc nif_funcs[] =
{
/* Stupid Erlang bug with enif_make_string() is fixed
in R14A only (OTP-8685), so we can't use
element_to_string yet.*/
/* {"element_to_string", 1, element_to_string}, */
{"element_to_binary", 1, element_to_binary}
};
ERL_NIF_INIT(xml, nif_funcs, NULL, NULL, NULL, NULL)

19
src/xml.erl
View File

@ -37,8 +37,11 @@
get_subtag/2, get_subtag_cdata/2,
append_subtags/2,
get_path_s/2,
start/0,
replace_tag_attr/3]).
-include("ejabberd.hrl").
%% Select at compile time how to escape characters in binary text
%% nodes.
%% Can be choosen with ./configure --enable-full-xml
@ -48,6 +51,22 @@
-define(ESCAPE_BINARY(CData), crypt(CData)).
-endif.
%% Replace element_to_binary/1 with NIF
%% Can be choosen with ./configure --enable-nif
-ifdef(NIF).
start() ->
SOPath = filename:join(ejabberd:get_so_path(), "xml"),
case catch erlang:load_nif(SOPath, 0) of
ok ->
ok;
Err ->
?WARNING_MSG("unable to load xml NIF: ~p", [Err])
end.
-else.
start() ->
ok.
-endif.
element_to_binary(El) ->
iolist_to_binary(element_to_string(El)).

3
src/xml_stream.erl
View File

@ -31,6 +31,7 @@
new/2,
parse/2,
close/1,
change_callback_pid/2,
parse_element/1]).
-define(XML_START, 0).
@ -106,6 +107,8 @@ new(CallbackPid, MaxSize) ->
size = 0,
maxsize = MaxSize}.
change_callback_pid(State, CallbackPid) ->
State#xml_stream_state{callback_pid = CallbackPid}.
parse(#xml_stream_state{callback_pid = CallbackPid,
port = Port,