正文
ejabberd中的http反向推送
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
http的反向推送通常使用"长轮询"或"长连接"的方式。 所谓"长轮询"是指客户端发送请求给服务器，服务器发现没有数据需要发送给客户端。
http的反向推送通常使用"长轮询"或"长连接"的方式。
所谓"长轮询"是指客户端发送请求给服务器，服务器发现没有数据需要发送给客户端于是hold住不及时返回，等有数据需要发送给客户端时，进行回复，然后关闭连接，客户端收到回复后再发送新的http请求，以便服务器能有对应的请求用于消息的反向推送。
而"长连接"是在长轮询的基础上增加"keep-alive"属性，服务器收到请求后不直接回复，等有数据需要发送给客户端时再进行response，但是并不关闭连接，这样客户端收到服务器的response后在同一连接上再次发送http请求。
|
在ejabberd的实现中，采用了bosh技术来完成对应的工作，具体定义可参考： 英文: http://go.rritw.com/xmpp.org/extensions/xep-0124.html 中文: http://go.rritw.com/wiki.jabbercn.org/XEP-0124 大概实现原理：ejabberd收到一个客户端http请求后会为该客户端最终创建三个进程：ejabberd_http, ejabberd_http_bind, ejabberd_c2s。 ejabberd_http进程不断的从对应的socket上收客户端的请求，并转发交给对应的ejabberd_http_bind进程进行处理，然后同步等待处理结果，并将结果返回给客户端。 |
init() ->
...
receive_headers(State). receive_headers(#state{trail=Trail} = State) ->
SockMod = State#state.sockmod,
Socket = State#state.socket,
Data = SockMod:recv(Socket, 0, 300000),
case State#state.sockmod of
gen_tcp ->
NewState = process_header(State, Data),
case NewState#state.end_of_request of
true -> ok;
_ -> receive_headers(NewState)
end;
_ ->
case Data of
{ok, D} ->
parse_headers(State#state{trail = <<Trail/binary, D/binary>>});
{error, _} ->
ok
end
end. process_header(State, Data) ->
case Data of
...
{ok, http_eoh} ->
...
Out = process_request(State2),
send_text(State2, Out),
case State2#state.request_keepalive of
true ->
...
#state{sockmod = SockMod,
socket = Socket,
request_handlers = State#state.request_handlers};
_ ->
#state{end_of_request = true,
request_handlers = State#state.request_handlers}
从代码中可看出，未设置keep-alive属性的时候，该进程处理完一次http请求后便自己结束（长轮询模式）。设置了keep-alive属性的时候，该进程不断的循环接收http请求，并转发接收与响应（长连接模式）。
ejabberd_http_bind进程负责hold住http请求，对于正常的客户端请求，ejabberd_http_bind进程会将请求转发给对应的ejabberd_c2s进程进行实际业务的处理，而对于空的请求（便于服务器反向推送数据），ejabberd_http_bind设置定时器，等待ejabberd_c2s进程对实际请求的响应或者是需要推送给客户端的消息。
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({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_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;
ejabberd_http进程最终会调用ejabberd_http_bind的http_get方法获取请求的响应结果或者是需要推送的数据，ejabberd_http_bind进程收到请求后会进行相应处理，比如有数据则直接回复，或者设置定时器。当收到ejabberd_c2s进程推送过来的数据时，停止定时器并将数据立即回复给ejabberd_http进程，如果定时器超时则回复一个空的消息。
=============================
ejabberd_c2s为客户端对应的会话进程，负责维护客户端的在线状态，联系人列表，请求处理等等。
============
服务器在创建ejabberd_http_bind进程时，会生成一个唯一的sid，用于标识该进程，该sid与ejabberd_http_bind进程pid的对应关系会存储到mnesia中，同时服务器会将sid也会告诉客户端，客户端后续的请求也都需要带上该sid。http_bind收到请求后根据sid从mnesia查找匹配的ejabberd_http_bind进程。
process_request(Data, IP) ->
...
case catch parse_request(Data, PayloadSize, MaxStanzaSize) of
%% No existing session:
{ok, {"", Rid, Attrs, Payload}} ->
...
Sid = sha:sha(term_to_binary({now(), make_ref()})),
case start(XmppDomain, Sid, "", IP) of
{ok, Pid} ->
handle_session_start(Pid, XmppDomain, Sid, Rid, Attrs,Payload, PayloadSize, IP)
...
end;
{ok, {Sid, Rid, Attrs, Payload1}} ->
...
handle_http_put(Sid, Rid, Attrs, Payload2, PayloadSize,
StreamStart, IP); handle_session_start(Pid, XmppDomain, Sid, Rid, Attrs,
Payload, PayloadSize, IP) ->
...
mnesia:dirty_write(#http_bind{id = Sid,
pid = Pid,
to = {XmppDomain,XmppVersion},
hold = Hold,
wait = Wait,
process_delay = Pdelay,
version = Version}),
handle_http_put(Sid, Rid, Attrs, Payload, PayloadSize, true, IP). http_put(Sid, Rid, Attrs, Payload, PayloadSize, StreamStart, IP) ->
case mnesia:dirty_read({http_bind, Sid}) of
[] ->
{error, not_exists};
[#http_bind{pid = FsmRef, hold=Hold, to={To, StreamVersion}}=Sess] ->
...
{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.
ejabberd_http_bind与ejabberd_c2s会互相记住对方的进程pid，这样每个客户端都有自己唯一的ejabberd_http_bind与ejabberd_c2s进程进行相应的请求处理。【转】 http://my.oschina.net/hncscwc/blog/178471
