1194 lines
36 KiB
Erlang
1194 lines
36 KiB
Erlang
%%--------------------------------------------------------------------
|
|
%% Copyright (c) 2020-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
%%
|
|
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
%% you may not use this file except in compliance with the License.
|
|
%% You may obtain a copy of the License at
|
|
%%
|
|
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
%%
|
|
%% Unless required by applicable law or agreed to in writing, software
|
|
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
%% See the License for the specific language governing permissions and
|
|
%% limitations under the License.
|
|
%%--------------------------------------------------------------------
|
|
|
|
-module(emqx_stomp_SUITE).
|
|
|
|
-include_lib("eunit/include/eunit.hrl").
|
|
-include_lib("common_test/include/ct.hrl").
|
|
-include("emqx_stomp.hrl").
|
|
|
|
-compile(export_all).
|
|
-compile(nowarn_export_all).
|
|
|
|
-import(
|
|
emqx_gateway_test_utils,
|
|
[
|
|
assert_fields_exist/2,
|
|
request/2,
|
|
request/3
|
|
]
|
|
).
|
|
|
|
-define(HEARTBEAT, <<$\n>>).
|
|
|
|
-define(CONF_DEFAULT, <<
|
|
"\n"
|
|
"gateway.stomp {\n"
|
|
" clientinfo_override {\n"
|
|
" username = \"${Packet.headers.login}\"\n"
|
|
" password = \"${Packet.headers.passcode}\"\n"
|
|
" }\n"
|
|
" frame {\n"
|
|
" max_headers = 10\n"
|
|
" max_headers_length = 100\n"
|
|
" max_body_length = 1024\n"
|
|
" }\n"
|
|
" listeners.tcp.default {\n"
|
|
" bind = 61613\n"
|
|
" }\n"
|
|
"}\n"
|
|
>>).
|
|
|
|
all() -> emqx_common_test_helpers:all(?MODULE).
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% Setups
|
|
%%--------------------------------------------------------------------
|
|
|
|
init_per_suite(Config) ->
|
|
application:load(emqx_gateway_stomp),
|
|
Apps = emqx_cth_suite:start(
|
|
[
|
|
{emqx_conf, ?CONF_DEFAULT},
|
|
emqx_gateway,
|
|
emqx_auth,
|
|
emqx_management,
|
|
{emqx_dashboard, "dashboard.listeners.http { enable = true, bind = 18083 }"}
|
|
],
|
|
#{work_dir => emqx_cth_suite:work_dir(Config)}
|
|
),
|
|
emqx_common_test_http:create_default_app(),
|
|
[{suite_apps, Apps} | Config].
|
|
|
|
end_per_suite(Config) ->
|
|
emqx_common_test_http:delete_default_app(),
|
|
emqx_cth_suite:stop(?config(suite_apps, Config)),
|
|
ok.
|
|
|
|
default_config() ->
|
|
?CONF_DEFAULT.
|
|
|
|
stomp_ver() ->
|
|
?STOMP_VER.
|
|
|
|
restart_stomp_with_mountpoint(Mountpoint) ->
|
|
Conf = emqx:get_raw_config([gateway, stomp]),
|
|
emqx_gateway_conf:update_gateway(
|
|
stomp,
|
|
Conf#{<<"mountpoint">> => Mountpoint}
|
|
).
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% Test Cases
|
|
%%--------------------------------------------------------------------
|
|
|
|
t_connect(_) ->
|
|
%% Successful connect
|
|
ConnectSucced = fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"guest">>, <<"guest">>, <<"1000,2000">>),
|
|
{ok, Frame} = recv_a_frame(Sock),
|
|
?assertMatch(<<"CONNECTED">>, Frame#stomp_frame.command),
|
|
?assertEqual(
|
|
<<"2000,1000">>, proplists:get_value(<<"heart-beat">>, Frame#stomp_frame.headers)
|
|
),
|
|
|
|
ok = send_disconnect_frame(Sock, <<"12345">>),
|
|
?assertMatch(
|
|
{ok, #stomp_frame{
|
|
command = <<"RECEIPT">>,
|
|
headers = [{<<"receipt-id">>, <<"12345">>}]
|
|
}},
|
|
recv_a_frame(Sock)
|
|
)
|
|
end,
|
|
with_connection(ConnectSucced),
|
|
|
|
%% Connect will be failed, because of bad version
|
|
ProtocolError = fun(Sock) ->
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"CONNECT">>,
|
|
[
|
|
{<<"accept-version">>, <<"2.0,2.1">>},
|
|
{<<"host">>, <<"127.0.0.1:61613">>},
|
|
{<<"login">>, <<"guest">>},
|
|
{<<"passcode">>, <<"guest">>},
|
|
{<<"heart-beat">>, <<"1000,2000">>}
|
|
]
|
|
)
|
|
),
|
|
{ok, Data} = gen_tcp:recv(Sock, 0),
|
|
{ok, Frame, _, _} = parse(Data),
|
|
#stomp_frame{
|
|
command = <<"ERROR">>,
|
|
headers = _,
|
|
body = <<"Login Failed: Supported protocol versions < 1.2">>
|
|
} = Frame
|
|
end,
|
|
with_connection(ProtocolError).
|
|
|
|
t_heartbeat(_) ->
|
|
%% Test heart beat
|
|
with_connection(fun(Sock) ->
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"CONNECT">>,
|
|
[
|
|
{<<"accept-version">>, ?STOMP_VER},
|
|
{<<"host">>, <<"127.0.0.1:61613">>},
|
|
{<<"login">>, <<"guest">>},
|
|
{<<"passcode">>, <<"guest">>},
|
|
{<<"heart-beat">>, <<"1000,800">>}
|
|
]
|
|
)
|
|
),
|
|
{ok, Data} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"CONNECTED">>,
|
|
headers = _,
|
|
body = _
|
|
},
|
|
_, _} = parse(Data),
|
|
|
|
{ok, ?HEARTBEAT} = gen_tcp:recv(Sock, 0),
|
|
%% Server will close the connection because never receive the heart beat from client
|
|
{error, closed} = gen_tcp:recv(Sock, 0)
|
|
end).
|
|
|
|
t_subscribe(_) ->
|
|
with_connection(fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"guest">>, <<"guest">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
|
|
ok = send_subscribe_frame(Sock, 0, <<"/queue/foo">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
|
|
%% 'user-defined' header will be retain
|
|
ok = send_message_frame(Sock, <<"/queue/foo">>, <<"hello">>, [
|
|
{<<"user-defined">>, <<"emq">>},
|
|
{<<"content-type">>, <<"text/html">>}
|
|
]),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
|
|
{ok, Frame} = recv_a_frame(Sock),
|
|
?assertEqual(
|
|
<<"text/html">>, proplists:get_value(<<"content-type">>, Frame#stomp_frame.headers)
|
|
),
|
|
|
|
?assertMatch(
|
|
#stomp_frame{
|
|
command = <<"MESSAGE">>,
|
|
headers = _,
|
|
body = <<"hello">>
|
|
},
|
|
Frame
|
|
),
|
|
lists:foreach(
|
|
fun({Key, Val}) ->
|
|
Val = proplists:get_value(Key, Frame#stomp_frame.headers)
|
|
end,
|
|
[
|
|
{<<"destination">>, <<"/queue/foo">>},
|
|
{<<"subscription">>, <<"0">>},
|
|
{<<"user-defined">>, <<"emq">>}
|
|
]
|
|
),
|
|
|
|
%% assert subscription stats
|
|
[ClientInfo1] = clients(),
|
|
?assertMatch(#{subscriptions_cnt := 1}, ClientInfo1),
|
|
|
|
%% Unsubscribe
|
|
ok = send_unsubscribe_frame(Sock, 0),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
|
|
%% assert subscription stats
|
|
[ClientInfo2] = clients(),
|
|
?assertMatch(#{subscriptions_cnt := 0}, ClientInfo2),
|
|
|
|
ok = send_message_frame(Sock, <<"/queue/foo">>, <<"You will not receive this msg">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
|
|
{error, timeout} = gen_tcp:recv(Sock, 0, 500)
|
|
end).
|
|
|
|
t_subscribe_inuse(_) ->
|
|
UsedTopic = <<"/queue/foo">>,
|
|
UsedSubId = <<"0">>,
|
|
Setup =
|
|
fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"guest">>, <<"guest">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
ok = send_subscribe_frame(Sock, UsedSubId, UsedTopic),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock))
|
|
end,
|
|
TopicIdInuse =
|
|
fun(Sock) ->
|
|
Setup(Sock),
|
|
%% topic-id is in use
|
|
ok = send_subscribe_frame(Sock, UsedSubId, <<"/queue/bar">>),
|
|
|
|
{ok, ErrorFrame} = recv_a_frame(Sock),
|
|
?assertMatch(#stomp_frame{command = <<"ERROR">>}, ErrorFrame),
|
|
?assertEqual(<<"Subscription id 0 is in used">>, ErrorFrame#stomp_frame.body),
|
|
?assertMatch({error, closed}, gen_tcp:recv(Sock, 0))
|
|
end,
|
|
|
|
SubscriptionInuse =
|
|
fun(Sock) ->
|
|
Setup(Sock),
|
|
%% topic is in use
|
|
ok = send_subscribe_frame(Sock, 1, UsedTopic),
|
|
|
|
{ok, ErrorFrame} = recv_a_frame(Sock),
|
|
?assertMatch(#stomp_frame{command = <<"ERROR">>}, ErrorFrame),
|
|
?assertEqual(<<"Topic /queue/foo already in subscribed">>, ErrorFrame#stomp_frame.body),
|
|
?assertMatch({error, closed}, gen_tcp:recv(Sock, 0))
|
|
end,
|
|
|
|
TopicIdInuseViaHttp =
|
|
fun(Sock) ->
|
|
Setup(Sock),
|
|
%% assert subscription stats
|
|
[#{clientid := ClientId}] = clients(),
|
|
{error, ErrMsg} = create_subscription(ClientId, <<"/queue/bar">>, UsedSubId),
|
|
?assertEqual(<<"Subscription id 0 is in used">>, ErrMsg),
|
|
|
|
ok = send_disconnect_frame(Sock)
|
|
end,
|
|
|
|
SubscriptionInuseViaHttp =
|
|
fun(Sock) ->
|
|
Setup(Sock),
|
|
%% assert subscription stats
|
|
[#{clientid := ClientId}] = clients(),
|
|
{error, ErrMsg} = create_subscription(ClientId, UsedTopic, <<"1">>),
|
|
?assertEqual(<<"Topic /queue/foo already in subscribed">>, ErrMsg),
|
|
|
|
ok = send_disconnect_frame(Sock)
|
|
end,
|
|
|
|
with_connection(TopicIdInuse),
|
|
with_connection(SubscriptionInuse),
|
|
with_connection(TopicIdInuseViaHttp),
|
|
with_connection(SubscriptionInuseViaHttp).
|
|
|
|
t_transaction(_) ->
|
|
with_connection(fun(Sock) ->
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"CONNECT">>,
|
|
[
|
|
{<<"accept-version">>, ?STOMP_VER},
|
|
{<<"host">>, <<"127.0.0.1:61613">>},
|
|
{<<"login">>, <<"guest">>},
|
|
{<<"passcode">>, <<"guest">>},
|
|
{<<"heart-beat">>, <<"0,0">>}
|
|
]
|
|
)
|
|
),
|
|
{ok, Data} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"CONNECTED">>,
|
|
headers = _,
|
|
body = _
|
|
},
|
|
_, _} = parse(Data),
|
|
|
|
%% Subscribe
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"SUBSCRIBE">>,
|
|
[
|
|
{<<"id">>, 0},
|
|
{<<"destination">>, <<"/queue/foo">>},
|
|
{<<"ack">>, <<"auto">>}
|
|
]
|
|
)
|
|
),
|
|
|
|
%% Transaction: tx1
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"BEGIN">>,
|
|
[{<<"transaction">>, <<"tx1">>}]
|
|
)
|
|
),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"SEND">>,
|
|
[
|
|
{<<"destination">>, <<"/queue/foo">>},
|
|
{<<"transaction">>, <<"tx1">>}
|
|
],
|
|
<<"hello">>
|
|
)
|
|
),
|
|
|
|
%% You will not receive any messages
|
|
{error, timeout} = gen_tcp:recv(Sock, 0, 1000),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"SEND">>,
|
|
[
|
|
{<<"destination">>, <<"/queue/foo">>},
|
|
{<<"transaction">>, <<"tx1">>}
|
|
],
|
|
<<"hello again">>
|
|
)
|
|
),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"COMMIT">>,
|
|
[{<<"transaction">>, <<"tx1">>}]
|
|
)
|
|
),
|
|
|
|
ct:sleep(1000),
|
|
{ok, Data1} = gen_tcp:recv(Sock, 0, 500),
|
|
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"MESSAGE">>,
|
|
headers = _,
|
|
body = <<"hello">>
|
|
},
|
|
Rest1, _} = parse(Data1),
|
|
|
|
%{ok, Data2} = gen_tcp:recv(Sock, 0, 500),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"MESSAGE">>,
|
|
headers = _,
|
|
body = <<"hello again">>
|
|
},
|
|
_Rest2, _} = parse(Rest1),
|
|
|
|
%% Transaction: tx2
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"BEGIN">>,
|
|
[{<<"transaction">>, <<"tx2">>}]
|
|
)
|
|
),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"SEND">>,
|
|
[
|
|
{<<"destination">>, <<"/queue/foo">>},
|
|
{<<"transaction">>, <<"tx2">>}
|
|
],
|
|
<<"hello">>
|
|
)
|
|
),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"ABORT">>,
|
|
[{<<"transaction">>, <<"tx2">>}]
|
|
)
|
|
),
|
|
|
|
%% You will not receive any messages
|
|
{error, timeout} = gen_tcp:recv(Sock, 0, 1000),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"DISCONNECT">>,
|
|
[{<<"receipt">>, <<"12345">>}]
|
|
)
|
|
),
|
|
|
|
{ok, Data3} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"RECEIPT">>,
|
|
headers = [{<<"receipt-id">>, <<"12345">>}],
|
|
body = _
|
|
},
|
|
_, _} = parse(Data3)
|
|
end).
|
|
|
|
t_receipt_in_error(_) ->
|
|
with_connection(fun(Sock) ->
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"CONNECT">>,
|
|
[
|
|
{<<"accept-version">>, ?STOMP_VER},
|
|
{<<"host">>, <<"127.0.0.1:61613">>},
|
|
{<<"login">>, <<"guest">>},
|
|
{<<"passcode">>, <<"guest">>},
|
|
{<<"heart-beat">>, <<"0,0">>}
|
|
]
|
|
)
|
|
),
|
|
{ok, Data} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"CONNECTED">>,
|
|
headers = _,
|
|
body = _
|
|
},
|
|
_, _} = parse(Data),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"ABORT">>,
|
|
[
|
|
{<<"transaction">>, <<"tx1">>},
|
|
{<<"receipt">>, <<"12345">>}
|
|
]
|
|
)
|
|
),
|
|
|
|
{ok, Data1} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
Frame = #stomp_frame{
|
|
command = <<"ERROR">>,
|
|
headers = _,
|
|
body = <<"Transaction tx1 not found">>
|
|
},
|
|
_, _} = parse(Data1),
|
|
|
|
<<"12345">> = proplists:get_value(<<"receipt-id">>, Frame#stomp_frame.headers)
|
|
end).
|
|
|
|
t_ack(_) ->
|
|
with_connection(fun(Sock) ->
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"CONNECT">>,
|
|
[
|
|
{<<"accept-version">>, ?STOMP_VER},
|
|
{<<"host">>, <<"127.0.0.1:61613">>},
|
|
{<<"login">>, <<"guest">>},
|
|
{<<"passcode">>, <<"guest">>},
|
|
{<<"heart-beat">>, <<"0,0">>}
|
|
]
|
|
)
|
|
),
|
|
{ok, Data} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"CONNECTED">>,
|
|
headers = _,
|
|
body = _
|
|
},
|
|
_, _} = parse(Data),
|
|
|
|
%% Subscribe
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"SUBSCRIBE">>,
|
|
[
|
|
{<<"id">>, 0},
|
|
{<<"destination">>, <<"/queue/foo">>},
|
|
{<<"ack">>, <<"client">>}
|
|
]
|
|
)
|
|
),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"SEND">>,
|
|
[{<<"destination">>, <<"/queue/foo">>}],
|
|
<<"ack test">>
|
|
)
|
|
),
|
|
|
|
{ok, Data1} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
Frame = #stomp_frame{
|
|
command = <<"MESSAGE">>,
|
|
headers = _,
|
|
body = <<"ack test">>
|
|
},
|
|
_, _} = parse(Data1),
|
|
|
|
AckId = proplists:get_value(<<"ack">>, Frame#stomp_frame.headers),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"ACK">>,
|
|
[
|
|
{<<"id">>, AckId},
|
|
{<<"receipt">>, <<"12345">>}
|
|
]
|
|
)
|
|
),
|
|
|
|
{ok, Data2} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"RECEIPT">>,
|
|
headers = [{<<"receipt-id">>, <<"12345">>}],
|
|
body = _
|
|
},
|
|
_, _} = parse(Data2),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"SEND">>,
|
|
[{<<"destination">>, <<"/queue/foo">>}],
|
|
<<"nack test">>
|
|
)
|
|
),
|
|
|
|
{ok, Data3} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
Frame1 = #stomp_frame{
|
|
command = <<"MESSAGE">>,
|
|
headers = _,
|
|
body = <<"nack test">>
|
|
},
|
|
_, _} = parse(Data3),
|
|
|
|
AckId1 = proplists:get_value(<<"ack">>, Frame1#stomp_frame.headers),
|
|
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"NACK">>,
|
|
[
|
|
{<<"id">>, AckId1},
|
|
{<<"receipt">>, <<"12345">>}
|
|
]
|
|
)
|
|
),
|
|
|
|
{ok, Data4} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"RECEIPT">>,
|
|
headers = [{<<"receipt-id">>, <<"12345">>}],
|
|
body = _
|
|
},
|
|
_, _} = parse(Data4)
|
|
end).
|
|
|
|
t_1000_msg_send(_) ->
|
|
with_connection(fun(Sock) ->
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"CONNECT">>,
|
|
[
|
|
{<<"accept-version">>, ?STOMP_VER},
|
|
{<<"host">>, <<"127.0.0.1:61613">>},
|
|
{<<"login">>, <<"guest">>},
|
|
{<<"passcode">>, <<"guest">>},
|
|
{<<"heart-beat">>, <<"0,0">>}
|
|
]
|
|
)
|
|
),
|
|
{ok, Data} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"CONNECTED">>,
|
|
headers = _,
|
|
body = _
|
|
},
|
|
_, _} = parse(Data),
|
|
|
|
Topic = <<"/queue/foo">>,
|
|
SendFun = fun() ->
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"SEND">>,
|
|
[{<<"destination">>, Topic}],
|
|
<<"msgtest">>
|
|
)
|
|
)
|
|
end,
|
|
|
|
RecvFun = fun() ->
|
|
receive
|
|
{deliver, Topic, _Msg} ->
|
|
ok
|
|
after 100 ->
|
|
?assert(false, "waiting message timeout")
|
|
end
|
|
end,
|
|
|
|
emqx:subscribe(Topic),
|
|
lists:foreach(fun(_) -> SendFun() end, lists:seq(1, 1000)),
|
|
lists:foreach(fun(_) -> RecvFun() end, lists:seq(1, 1000))
|
|
end).
|
|
|
|
t_sticky_packets_truncate_after_headers(_) ->
|
|
with_connection(fun(Sock) ->
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"CONNECT">>,
|
|
[
|
|
{<<"accept-version">>, ?STOMP_VER},
|
|
{<<"host">>, <<"127.0.0.1:61613">>},
|
|
{<<"login">>, <<"guest">>},
|
|
{<<"passcode">>, <<"guest">>},
|
|
{<<"heart-beat">>, <<"0,0">>}
|
|
]
|
|
)
|
|
),
|
|
{ok, Data} = gen_tcp:recv(Sock, 0),
|
|
{ok,
|
|
#stomp_frame{
|
|
command = <<"CONNECTED">>,
|
|
headers = _,
|
|
body = _
|
|
},
|
|
_, _} = parse(Data),
|
|
|
|
Topic = <<"/queue/foo">>,
|
|
|
|
emqx:subscribe(Topic),
|
|
gen_tcp:send(Sock, [
|
|
"SEND\n",
|
|
"content-length:3\n",
|
|
"destination:/queue/foo\n"
|
|
]),
|
|
timer:sleep(300),
|
|
gen_tcp:send(Sock, ["\nfoo", 0]),
|
|
receive
|
|
{deliver, Topic, _Msg} ->
|
|
ok
|
|
after 100 ->
|
|
?assert(false, "waiting message timeout")
|
|
end
|
|
end).
|
|
|
|
t_frame_error_in_connect(_) ->
|
|
with_connection(fun(Sock) ->
|
|
gen_tcp:send(
|
|
Sock,
|
|
serialize(
|
|
<<"CONNECT">>,
|
|
[
|
|
{<<"accept-version">>, ?STOMP_VER},
|
|
{<<"host">>, <<"127.0.0.1:61613">>},
|
|
{<<"login">>, <<"guest">>},
|
|
{<<"passcode">>, <<"guest">>},
|
|
{<<"heart-beat">>, <<"0,0">>},
|
|
{<<"custome_header1">>, <<"val">>},
|
|
{<<"custome_header2">>, <<"val">>},
|
|
{<<"custome_header3">>, <<"val">>},
|
|
{<<"custome_header4">>, <<"val">>},
|
|
{<<"custome_header5">>, <<"val">>},
|
|
{<<"custome_header6">>, <<"val">>}
|
|
]
|
|
)
|
|
),
|
|
?assertMatch({error, closed}, gen_tcp:recv(Sock, 0))
|
|
end).
|
|
|
|
t_frame_error_too_many_headers(_) ->
|
|
Frame = serialize(
|
|
<<"SEND">>,
|
|
[
|
|
{<<"destination">>, <<"/queue/foo">>},
|
|
{<<"custome_header1">>, <<"val">>},
|
|
{<<"custome_header2">>, <<"val">>},
|
|
{<<"custome_header3">>, <<"val">>},
|
|
{<<"custome_header4">>, <<"val">>},
|
|
{<<"custome_header5">>, <<"val">>},
|
|
{<<"custome_header6">>, <<"val">>},
|
|
{<<"custome_header7">>, <<"val">>},
|
|
{<<"custome_header8">>, <<"val">>},
|
|
{<<"custome_header9">>, <<"val">>},
|
|
{<<"custome_header10">>, <<"val">>}
|
|
],
|
|
<<"test">>
|
|
),
|
|
Assert =
|
|
fun(Sock) ->
|
|
{ok, ErrorFrame} = recv_a_frame(Sock),
|
|
?assertMatch(#stomp_frame{command = <<"ERROR">>}, ErrorFrame),
|
|
?assertMatch(
|
|
match, re:run(ErrorFrame#stomp_frame.body, "too_many_headers", [{capture, none}])
|
|
),
|
|
?assertMatch({error, closed}, gen_tcp:recv(Sock, 0))
|
|
end,
|
|
test_frame_error(Frame, Assert).
|
|
|
|
t_frame_error_too_long_header(_) ->
|
|
LongHeaderVal = emqx_utils:bin_to_hexstr(crypto:strong_rand_bytes(50), upper),
|
|
Frame = serialize(
|
|
<<"SEND">>,
|
|
[
|
|
{<<"destination">>, <<"/queue/foo">>},
|
|
{<<"custome_header10">>, LongHeaderVal}
|
|
],
|
|
<<"test">>
|
|
),
|
|
Assert =
|
|
fun(Sock) ->
|
|
{ok, ErrorFrame} = recv_a_frame(Sock),
|
|
?assertMatch(#stomp_frame{command = <<"ERROR">>}, ErrorFrame),
|
|
?assertMatch(
|
|
match, re:run(ErrorFrame#stomp_frame.body, "too_long_header", [{capture, none}])
|
|
),
|
|
?assertMatch({error, closed}, gen_tcp:recv(Sock, 0))
|
|
end,
|
|
test_frame_error(Frame, Assert).
|
|
|
|
t_frame_error_too_long_body(_) ->
|
|
LongBody = emqx_utils:bin_to_hexstr(crypto:strong_rand_bytes(513), upper),
|
|
Frame = serialize(
|
|
<<"SEND">>,
|
|
[{<<"destination">>, <<"/queue/foo">>}],
|
|
LongBody
|
|
),
|
|
Assert =
|
|
fun(Sock) ->
|
|
{ok, ErrorFrame} = recv_a_frame(Sock),
|
|
?assertMatch(#stomp_frame{command = <<"ERROR">>}, ErrorFrame),
|
|
?assertMatch(
|
|
match, re:run(ErrorFrame#stomp_frame.body, "too_long_body", [{capture, none}])
|
|
),
|
|
?assertMatch({error, closed}, gen_tcp:recv(Sock, 0))
|
|
end,
|
|
test_frame_error(Frame, Assert).
|
|
|
|
test_frame_error(Frame, AssertFun) ->
|
|
with_connection(fun(Sock) ->
|
|
send_connection_frame(Sock, <<"guest">>, <<"guest">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
gen_tcp:send(Sock, Frame),
|
|
AssertFun(Sock)
|
|
end).
|
|
|
|
t_rest_clientid_info(_) ->
|
|
with_connection(fun(Sock) ->
|
|
send_connection_frame(Sock, <<"guest">>, <<"guest">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
|
|
%% client lists
|
|
{200, Clients} = request(get, "/gateways/stomp/clients"),
|
|
?assertEqual(1, length(maps:get(data, Clients))),
|
|
StompClient = lists:nth(1, maps:get(data, Clients)),
|
|
ClientId = maps:get(clientid, StompClient),
|
|
ClientPath =
|
|
"/gateways/stomp/clients/" ++
|
|
binary_to_list(ClientId),
|
|
{200, StompClient1} = request(get, ClientPath),
|
|
?assertEqual(StompClient, StompClient1),
|
|
assert_fields_exist(
|
|
[
|
|
proto_name,
|
|
awaiting_rel_max,
|
|
inflight_cnt,
|
|
disconnected_at,
|
|
send_msg,
|
|
heap_size,
|
|
connected,
|
|
recv_cnt,
|
|
send_pkt,
|
|
mailbox_len,
|
|
username,
|
|
recv_pkt,
|
|
expiry_interval,
|
|
clientid,
|
|
mqueue_max,
|
|
send_oct,
|
|
ip_address,
|
|
is_bridge,
|
|
awaiting_rel_cnt,
|
|
mqueue_dropped,
|
|
mqueue_len,
|
|
node,
|
|
inflight_max,
|
|
reductions,
|
|
subscriptions_max,
|
|
connected_at,
|
|
keepalive,
|
|
created_at,
|
|
clean_start,
|
|
subscriptions_cnt,
|
|
recv_msg,
|
|
send_cnt,
|
|
proto_ver,
|
|
recv_oct
|
|
],
|
|
StompClient
|
|
),
|
|
|
|
%% sub & unsub
|
|
{200, []} = request(get, ClientPath ++ "/subscriptions"),
|
|
ok = send_subscribe_frame(Sock, 0, <<"/queue/foo">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
|
|
{200, Subs} = request(get, ClientPath ++ "/subscriptions"),
|
|
?assertEqual(1, length(Subs)),
|
|
assert_fields_exist([topic, qos], lists:nth(1, Subs)),
|
|
|
|
{201, _} = request(
|
|
post,
|
|
ClientPath ++ "/subscriptions",
|
|
#{
|
|
topic => <<"t/a">>,
|
|
qos => 1,
|
|
sub_props => #{subid => <<"1001">>}
|
|
}
|
|
),
|
|
|
|
{200, Subs1} = request(get, ClientPath ++ "/subscriptions"),
|
|
?assertEqual(2, length(Subs1)),
|
|
{200, StompClient2} = request(get, ClientPath),
|
|
?assertMatch(#{subscriptions_cnt := 2}, StompClient2),
|
|
|
|
{204, _} = request(delete, ClientPath ++ "/subscriptions/t%2Fa"),
|
|
{200, Subs2} = request(get, ClientPath ++ "/subscriptions"),
|
|
?assertEqual(1, length(Subs2)),
|
|
{200, StompClient3} = request(get, ClientPath),
|
|
?assertMatch(#{subscriptions_cnt := 1}, StompClient3),
|
|
|
|
%% kickout
|
|
{204, _} = request(delete, ClientPath),
|
|
% sync
|
|
ignored = gen_server:call(emqx_cm, ignore, infinity),
|
|
ok = emqx_pool:flush_async_tasks(),
|
|
{200, Clients2} = request(get, "/gateways/stomp/clients"),
|
|
?assertEqual(0, length(maps:get(data, Clients2)))
|
|
end).
|
|
|
|
t_authn_superuser(_) ->
|
|
%% mock authn
|
|
meck:new(emqx_access_control, [passthrough]),
|
|
meck:expect(
|
|
emqx_access_control,
|
|
authenticate,
|
|
fun
|
|
(#{username := <<"admin">>}) ->
|
|
{ok, #{is_superuser => true}};
|
|
(#{username := <<"bad_user">>}) ->
|
|
{error, not_authorized};
|
|
(_) ->
|
|
{ok, #{is_superuser => false}}
|
|
end
|
|
),
|
|
%% mock authz
|
|
meck:expect(
|
|
emqx_access_control,
|
|
authorize,
|
|
fun
|
|
(_ClientInfo = #{is_superuser := true}, _PubSub, _Topic) ->
|
|
allow;
|
|
(_ClientInfo, _PubSub, _Topic) ->
|
|
deny
|
|
end
|
|
),
|
|
|
|
LoginFailure = fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"bad_user">>, <<"public">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"ERROR">>}}, recv_a_frame(Sock)),
|
|
?assertMatch({error, closed}, recv_a_frame(Sock))
|
|
end,
|
|
|
|
PublishFailure = fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"user1">>, <<"public">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
ok = send_message_frame(Sock, <<"t/a">>, <<"hello">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"ERROR">>}}, recv_a_frame(Sock)),
|
|
?assertMatch({error, closed}, recv_a_frame(Sock))
|
|
end,
|
|
|
|
SubscribeFailed = fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"user1">>, <<"public">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
ok = send_subscribe_frame(Sock, 0, <<"t/a">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"ERROR">>}}, recv_a_frame(Sock)),
|
|
?assertMatch({error, closed}, recv_a_frame(Sock))
|
|
end,
|
|
|
|
LoginAsSuperUser = fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"admin">>, <<"public">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
ok = send_subscribe_frame(Sock, 0, <<"t/a">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
ok = send_message_frame(Sock, <<"t/a">>, <<"hello">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
?assertMatch(
|
|
{ok, #stomp_frame{
|
|
command = <<"MESSAGE">>,
|
|
body = <<"hello">>
|
|
}},
|
|
recv_a_frame(Sock)
|
|
),
|
|
ok = send_disconnect_frame(Sock)
|
|
end,
|
|
|
|
with_connection(LoginFailure),
|
|
with_connection(PublishFailure),
|
|
with_connection(SubscribeFailed),
|
|
with_connection(LoginAsSuperUser),
|
|
meck:unload(emqx_access_control).
|
|
|
|
t_mountpoint(_) ->
|
|
restart_stomp_with_mountpoint(<<"stomp/">>),
|
|
|
|
PubSub = fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"user1">>, <<"public">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
ok = send_subscribe_frame(Sock, 0, <<"t/a">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
ok = send_message_frame(Sock, <<"t/a">>, <<"hello">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
|
|
{ok, #stomp_frame{
|
|
command = <<"MESSAGE">>,
|
|
headers = Headers,
|
|
body = <<"hello">>
|
|
}} = recv_a_frame(Sock),
|
|
?assertEqual(<<"t/a">>, proplists:get_value(<<"destination">>, Headers)),
|
|
|
|
?assertEqual(
|
|
<<"text/plain">>, proplists:get_value(<<"content-type">>, Headers)
|
|
),
|
|
|
|
ok = send_disconnect_frame(Sock)
|
|
end,
|
|
|
|
PubToMqtt = fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"user1">>, <<"public">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
|
|
ok = emqx:subscribe(<<"stomp/t/a">>),
|
|
ok = send_message_frame(Sock, <<"t/a">>, <<"hello">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
|
|
receive
|
|
{deliver, Topic, Msg} ->
|
|
?assertEqual(<<"stomp/t/a">>, Topic),
|
|
?assertEqual(<<"hello">>, emqx_message:payload(Msg))
|
|
after 100 ->
|
|
?assert(false, "waiting message timeout")
|
|
end,
|
|
ok = send_disconnect_frame(Sock)
|
|
end,
|
|
|
|
ReceiveMsgFromMqtt = fun(Sock) ->
|
|
ok = send_connection_frame(Sock, <<"user1">>, <<"public">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"CONNECTED">>}}, recv_a_frame(Sock)),
|
|
ok = send_subscribe_frame(Sock, 0, <<"t/a">>),
|
|
?assertMatch({ok, #stomp_frame{command = <<"RECEIPT">>}}, recv_a_frame(Sock)),
|
|
|
|
Msg = emqx_message:make(<<"stomp/t/a">>, <<"hello">>),
|
|
emqx:publish(Msg),
|
|
|
|
{ok, #stomp_frame{
|
|
command = <<"MESSAGE">>,
|
|
headers = Headers,
|
|
body = <<"hello">>
|
|
}} = recv_a_frame(Sock),
|
|
?assertEqual(<<"t/a">>, proplists:get_value(<<"destination">>, Headers)),
|
|
|
|
ok = send_disconnect_frame(Sock)
|
|
end,
|
|
|
|
with_connection(PubSub),
|
|
with_connection(PubToMqtt),
|
|
with_connection(ReceiveMsgFromMqtt),
|
|
restart_stomp_with_mountpoint(<<>>).
|
|
|
|
%% TODO: Mountpoint, AuthChain, Authorization + Mountpoint, ClientInfoOverride,
|
|
%% Listeners, Metrics, Stats, ClientInfo
|
|
%%
|
|
%% TODO: Start/Stop, List Instance
|
|
%%
|
|
%% TODO: RateLimit, OOM,
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% helpers
|
|
|
|
with_connection(DoFun) ->
|
|
{ok, Sock} = gen_tcp:connect(
|
|
{127, 0, 0, 1},
|
|
61613,
|
|
[binary, {packet, raw}, {active, false}],
|
|
3000
|
|
),
|
|
try
|
|
DoFun(Sock)
|
|
after
|
|
erase(parser),
|
|
erase(rest),
|
|
gen_tcp:close(Sock)
|
|
end.
|
|
|
|
serialize(Command, Headers) ->
|
|
emqx_stomp_frame:serialize_pkt(emqx_stomp_frame:make(Command, Headers), #{}).
|
|
|
|
serialize(Command, Headers, Body) ->
|
|
emqx_stomp_frame:serialize_pkt(emqx_stomp_frame:make(Command, Headers, Body), #{}).
|
|
|
|
recv_a_frame(Sock) ->
|
|
Parser =
|
|
case get(parser) of
|
|
undefined ->
|
|
ProtoEnv = #{
|
|
max_headers => 1024,
|
|
max_header_length => 10240,
|
|
max_body_length => 81920
|
|
},
|
|
emqx_stomp_frame:initial_parse_state(ProtoEnv);
|
|
P ->
|
|
P
|
|
end,
|
|
LastRest =
|
|
case get(rest) of
|
|
undefined -> <<>>;
|
|
R -> R
|
|
end,
|
|
case emqx_stomp_frame:parse(LastRest, Parser) of
|
|
{more, NParser} ->
|
|
case gen_tcp:recv(Sock, 0, 5000) of
|
|
{ok, Data} ->
|
|
put(parser, NParser),
|
|
put(rest, <<LastRest/binary, Data/binary>>),
|
|
recv_a_frame(Sock);
|
|
{error, _} = Err1 ->
|
|
erase(parser),
|
|
erase(rest),
|
|
Err1
|
|
end;
|
|
{ok, Frame, Rest, NParser} ->
|
|
put(parser, NParser),
|
|
put(rest, Rest),
|
|
ct:pal("recv_a_frame: ~p~n", [Frame]),
|
|
{ok, Frame};
|
|
{error, _} = Err ->
|
|
erase(parser),
|
|
erase(rest),
|
|
Err
|
|
end.
|
|
|
|
parse(Data) ->
|
|
ProtoEnv = #{
|
|
max_headers => 1024,
|
|
max_header_length => 10240,
|
|
max_body_length => 81920
|
|
},
|
|
Parser = emqx_stomp_frame:initial_parse_state(ProtoEnv),
|
|
emqx_stomp_frame:parse(Data, Parser).
|
|
|
|
get_field(command, #stomp_frame{command = Command}) ->
|
|
Command;
|
|
get_field(body, #stomp_frame{body = Body}) ->
|
|
Body.
|
|
|
|
send_connection_frame(Sock, Username, Password) ->
|
|
send_connection_frame(Sock, Username, Password, <<"0,0">>).
|
|
|
|
send_connection_frame(Sock, Username, Password, Heartbeat) ->
|
|
Headers =
|
|
case Username == undefined of
|
|
true -> [];
|
|
false -> [{<<"login">>, Username}]
|
|
end ++
|
|
case Password == undefined of
|
|
true -> [];
|
|
false -> [{<<"passcode">>, Password}]
|
|
end,
|
|
Headers1 = [
|
|
{<<"accept-version">>, ?STOMP_VER},
|
|
{<<"host">>, <<"127.0.0.1:61613">>},
|
|
{<<"heart-beat">>, Heartbeat}
|
|
| Headers
|
|
],
|
|
ok = gen_tcp:send(Sock, serialize(<<"CONNECT">>, Headers1)).
|
|
|
|
send_subscribe_frame(Sock, Id, Topic) ->
|
|
Headers =
|
|
[
|
|
{<<"id">>, Id},
|
|
{<<"receipt">>, Id},
|
|
{<<"destination">>, Topic},
|
|
{<<"ack">>, <<"auto">>}
|
|
],
|
|
ok = gen_tcp:send(Sock, serialize(<<"SUBSCRIBE">>, Headers)).
|
|
|
|
send_unsubscribe_frame(Sock, Id) when is_integer(Id) ->
|
|
Headers =
|
|
[
|
|
{<<"id">>, Id},
|
|
{<<"receipt">>, <<"rp-", (integer_to_binary(Id))/binary>>}
|
|
],
|
|
gen_tcp:send(Sock, serialize(<<"UNSUBSCRIBE">>, Headers)).
|
|
|
|
send_message_frame(Sock, Topic, Payload) ->
|
|
send_message_frame(Sock, Topic, Payload, []).
|
|
|
|
send_message_frame(Sock, Topic, Payload, Headers0) ->
|
|
Headers =
|
|
[
|
|
{<<"destination">>, Topic},
|
|
{<<"receipt">>, <<"rp-", Topic/binary>>}
|
|
| Headers0
|
|
],
|
|
ok = gen_tcp:send(Sock, serialize(<<"SEND">>, Headers, Payload)).
|
|
|
|
send_disconnect_frame(Sock) ->
|
|
ok = gen_tcp:send(Sock, serialize(<<"DISCONNECT">>, [])).
|
|
|
|
send_disconnect_frame(Sock, ReceiptId) ->
|
|
Headers = [{<<"receipt">>, ReceiptId}],
|
|
ok = gen_tcp:send(Sock, serialize(<<"DISCONNECT">>, Headers)).
|
|
|
|
clients() ->
|
|
{200, Clients} = request(get, "/gateways/stomp/clients"),
|
|
maps:get(data, Clients).
|
|
|
|
create_subscription(ClientId, Topic, SubId) ->
|
|
Path = io_lib:format("/gateways/stomp/clients/~s/subscriptions", [ClientId]),
|
|
Body = #{
|
|
topic => Topic,
|
|
qos => 1,
|
|
sub_props => #{subid => SubId}
|
|
},
|
|
case request(post, Path, Body) of
|
|
{201, _} ->
|
|
ok;
|
|
{400, #{message := Message}} ->
|
|
{error, Message}
|
|
end.
|