xmpp.chapril.org-ejabberd/src/web/ejabberd_http_bind.erl

%%%----------------------------------------------------------------------
%%% File    : ejabberd_http_bind.erl
%%% Author  : Stefan Strigler <steve@zeank.in-berlin.de>
%%% Purpose : Implements XMPP over BOSH (XEP-0205) (formerly known as
%%%           HTTP Binding)
%%% Created : 21 Sep 2005 by Stefan Strigler <steve@zeank.in-berlin.de>
%%% Modified: may 2009 by Mickael Remond, Alexey Schepin
%%% Id      : $Id: ejabberd_http_bind.erl 953 2009-05-07 10:40:40Z alexey $
%%%----------------------------------------------------------------------

-module(ejabberd_http_bind).

-behaviour(gen_fsm).

%% External exports
-export([start_link/3,
	 init/1,
	 handle_event/3,
	 handle_sync_event/4,
	 code_change/4,
	 handle_info/3,
	 terminate/3,
	 send/2,
	 send_xml/2,
	 sockname/1,
	 peername/1,
	 setopts/2,
	 controlling_process/2,
	 become_controller/2,
	 custom_receiver/1,
	 reset_stream/1,
	 change_shaper/2,
	 monitor/1,
	 close/1,
	 process_request/2]).

-include("ejabberd.hrl").
-include("jlib.hrl").
-include("ejabberd_http.hrl").

%% TODO: Use exmpp library instead of including this
-define(NS_STREAM, "http://etherx.jabber.org/streams").

-record(http_bind, {id, pid, to, hold, wait, version}).

-define(NULL_PEER, {{0, 0, 0, 0}, 0}).

%% http binding request
-record(hbr, {rid,
	      key,
	      out}).

-record(state, {id,
		rid = none,
		key,
		socket,
		output = "",
		input = queue:new(),
		waiting_input = false,
		shaper_state,
		shaper_timer,
		last_receiver,
		last_poll,
		http_receiver,
		wait_timer,
		ctime = 0,
		timer,
		pause=0,
		unprocessed_req_list = [], % list of request that have been delayed for proper reordering: {Request, PID}
		req_list = [], % list of requests (cache)
		max_inactivity,
		max_pause,
		ip = ?NULL_PEER
	       }).

%% Internal request format:
-record(http_put, {rid,
		   attrs,
		   payload,
		   payload_size,
		   hold,
		   stream,
		   ip}).

%%-define(DBGFSM, true).
-ifdef(DBGFSM).
-define(FSMOPTS, [{debug, [trace]}]).
-else.
-define(FSMOPTS, []).
-endif.

-define(BOSH_VERSION, "1.8").
-define(NS_CLIENT, "jabber:client").
-define(NS_BOSH, "urn:xmpp:xbosh").
-define(NS_HTTP_BIND, "http://jabber.org/protocol/httpbind").

-define(MAX_REQUESTS, 2).  % number of simultaneous requests
-define(MIN_POLLING, 2000000). % don't poll faster than that or we will
                               % shoot you (time in microsec)
-define(MAX_WAIT, 3600). % max num of secs to keep a request on hold
-define(MAX_INACTIVITY, 30000). % msecs to wait before terminating
                                % idle sessions
-define(MAX_PAUSE, 120). % may num of sec a client is allowed to pause
                         % the session

-define(CT, {"Content-Type", "text/xml; charset=utf-8"}).
-define(HEADER, [?CT]).


%%%----------------------------------------------------------------------
%%% API
%%%----------------------------------------------------------------------
%% TODO: If compile with no supervisor option, start the session without
%%       supervisor
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"}
    end.

start_link(Sid, Key, IP) ->
    gen_fsm:start_link(?MODULE, [Sid, Key, IP], ?FSMOPTS).

send({http_bind, FsmRef, _IP}, Packet) ->
    gen_fsm:sync_send_all_state_event(FsmRef, {send, Packet}).

send_xml({http_bind, FsmRef, _IP}, Packet) ->
    gen_fsm:sync_send_all_state_event(FsmRef, {send_xml, Packet}).

setopts({http_bind, FsmRef, _IP}, Opts) ->
    case lists:member({active, once}, Opts) of
	true ->
	    gen_fsm:send_all_state_event(FsmRef, {activate, self()});
	_ ->
	    ok
    end.

controlling_process(_Socket, _Pid) ->
    ok.

custom_receiver({http_bind, FsmRef, _IP}) ->
    {receiver, ?MODULE, FsmRef}.

become_controller(FsmRef, C2SPid) ->
    gen_fsm:send_all_state_event(FsmRef, {become_controller, C2SPid}).

reset_stream({http_bind, _FsmRef, _IP}) ->
    ok.

change_shaper({http_bind, FsmRef, _IP}, Shaper) ->
    gen_fsm:send_all_state_event(FsmRef, {change_shaper, Shaper}).

monitor({http_bind, FsmRef, _IP}) ->
    erlang:monitor(process, FsmRef).

close({http_bind, FsmRef, _IP}) ->
    catch gen_fsm:sync_send_all_state_event(FsmRef, {stop, close}).

sockname(_Socket) ->
    {ok, ?NULL_PEER}.

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(),
    Opts = [{xml_socket, true} | Opts1],
    MaxStanzaSize =
	case lists:keysearch(max_stanza_size, 1, Opts) of
	    {value, {_, Size}} -> Size;
	    _ -> infinity
	end,
    PayloadSize = iolist_size(Data),
    case catch parse_request(Data, PayloadSize, MaxStanzaSize) of
	%% No existing session:
	{ok, {"", Rid, Attrs, Payload}} ->
	    case xml:get_attr_s("to",Attrs) of
                "" ->
		    ?ERROR_MSG("Session not created (Improper addressing)", []),
		    {200, ?HEADER, "<body type='terminate' "
		     "condition='improper-addressing' "
		     "xmlns='" ++ ?NS_HTTP_BIND ++ "'/>"};
                XmppDomain ->
                    %% create new session
                    Sid = sha:sha(term_to_binary({now(), make_ref()})),
                    case start(XmppDomain, Sid, "", IP) of
			{error, _} ->
			    {200, ?HEADER, "<body type='terminate' "
			     "condition='internal-server-error' "
			     "xmlns='" ++ ?NS_HTTP_BIND ++ "'>BOSH module not started</body"};
			{ok, Pid} ->
			    handle_session_start(
			      Pid, XmppDomain, Sid, Rid, Attrs,
			      Payload, PayloadSize, IP)
		    end
            end;
	%% Existing session
        {ok, {Sid, Rid, Attrs, Payload1}} ->
            StreamStart =
                case xml:get_attr_s("xmpp:restart",Attrs) of
                    "true" ->
                        true;
                    _ ->
                        false
                end,
            Payload2 = case xml:get_attr_s("type",Attrs) of
                           "terminate" ->
                               %% close stream
                               Payload1 ++ [{xmlstreamend, "stream:stream"}];
                           _ ->
                               Payload1
                       end,
            handle_http_put(Sid, Rid, Attrs, Payload2, PayloadSize,
			    StreamStart, IP);
        {size_limit, Sid} ->
	    case mnesia:dirty_read({http_bind, Sid}) of
		[] ->
		    {404, ?HEADER, ""};
		[#http_bind{pid = FsmRef}] ->
		    gen_fsm:sync_send_all_state_event(FsmRef, {stop, close}),
		    {200, ?HEADER, "<body type='terminate' "
		     "condition='undefined-condition' "
		     "xmlns='" ++ ?NS_HTTP_BIND ++ "'>Request Too Large</body>"}
            end;
        _ ->
	    ?DEBUG("Received bad request: ~p", [Data]),
            {400, ?HEADER, ""}
    end.

handle_session_start(Pid, XmppDomain, Sid, Rid, Attrs,
		     Payload, PayloadSize, IP) ->
    ?DEBUG("got pid: ~p", [Pid]),
    Wait = case string:to_integer(xml:get_attr_s("wait",Attrs)) of
	       {error, _} ->
		   ?MAX_WAIT;
	       {CWait, _} ->
		   if
		       (CWait > ?MAX_WAIT) ->
			   ?MAX_WAIT;
		       true ->
			   CWait
		   end
	   end,
    Hold = case string:to_integer(xml:get_attr_s("hold",Attrs)) of
	       {error, _} ->
		   (?MAX_REQUESTS - 1);
	       {CHold, _} ->
		   if
		       (CHold > (?MAX_REQUESTS - 1)) ->
			   (?MAX_REQUESTS - 1);
		       true ->
			   CHold
		   end
	   end,
    Version =
	case catch list_to_float(
		     xml:get_attr_s("ver", Attrs)) of
	    {'EXIT', _} -> 0.0;
	    V -> V
	end,
    XmppVersion = xml:get_attr_s("xmpp:version", Attrs),
    ?DEBUG("Create session: ~p", [Sid]),
    mnesia:transaction(
      fun() ->
	      mnesia:write(
		#http_bind{id = Sid,
			   pid = Pid,
			   to = {XmppDomain,
				 XmppVersion},
			   hold = Hold,
			   wait = Wait,
			   version = Version
			  })
      end),
    handle_http_put(Sid, Rid, Attrs, Payload, PayloadSize, true, IP).

%%%----------------------------------------------------------------------
%%% Callback functions from gen_fsm
%%%----------------------------------------------------------------------

%%----------------------------------------------------------------------
%% Func: init/1
%% Returns: {ok, StateName, StateData}          |
%%          {ok, StateName, StateData, Timeout} |
%%          ignore                              |
%%          {stop, StopReason}
%%----------------------------------------------------------------------
init([Sid, Key, IP]) ->
    ?DEBUG("started: ~p", [{Sid, Key, IP}]),

    %% Read c2s options from the first ejabberd_c2s configuration in
    %% the config file listen section
    %% TODO: We should have different access and shaper values for
    %% each connector. The default behaviour should be however to use
    %% the default c2s restrictions if not defined for the current
    %% connector.
    Opts1 = ejabberd_c2s_config:get_c2s_limits(),
    Opts = [{xml_socket, true} | Opts1],

    Shaper = none,
    ShaperState = shaper:new(Shaper),
    Socket = {http_bind, self(), IP},
    ejabberd_socket:start(ejabberd_c2s, ?MODULE, Socket, Opts),
    Timer = erlang:start_timer(?MAX_INACTIVITY, self(), []),
    {ok, loop, #state{id = Sid,
		      key = Key,
		      socket = Socket,
		      shaper_state = ShaperState,
		      max_inactivity = ?MAX_INACTIVITY,
		      max_pause = ?MAX_PAUSE,
		      timer = Timer}}.

%%----------------------------------------------------------------------
%% Func: handle_event/3
%% Returns: {next_state, NextStateName, NextStateData}          |
%%          {next_state, NextStateName, NextStateData, Timeout} |
%%          {stop, Reason, NewStateData}
%%----------------------------------------------------------------------
handle_event({become_controller, C2SPid}, StateName, StateData) ->
    case StateData#state.input of
	cancel ->
	    {next_state, StateName, StateData#state{
				      waiting_input = C2SPid}};
	Input ->
	    lists:foreach(
	      fun(Event) ->
		      C2SPid ! Event
	      end, queue:to_list(Input)),
	    {next_state, StateName, StateData#state{
				      input = queue:new(),
				      waiting_input = C2SPid}}
    end;

handle_event({change_shaper, Shaper}, StateName, StateData) ->
    NewShaperState = shaper:new(Shaper),
    {next_state, StateName, StateData#state{shaper_state = NewShaperState}};
handle_event(_Event, StateName, StateData) ->
    {next_state, StateName, StateData}.

%%----------------------------------------------------------------------
%% Func: handle_sync_event/4
%% Returns: {next_state, NextStateName, NextStateData}            |
%%          {next_state, NextStateName, NextStateData, Timeout}   |
%%          {reply, Reply, NextStateName, NextStateData}          |
%%          {reply, Reply, NextStateName, NextStateData, Timeout} |
%%          {stop, Reason, NewStateData}                          |
%%          {stop, Reason, Reply, NewStateData}
%%----------------------------------------------------------------------
handle_sync_event({send_xml, Packet}, _From, StateName,
		  #state{http_receiver = undefined} = StateData) ->
    Output = [Packet | StateData#state.output],
    Reply = ok,
    {reply, Reply, StateName, StateData#state{output = Output}};
handle_sync_event({send_xml, Packet}, _From, StateName, StateData) ->
    Output = [Packet | StateData#state.output],
    cancel_timer(StateData#state.timer),
    Timer = set_inactivity_timer(StateData#state.pause,
				 StateData#state.max_inactivity),
    HTTPReply = {ok, Output},
    gen_fsm:reply(StateData#state.http_receiver, HTTPReply),
    cancel_timer(StateData#state.wait_timer),
    Rid = StateData#state.rid,
    ReqList = [#hbr{rid = Rid,
		    key = StateData#state.key,
		    out = Output
		   } |
	       [El || El <- StateData#state.req_list,
		      El#hbr.rid /= Rid ]
	      ],
    Reply = ok,
    {reply, Reply, StateName,
     StateData#state{output = [],
		     http_receiver = undefined,
		     req_list = ReqList,
		     wait_timer = undefined,
		     timer = Timer}};

handle_sync_event({stop,close}, _From, _StateName, StateData) ->
    Reply = ok,
    {stop, normal, Reply, StateData};
handle_sync_event({stop,stream_closed}, _From, _StateName, StateData) ->
    Reply = ok,
    {stop, normal, Reply, StateData};
handle_sync_event({stop,Reason}, _From, _StateName, StateData) ->
    ?ERROR_MSG("Closing bind session ~p - Reason: ~p", [StateData#state.id, Reason]),
    Reply = ok,
    {stop, normal, Reply, StateData};

%% HTTP PUT: Receive packets from the client
handle_sync_event(#http_put{rid = Rid},
		  _From, StateName, StateData)
  when StateData#state.shaper_timer /= undefined ->
    Pause =
	case erlang:read_timer(StateData#state.shaper_timer) of
	    false ->
		0;
	    P -> P
	end,
    Reply = {wait, Pause},
    ?DEBUG("Shaper timer for RID ~p: ~p", [Rid, Reply]),
    {reply, Reply, StateName, StateData};

handle_sync_event(#http_put{payload_size = PayloadSize} = Request,
		  _From, StateName, StateData) ->
    ?DEBUG("New request: ~p",[Request]),
    %% Updating trafic shaper
    {NewShaperState, NewShaperTimer} =
	update_shaper(StateData#state.shaper_state, PayloadSize),

    handle_http_put_event(Request, StateName,
			  StateData#state{shaper_state = NewShaperState,
					  shaper_timer = NewShaperTimer});

%% HTTP GET: send packets to the client
handle_sync_event({http_get, Rid, Wait, Hold}, From, StateName, StateData) ->
    %% setup timer
    send_receiver_reply(StateData#state.http_receiver, {ok, empty}),
    cancel_timer(StateData#state.wait_timer),
    TNow = tnow(),
    if
	(Hold > 0) and
	(StateData#state.output == []) and
	((TNow - StateData#state.ctime) < (Wait*1000*1000)) and
	(StateData#state.rid == Rid) and
	(StateData#state.input /= cancel) and
        (StateData#state.pause == 0) ->
	    WaitTimer = erlang:start_timer(Wait * 1000, self(), []),
	    %% MR: Not sure we should cancel the state timer here.
	    cancel_timer(StateData#state.timer),
	    {next_state, StateName, StateData#state{
				      http_receiver = From,
				      wait_timer = WaitTimer,
				      timer = undefined}};
	(StateData#state.input == cancel) ->
	    cancel_timer(StateData#state.timer),
	    Timer = set_inactivity_timer(StateData#state.pause,
					 StateData#state.max_inactivity),
	    Reply = {ok, cancel},
	    {reply, Reply, StateName, StateData#state{
					input = queue:new(),
					http_receiver = undefined,
					wait_timer = undefined,
					timer = Timer}};
	true ->
	    cancel_timer(StateData#state.timer),
	    Timer = set_inactivity_timer(StateData#state.pause,
					 StateData#state.max_inactivity),
	    Reply = {ok, StateData#state.output},
	    %% save request
	    ReqList = [#hbr{rid = Rid,
			    key = StateData#state.key,
			    out = StateData#state.output
			   } |
		       [El || El <- StateData#state.req_list,
			      El#hbr.rid /= Rid ]
		      ],
	    {reply, Reply, StateName, StateData#state{
					output = [],
					http_receiver = undefined,
					wait_timer = undefined,
					timer = Timer,
					req_list = ReqList}}
    end;

handle_sync_event(peername, _From, StateName, StateData) ->
    Reply = {ok, StateData#state.ip},
    {reply, Reply, StateName, StateData};

handle_sync_event(_Event, _From, StateName, StateData) ->
    Reply = ok,
    {reply, Reply, StateName, StateData}.

code_change(_OldVsn, StateName, StateData, _Extra) ->
    {ok, StateName, StateData}.

%%----------------------------------------------------------------------
%% Func: handle_info/3
%% Returns: {next_state, NextStateName, NextStateData}          |
%%          {next_state, NextStateName, NextStateData, Timeout} |
%%          {stop, Reason, NewStateData}
%%----------------------------------------------------------------------
%% We reached the max_inactivity timeout:
handle_info({timeout, Timer, _}, _StateName,
	    #state{id=SID, timer = Timer} = StateData) ->
    ?WARNING_MSG("Session timeout. Closing the HTTP bind session: ~p", [SID]),
    {stop, normal, StateData};

handle_info({timeout, WaitTimer, _}, StateName,
	    #state{wait_timer = WaitTimer} = StateData) ->
    if
	StateData#state.http_receiver /= undefined ->
	    cancel_timer(StateData#state.timer),
	    Timer = set_inactivity_timer(StateData#state.pause,
					 StateData#state.max_inactivity),
	    gen_fsm:reply(StateData#state.http_receiver, {ok, empty}),
	    Rid = StateData#state.rid,
	    ReqList = [#hbr{rid = Rid,
			    key = StateData#state.key,
			    out = []
			   } |
		       [El || El <- StateData#state.req_list,
			      El#hbr.rid /= Rid ]
		      ],
	    {next_state, StateName,
	     StateData#state{http_receiver = undefined,
			     req_list = ReqList,
			     wait_timer = undefined,
			     timer = Timer}};
	true ->
	    {next_state, StateName, StateData}
    end;

handle_info({timeout, ShaperTimer, _}, StateName,
	    #state{shaper_timer = ShaperTimer} = StateData) ->
    {next_state, StateName, StateData#state{shaper_timer = undefined}};

handle_info(_, StateName, StateData) ->
    {next_state, StateName, StateData}.

%%----------------------------------------------------------------------
%% Func: terminate/3
%% Purpose: Shutdown the fsm
%% Returns: any
%%----------------------------------------------------------------------
terminate(_Reason, _StateName, StateData) ->
    ?DEBUG("terminate: Deleting session ~s", [StateData#state.id]),
    mnesia:transaction(
      fun() ->
	      mnesia:delete({http_bind, StateData#state.id})
      end),
    send_receiver_reply(StateData#state.http_receiver, {ok, terminate}),
    case StateData#state.waiting_input of
	false ->
	    ok;
	C2SPid ->
	    gen_fsm:send_event(C2SPid, closed)
    end,
    ok.

%%%----------------------------------------------------------------------
%%% Internal functions
%%%----------------------------------------------------------------------

%% PUT / Get processing:
handle_http_put_event(#http_put{rid = Rid, attrs = Attrs,
				hold = Hold} = Request,
		      StateName, StateData) ->
    ?DEBUG("New request: ~p",[Request]),
    %% Check if Rid valid
    RidAllow = rid_allow(StateData#state.rid, Rid, Attrs, Hold,
			 StateData#state.max_pause),

    %% Check if Rid is in sequence or out of sequence:
    case RidAllow of
	buffer ->
	    ?DEBUG("Buffered request: ~p", [Request]),
	    %% Request is out of sequence:
	    PendingRequests = StateData#state.unprocessed_req_list,
	    %% In case an existing RID was already buffered:
	    Requests = lists:keydelete(Rid, 2, PendingRequests),
	    ReqList = [#hbr{rid = Rid,
			    key = StateData#state.key,
			    out = []
			   } |
		       [El || El <- StateData#state.req_list,
			      El#hbr.rid > (Rid - 1 - Hold)]
		      ],
	    ?DEBUG("reqlist: ~p", [ReqList]),
	    UnprocessedReqList = [Request | Requests],
	    cancel_timer(StateData#state.timer),
	    Timer = set_inactivity_timer(0, StateData#state.max_inactivity),
	    {reply, buffered, StateName,
	     StateData#state{unprocessed_req_list = UnprocessedReqList,
			     req_list = ReqList,
			     timer = Timer}};
	_ ->
	    %% Request is in sequence:
	    process_http_put(Request, StateName, StateData, RidAllow)
    end.

process_http_put(#http_put{rid = Rid, attrs = Attrs, payload = Payload,
			   hold = Hold,  stream = StreamTo,
			   ip = IP} = Request,
		 StateName, StateData, RidAllow) ->
    ?DEBUG("Actually processing request: ~p", [Request]),
    %% Check if key valid
    Key = xml:get_attr_s("key", Attrs),
    NewKey = xml:get_attr_s("newkey", Attrs),
    KeyAllow =
	case RidAllow of
	    repeat ->
		true;
	    false ->
		false;
	    {true, _} ->
		case StateData#state.key of
		    "" ->
			true;
		    OldKey ->
			NextKey = sha:sha(Key),
			?DEBUG("Key/OldKey/NextKey: ~s/~s/~s", [Key, OldKey, NextKey]),
			if
			    OldKey == NextKey ->
				true;
			    true ->
				?DEBUG("wrong key: ~s",[Key]),
				false
			end
		end
	end,
    TNow = tnow(),
    LastPoll = if
		   Payload == [] ->
		       TNow;
		   true ->
		       0
	       end,
    if
	(Payload == []) and
        (Hold == 0) and
	(TNow - StateData#state.last_poll < ?MIN_POLLING) ->
	    Reply = {error, polling_too_frequently},
	    {reply, Reply, StateName, StateData};
	KeyAllow ->
	    case RidAllow of
		false ->
		    Reply = {error, not_exists},
		    {reply, Reply, StateName, StateData};
		repeat ->
		    ?DEBUG("REPEATING ~p", [Rid]),
		    Reply = case [El#hbr.out ||
				     El <- StateData#state.req_list,
				     El#hbr.rid == Rid] of
				[] ->
				    {error, not_exists};
				[Out | _XS] ->
				    {repeat, lists:reverse(Out)}
			    end,
		    {reply, Reply, StateName, StateData#state{input = cancel,
							      last_poll = LastPoll}};
		{true, Pause} ->
		    SaveKey = if
				  NewKey == "" ->
				      Key;
				  true ->
				      NewKey
			      end,
		    ?DEBUG(" -- SaveKey: ~s~n", [SaveKey]),

		    %% save request
		    ReqList1 =
			[El || El <- StateData#state.req_list,
			       El#hbr.rid > (Rid - 1 - Hold)],
		    ReqList =
			case lists:keymember(Rid, #hbr.rid, ReqList1) of
			    true ->
				ReqList1;
			    false ->
				[#hbr{rid = Rid,
				      key = StateData#state.key,
				      out = []
				     } |
				 ReqList1
				]
			end,
		    ?DEBUG("reqlist: ~p", [ReqList]),

                    %% setup next timer
		    cancel_timer(StateData#state.timer),
		    Timer = set_inactivity_timer(Pause,
						 StateData#state.max_inactivity),
		    case StateData#state.waiting_input of
			false ->
			    Input =
				lists:foldl(
				  fun queue:in/2,
				  StateData#state.input, Payload),
			    Reply = ok,
			    process_buffered_request(Reply, StateName,
						     StateData#state{input = Input,
								     rid = Rid,
								     key = SaveKey,
								     ctime = TNow,
								     timer = Timer,
								     pause = Pause,
								     last_poll = LastPoll,
								     req_list = ReqList,
								     ip = IP
								    });
			C2SPid ->
			    case StreamTo of
				{To, ""} ->
				    gen_fsm:send_event(
				      C2SPid,
				      {xmlstreamstart, "stream:stream",
				       [{"to", To},
					{"xmlns", ?NS_CLIENT},
					{"xmlns:stream", ?NS_STREAM}]});
				{To, Version} ->
				    gen_fsm:send_event(
				      C2SPid,
				      {xmlstreamstart, "stream:stream",
				       [{"to", To},
					{"xmlns", ?NS_CLIENT},
					{"version", Version},
					{"xmlns:stream", ?NS_STREAM}]});
				_ ->
				    ok
			    end,

                            MaxInactivity = get_max_inactivity(StreamTo, StateData#state.max_inactivity),
                            MaxPause = get_max_inactivity(StreamTo, StateData#state.max_pause),

			    ?DEBUG("really sending now: ~p", [Payload]),
			    lists:foreach(
			      fun({xmlstreamend, End}) ->
				      gen_fsm:send_event(
					C2SPid, {xmlstreamend, End});
				 (El) ->
				      gen_fsm:send_event(
					C2SPid, {xmlstreamelement, El})
			      end, Payload),
			    Reply = ok,
			    process_buffered_request(Reply, StateName,
						     StateData#state{input = queue:new(),
								     rid = Rid,
								     key = SaveKey,
								     ctime = TNow,
								     timer = Timer,
								     pause = Pause,
								     last_poll = LastPoll,
								     req_list = ReqList,
								     max_inactivity = MaxInactivity,
								     max_pause = MaxPause,
								     ip = IP
								    })
		    end
	    end;
	true ->
	    Reply = {error, bad_key},
	    {reply, Reply, StateName, StateData}
    end.

process_buffered_request(Reply, StateName, StateData) ->
    Rid = StateData#state.rid,
    Requests = StateData#state.unprocessed_req_list,
    case lists:keysearch(Rid+1, 2, Requests) of
	{value, Request} ->
	    ?DEBUG("Processing buffered request: ~p", [Request]),
	    NewRequests = lists:keydelete(Rid+1, 2, Requests),
	    handle_http_put_event(
	      Request, StateName,
	      StateData#state{unprocessed_req_list = NewRequests});
	_ ->
	    {reply, Reply, StateName, StateData}
    end.

handle_http_put(Sid, Rid, Attrs, Payload, PayloadSize, StreamStart, IP) ->
    case http_put(Sid, Rid, Attrs, Payload, PayloadSize, StreamStart, IP) of
        {error, not_exists} ->
            ?ERROR_MSG("no session associated with sid: ~p", [Sid]),
            {404, ?HEADER, ""};
        {{error, Reason}, Sess} ->
            ?ERROR_MSG("Error on HTTP put. Reason: ~p", [Reason]),
            handle_http_put_error(Reason, Sess);
        {{repeat, OutPacket}, Sess} ->
            ?DEBUG("http_put said 'repeat!' ...~nOutPacket: ~p", [OutPacket]),
            send_outpacket(Sess, OutPacket);
        {{wait, Pause}, _Sess} ->
	    ?DEBUG("Trafic Shaper: Delaying request ~p", [Rid]),
	    timer:sleep(Pause),
            %{200, ?HEADER,
            % xml:element_to_string(
            %   {xmlelement, "body",
            %    [{"xmlns", ?NS_HTTP_BIND},
            %     {"type", "error"}], []})};
            handle_http_put(Sid, Rid, Attrs, Payload, PayloadSize,
			    StreamStart, IP);
        {buffered, _Sess} ->
            {200, ?HEADER, "<body xmlns='"++?NS_HTTP_BIND++"'/>"};
        {ok, Sess} ->
            prepare_response(Sess, Rid, Attrs, StreamStart)
    end.

http_put(Sid, Rid, Attrs, Payload, PayloadSize, StreamStart, IP) ->
    ?DEBUG("Looking for session: ~p", [Sid]),
    case mnesia:dirty_read({http_bind, Sid}) of
	[] ->
            {error, not_exists};
	[#http_bind{pid = FsmRef, hold=Hold, to={To, StreamVersion}}=Sess] ->
            NewStream =
                case StreamStart of
                    true ->
                        {To, StreamVersion};
                    _ ->
                        ""
                end,
            {gen_fsm:sync_send_all_state_event(
               FsmRef, #http_put{rid = Rid, attrs = Attrs, payload = Payload,
				 payload_size = PayloadSize, hold = Hold,
				 stream = NewStream, ip = IP}, 30000), Sess}
    end.

handle_http_put_error(Reason, #http_bind{pid=FsmRef, version=Version})
  when Version >= 0 ->
    gen_fsm:sync_send_all_state_event(FsmRef, {stop, {put_error,Reason}}),
    case Reason of
        not_exists ->
            {200, ?HEADER,
             xml:element_to_string(
               {xmlelement, "body",
                [{"xmlns", ?NS_HTTP_BIND},
                 {"type", "terminate"},
                 {"condition", "item-not-found"}], []})};
        bad_key ->
            {200, ?HEADER,
             xml:element_to_string(
               {xmlelement, "body",
                [{"xmlns", ?NS_HTTP_BIND},
                 {"type", "terminate"},
                 {"condition", "item-not-found"}], []})};
        polling_too_frequently ->
            {200, ?HEADER,
             xml:element_to_string(
               {xmlelement, "body",
                [{"xmlns", ?NS_HTTP_BIND},
                 {"type", "terminate"},
                 {"condition", "policy-violation"}], []})}
    end;
handle_http_put_error(Reason, #http_bind{pid=FsmRef}) ->
    gen_fsm:sync_send_all_state_event(FsmRef,{stop, {put_error_no_version, Reason}}),
    case Reason of
        not_exists -> %% bad rid
	    ?ERROR_MSG("Closing HTTP bind session (Bad rid).", []),
            {404, ?HEADER, ""};
        bad_key ->
	    ?ERROR_MSG("Closing HTTP bind session (Bad key).", []),
            {404, ?HEADER, ""};
        polling_too_frequently ->
	    ?ERROR_MSG("Closing HTTP bind session (User polling too frequently).", []),
            {403, ?HEADER, ""}
    end.

%% Control RID ordering
rid_allow(none, _NewRid, _Attrs, _Hold, _MaxPause) ->
    %% First request - nothing saved so far
    {true, 0};
rid_allow(OldRid, NewRid, Attrs, Hold, MaxPause) ->
    ?DEBUG("Previous rid / New rid: ~p/~p", [OldRid, NewRid]),
    if
	%% We did not miss any packet, we can process it immediately:
	NewRid == OldRid + 1 ->
	    case catch list_to_integer(
			 xml:get_attr_s("pause", Attrs)) of
		{'EXIT', _} ->
		    {true, 0};
		Pause1 when Pause1 =< MaxPause ->
		    ?DEBUG("got pause: ~p", [Pause1]),
		    {true, Pause1};
		_ ->
		    {true, 0}
	    end;
	%% We have missed packets, we need to cached it to process it later on:
	(OldRid < NewRid) and
	(NewRid =< (OldRid + Hold + 1)) ->
	    buffer;
	(NewRid =< OldRid) and
	(NewRid > OldRid - Hold - 1) ->
	    repeat;
	true ->
	    false
    end.

update_shaper(ShaperState, PayloadSize) ->
    {NewShaperState, Pause} = shaper:update(ShaperState, PayloadSize),
    if
	Pause > 0 ->
	    ShaperTimer = erlang:start_timer(Pause, self(), activate), %% MR: Seems timer is not needed. Activate is not handled
	    {NewShaperState, ShaperTimer};
	true ->
	    {NewShaperState, undefined}
    end.

prepare_response(#http_bind{id=Sid, wait=Wait, hold=Hold, to=To}=Sess,
                 Rid, _, StreamStart) ->
    receive after 100 -> ok end, %% TODO: Why is this needed. Argh. Bad programming practice.
    case catch http_get(Sess, Rid) of
	{ok, cancel} ->
	    %% actually it would be better if we could completely
	    %% cancel this request, but then we would have to hack
	    %% ejabberd_http and I'm too lazy now
            {200, ?HEADER, "<body type='error' xmlns='"++?NS_HTTP_BIND++"'/>"};
	{ok, empty} ->
            {200, ?HEADER, "<body xmlns='"++?NS_HTTP_BIND++"'/>"};
	{ok, terminate} ->
            {200, ?HEADER, "<body type='terminate' xmlns='"++?NS_HTTP_BIND++"'/>"};
	{ok, ROutPacket} ->
	    OutPacket = lists:reverse(ROutPacket),
            ?DEBUG("OutPacket: ~p", [OutPacket]),
	    case StreamStart of
                false ->
		    send_outpacket(Sess, OutPacket);
		true ->
		    case OutPacket of
			[{xmlstreamstart, _, OutAttrs} | Els] ->
			    AuthID = xml:get_attr_s("id", OutAttrs),
			    From = xml:get_attr_s("from", OutAttrs),
			    Version = xml:get_attr_s("version", OutAttrs),
			    OutEls =
				case Els of
				    [] ->
					[];
				    [{xmlstreamelement,
				      {xmlelement, "stream:features",
				       StreamAttribs, StreamEls}}
				     | StreamTail] ->
					TypedTail =
					    [check_default_xmlns(OEl) ||
						{xmlstreamelement, OEl} <-
						    StreamTail],
					[{xmlelement, "stream:features",
					  [{"xmlns:stream",
					    ?NS_STREAM}] ++
					  StreamAttribs, StreamEls}] ++
					    TypedTail;
				    StreamTail ->
					[check_default_xmlns(OEl) ||
					    {xmlstreamelement, OEl} <-
						StreamTail]
				end,
			    BOSH_attribs =
				[{"authid", AuthID},
				 {"xmlns:xmpp", ?NS_BOSH},
				 {"xmlns:stream", ?NS_STREAM}] ++
				case OutEls of
				    [] ->
					[];
				    _ ->
					[{"xmpp:version", Version}]
				end,
			    MaxInactivity = get_max_inactivity(To, ?MAX_INACTIVITY),
			    MaxPause = get_max_pause(To),
			    {200, ?HEADER,
			     xml:element_to_string(
			       {xmlelement,"body",
				[{"xmlns",
				  ?NS_HTTP_BIND},
				 {"sid", Sid},
				 {"wait", integer_to_list(Wait)},
				 {"requests", integer_to_list(Hold+1)},
				 {"inactivity",
				  integer_to_list(
				    trunc(MaxInactivity/1000))},
				 {"maxpause",
				  integer_to_list(MaxPause)},
				 {"polling",
				  integer_to_list(
				    trunc(?MIN_POLLING/1000000))},
				 {"ver", ?BOSH_VERSION},
				 {"from", From},
				 {"secure", "true"} %% we're always being secure
				] ++ BOSH_attribs,OutEls})};
			{error, _} ->
			    {200, ?HEADER, "<body type='terminate' "
			     "condition='host-unknown' "
			     "xmlns='"++?NS_HTTP_BIND++"'/>"}
		    end
	    end;
	{'EXIT', {shutdown, _}} ->
            {200, ?HEADER, "<body type='terminate' condition='system-shutdown' xmlns='"++?NS_HTTP_BIND++"'/>"};
	{'EXIT', _Reason} ->
            {200, ?HEADER, "<body type='terminate' xmlns='"++?NS_HTTP_BIND++"'/>"}
    end.

http_get(#http_bind{pid = FsmRef, wait = Wait, hold = Hold}, Rid) ->
    gen_fsm:sync_send_all_state_event(
      FsmRef, {http_get, Rid, Wait, Hold}, 2 * ?MAX_WAIT * 1000).

send_outpacket(#http_bind{pid = FsmRef}, OutPacket) ->
    case OutPacket of
	[] ->
	    {200, ?HEADER, "<body xmlns='"++?NS_HTTP_BIND++"'/>"};
	[{xmlstreamend, _}] ->
            gen_fsm:sync_send_all_state_event(FsmRef,{stop,stream_closed}),
	    {200, ?HEADER, "<body xmlns='"++?NS_HTTP_BIND++"'/>"};
	_ ->
	    %% TODO: We parse to add a default namespace to packet,
	    %% The spec says adding the jabber:client namespace if
	    %% mandatory, even if some implementation do not do that
	    %% change on packets.
	    %% I think this should be an option to avoid modifying
	    %% packet in most case.
	    AllElements =
		lists:all(fun({xmlstreamelement,
			       {xmlelement, "stream:error", _, _}}) -> false;
			     ({xmlstreamelement, _}) -> true;
			     (_) -> false
			  end, OutPacket),
	    case AllElements of
		true ->
		    TypedEls = [check_default_xmlns(OEl) ||
				   {xmlstreamelement, OEl} <- OutPacket],
		    Body = xml:element_to_string(
			     {xmlelement,"body",
			      [{"xmlns",
				?NS_HTTP_BIND}],
			      TypedEls}),
		    ?DEBUG(" --- outgoing data --- ~n~s~n --- END --- ~n",
			   [Body]),
		    {200, ?HEADER, Body};
		false ->
		    case OutPacket of
			[{xmlstreamstart, _, _} | SEls] ->
			    OutEls =
				case SEls of
				    [{xmlstreamelement,
				      {xmlelement,
				       "stream:features",
				       StreamAttribs, StreamEls}} |
				     StreamTail] ->
					TypedTail =
					    [check_default_xmlns(OEl) ||
						{xmlstreamelement, OEl} <-
						    StreamTail],
					[{xmlelement,
					  "stream:features",
					  [{"xmlns:stream",
					    ?NS_STREAM}] ++
					  StreamAttribs, StreamEls}] ++
					    TypedTail;
				    StreamTail ->
					[check_default_xmlns(OEl) ||
					    {xmlstreamelement, OEl} <-
						StreamTail]
				end,
                            {200, ?HEADER,
                             xml:element_to_string(
                               {xmlelement,"body",
                                [{"xmlns",
                                  ?NS_HTTP_BIND}],
                                OutEls})};
			_ ->
			    SErrCond =
				lists:filter(
				  fun({xmlstreamelement,
				       {xmlelement, "stream:error",
					_, _}}) ->
					  true;
				     (_) -> false
				  end, OutPacket),
                            StreamErrCond =
				case SErrCond of
				    [] ->
					null;
				    [{xmlstreamelement,
				      {xmlelement, _, _, _Cond} =
				      StreamErrorTag} | _] ->
					[StreamErrorTag]
				end,
                            gen_fsm:sync_send_all_state_event(FsmRef,
                                                              {stop, {stream_error,OutPacket}}),
                            case StreamErrCond of
                                null ->
                                    {200, ?HEADER,
                                     "<body type='terminate' "
                                     "condition='internal-server-error' "
                                     "xmlns='"++?NS_HTTP_BIND++"'/>"};
                                _ ->
                                    {200, ?HEADER,
                                     "<body type='terminate' "
                                     "condition='remote-stream-error' "
                                     "xmlns='"++?NS_HTTP_BIND++"' " ++
                                     "xmlns:stream='"++?NS_STREAM++"'>" ++
                                     elements_to_string(StreamErrCond) ++
                                     "</body>"}
                            end
		    end
	    end
    end.

parse_request(Data, PayloadSize, MaxStanzaSize) ->
    ?DEBUG("--- incoming data --- ~n~s~n --- END --- ", [Data]),
    %% MR: I do not think it works if put put several elements in the
    %% same body:
    case xml_stream:parse_element(Data) of
	{xmlelement, "body", Attrs, Els} ->
	    Xmlns = xml:get_attr_s("xmlns",Attrs),
	    if
		Xmlns /= ?NS_HTTP_BIND ->
		    {error, bad_request};
		true ->
                    case catch list_to_integer(xml:get_attr_s("rid", Attrs)) of
                        {'EXIT', _} ->
                            {error, bad_request};
                        Rid ->
			    %% I guess this is to remove XMLCDATA: Is it really needed ?
                            FixedEls =
                                lists:filter(
                                  fun(I) ->
                                          case I of
                                              {xmlelement, _, _, _} ->
                                                  true;
                                              _ ->
                                                  false
                                          end
                                  end, Els),
                            Sid = xml:get_attr_s("sid",Attrs),
			    if
				PayloadSize =< MaxStanzaSize ->
				    {ok, {Sid, Rid, Attrs, FixedEls}};
				true ->
				    {size_limit, Sid}
			    end
                    end
	    end;
	{xmlelement, _Name, _Attrs, _Els} ->
	    {error, bad_request};
	{error, _Reason} ->
	    {error, bad_request}
    end.

send_receiver_reply(undefined, _Reply) ->
    ok;
send_receiver_reply(Receiver, Reply) ->
    gen_fsm:reply(Receiver, Reply).


%% Cancel timer and empty message queue.
cancel_timer(undefined) ->
    ok;
cancel_timer(Timer) ->
    erlang:cancel_timer(Timer),
    receive
	{timeout, Timer, _} ->
	    ok
    after 0 ->
	    ok
    end.

%% If client asked for a pause (pause > 0), we apply the pause value
%% as inactivity timer:
set_inactivity_timer(Pause, _MaxInactivity) when Pause > 0 ->
    erlang:start_timer(Pause*1000, self(), []);
%% Otherwise, we apply the max_inactivity value as inactivity timer:
set_inactivity_timer(_Pause, MaxInactivity) ->
    erlang:start_timer(MaxInactivity, self(), []).


%% TODO: Use tail recursion and list reverse ?
elements_to_string([]) ->
    [];
elements_to_string([El | Els]) ->
    xml:element_to_string(El) ++ elements_to_string(Els).

%% @spec (To, Default::integer()) -> integer()
%% where To = [] | {Host::string(), Version::string()}
get_max_inactivity({Host, _}, Default) ->
    case gen_mod:get_module_opt(Host, mod_http_bind, max_inactivity, undefined) of
	Seconds when is_integer(Seconds) ->
	    Seconds * 1000;
	undefined ->
	    Default
    end;
get_max_inactivity(_, Default) ->
    Default.

get_max_pause({Host, _}) ->
    gen_mod:get_module_opt(Host, mod_http_bind, max_pause, ?MAX_PAUSE);
get_max_pause(_) ->
    ?MAX_PAUSE.

%% Current time as integer
tnow() ->
    {TMegSec, TSec, TMSec} = now(),
    (TMegSec * 1000000 + TSec) * 1000000 + TMSec.

check_default_xmlns({xmlelement, Name, Attrs, Els} = El) ->
    case xml:get_tag_attr_s("xmlns", El) of
	"" -> {xmlelement, Name, [{"xmlns", ?NS_CLIENT} | Attrs], Els};
	_  -> El
    end.

%% Check that mod_http_bind has been defined in config file.
%% 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;
	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.",[])
    end.