yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Improve app src (#2114) 

* Add replayq in emqx.app.src

* Fix publish_readq_msg badmatch
This commit is contained in:
Gilbert 2018-12-28 20:04:52 +08:00 committed by GitHub
parent eed03ee297
commit 72791b569e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 4 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/emqx_bridge.erl
View File

@ -448,12 +448,11 @@ en_writeq(Msg, State = #state{writeq = WriteQ, replayq = ReplayQ,
NewReplayQ =replayq:append(ReplayQ, lists:reverse(WriteQ)), NewReplayQ =replayq:append(ReplayQ, lists:reverse(WriteQ)),
State#state{writeq = [Msg], replayq = NewReplayQ}. State#state{writeq = [Msg], replayq = NewReplayQ}.
publish_readq_msg(_ClientPid, [], ReadQ) -> publish_readq_msg(_ClientPid, [], NewReadQ) ->
{ok, ReadQ}; {ok, NewReadQ};
publish_readq_msg(ClientPid, [{_PktId, Msg} | ReadQ], ReadQ) -> publish_readq_msg(ClientPid, [{_PktId, Msg} | ReadQ], NewReadQ) ->
io:format("~n replay msg: ~p ~n", [Msg]),
{ok, PktId} = emqx_client:publish(ClientPid, Msg), {ok, PktId} = emqx_client:publish(ClientPid, Msg),
publish_readq_msg(ClientPid, ReadQ, [{PktId, Msg} | ReadQ]). publish_readq_msg(ClientPid, ReadQ, [{PktId, Msg} | NewReadQ]).
delete(_PktId, State = #state{readq = [], writeq = [], replayq = ReplayQ, ackref = AckRef}) -> delete(_PktId, State = #state{readq = [], writeq = [], replayq = ReplayQ, ackref = AckRef}) ->
ok = replayq:ack(ReplayQ, AckRef), ok = replayq:ack(ReplayQ, AckRef),