emqx_bridge_sqlserver_SUITE.erl 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688
  1. %%--------------------------------------------------------------------
  2. % Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
  3. %%--------------------------------------------------------------------
  4. -module(emqx_bridge_sqlserver_SUITE).
  5. -compile(nowarn_export_all).
  6. -compile(export_all).
  7. -include("emqx_bridge_sqlserver/include/emqx_bridge_sqlserver.hrl").
  8. -include_lib("eunit/include/eunit.hrl").
  9. -include_lib("common_test/include/ct.hrl").
  10. -include_lib("snabbkaffe/include/snabbkaffe.hrl").
  11. % SQL definitions
  12. -define(SQL_BRIDGE,
  13. "insert into t_mqtt_msg(msgid, topic, qos, payload) values ( ${id}, ${topic}, ${qos}, ${payload})"
  14. ).
  15. -define(SQL_SERVER_DRIVER, "ms-sql").
  16. -define(SQL_CREATE_DATABASE_IF_NOT_EXISTS,
  17. " IF NOT EXISTS(SELECT name FROM sys.databases WHERE name = 'mqtt')"
  18. " BEGIN"
  19. " CREATE DATABASE mqtt;"
  20. " END"
  21. ).
  22. -define(SQL_CREATE_TABLE_IN_DB_MQTT,
  23. " CREATE TABLE mqtt.dbo.t_mqtt_msg"
  24. " (id int PRIMARY KEY IDENTITY(1000000001,1) NOT NULL,"
  25. " msgid VARCHAR(64) NULL,"
  26. " topic VARCHAR(100) NULL,"
  27. " qos tinyint NOT NULL DEFAULT 0,"
  28. %% use VARCHAR to use utf8 encoding
  29. %% for default, sqlserver use utf16 encoding NVARCHAR()
  30. " payload VARCHAR(100) NULL,"
  31. " arrived DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP)"
  32. ).
  33. -define(SQL_DROP_DB_MQTT, "DROP DATABASE mqtt").
  34. -define(SQL_DROP_TABLE, "DROP TABLE mqtt.dbo.t_mqtt_msg").
  35. -define(SQL_DELETE, "DELETE from mqtt.dbo.t_mqtt_msg").
  36. -define(SQL_SELECT, "SELECT payload FROM mqtt.dbo.t_mqtt_msg").
  37. -define(SQL_SELECT_COUNT, "SELECT COUNT(*) FROM mqtt.dbo.t_mqtt_msg").
  38. % DB defaults
  39. -define(SQL_SERVER_DATABASE, "mqtt").
  40. -define(SQL_SERVER_USERNAME, "sa").
  41. -define(SQL_SERVER_PASSWORD, "mqtt_public1").
  42. -define(BATCH_SIZE, 10).
  43. -define(REQUEST_TIMEOUT_MS, 500).
  44. -define(WORKER_POOL_SIZE, 4).
  45. -define(WITH_CON(Process),
  46. Con = connect_direct_sqlserver(Config),
  47. Process,
  48. ok = disconnect(Con)
  49. ).
  50. %% How to run it locally (all commands are run in $PROJ_ROOT dir):
  51. %% A: run ct on host
  52. %% 1. Start all deps services
  53. %% ```bash
  54. %% sudo docker compose -f .ci/docker-compose-file/docker-compose.yaml \
  55. %% -f .ci/docker-compose-file/docker-compose-sqlserver.yaml \
  56. %% -f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \
  57. %% up --build
  58. %% ```
  59. %%
  60. %% 2. Run use cases with special environment variables
  61. %% 11433 is toxiproxy exported port.
  62. %% Local:
  63. %% ```bash
  64. %% SQLSERVER_HOST=toxiproxy SQLSERVER_PORT=11433 \
  65. %% PROXY_HOST=toxiproxy PROXY_PORT=1433 \
  66. %% ./rebar3 as test ct -c -v --readable true --name ct@127.0.0.1 \
  67. %% --suite apps/emqx_bridge_sqlserver/test/emqx_bridge_sqlserver_SUITE.erl
  68. %% ```
  69. %%
  70. %% B: run ct in docker container
  71. %% run script:
  72. %% ```bash
  73. %% ./scripts/ct/run.sh --ci --app apps/emqx_bridge_sqlserver/ -- \
  74. %% --name 'test@127.0.0.1' -c -v --readable true \
  75. %% --suite apps/emqx_bridge_sqlserver/test/emqx_bridge_sqlserver_SUITE.erl
  76. %% ````
  77. %%------------------------------------------------------------------------------
  78. %% CT boilerplate
  79. %%------------------------------------------------------------------------------
  80. all() ->
  81. [
  82. {group, async},
  83. {group, sync}
  84. ].
  85. groups() ->
  86. TCs = emqx_common_test_helpers:all(?MODULE),
  87. NonBatchCases = [t_write_timeout],
  88. BatchingGroups = [{group, with_batch}, {group, without_batch}],
  89. [
  90. {async, BatchingGroups},
  91. {sync, BatchingGroups},
  92. {with_batch, TCs -- NonBatchCases},
  93. {without_batch, TCs}
  94. ].
  95. init_per_group(async, Config) ->
  96. [{query_mode, async} | Config];
  97. init_per_group(sync, Config) ->
  98. [{query_mode, sync} | Config];
  99. init_per_group(with_batch, Config0) ->
  100. Config = [{enable_batch, true} | Config0],
  101. common_init(Config);
  102. init_per_group(without_batch, Config0) ->
  103. Config = [{enable_batch, false} | Config0],
  104. common_init(Config);
  105. init_per_group(_Group, Config) ->
  106. Config.
  107. end_per_group(Group, Config) when Group =:= with_batch; Group =:= without_batch ->
  108. connect_and_drop_table(Config),
  109. connect_and_drop_db(Config),
  110. ProxyHost = ?config(proxy_host, Config),
  111. ProxyPort = ?config(proxy_port, Config),
  112. emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
  113. ok;
  114. end_per_group(_Group, _Config) ->
  115. ok.
  116. init_per_suite(Config) ->
  117. Config.
  118. end_per_suite(_Config) ->
  119. emqx_mgmt_api_test_util:end_suite(),
  120. ok = emqx_common_test_helpers:stop_apps([emqx_bridge, emqx_conf]),
  121. ok.
  122. init_per_testcase(_Testcase, Config) ->
  123. %% drop database and table
  124. %% connect_and_clear_table(Config),
  125. %% create a new one
  126. %% TODO: create a new database for each test case
  127. delete_bridge(Config),
  128. snabbkaffe:start_trace(),
  129. Config.
  130. end_per_testcase(_Testcase, Config) ->
  131. ProxyHost = ?config(proxy_host, Config),
  132. ProxyPort = ?config(proxy_port, Config),
  133. emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
  134. connect_and_clear_table(Config),
  135. ok = snabbkaffe:stop(),
  136. delete_bridge(Config),
  137. ok.
  138. %%------------------------------------------------------------------------------
  139. %% Testcases
  140. %%------------------------------------------------------------------------------
  141. t_setup_via_config_and_publish(Config) ->
  142. ?assertMatch(
  143. {ok, _},
  144. create_bridge(Config)
  145. ),
  146. Val = str(erlang:unique_integer()),
  147. SentData = sent_data(Val),
  148. ?check_trace(
  149. begin
  150. ?wait_async_action(
  151. ?assertEqual(ok, send_message(Config, SentData)),
  152. #{?snk_kind := sqlserver_connector_query_return},
  153. 10_000
  154. ),
  155. ?assertMatch(
  156. [{Val}],
  157. connect_and_get_payload(Config)
  158. ),
  159. ok
  160. end,
  161. fun(Trace0) ->
  162. Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
  163. ?assertMatch([#{result := ok}], Trace),
  164. ok
  165. end
  166. ),
  167. ok.
  168. t_setup_via_http_api_and_publish(Config) ->
  169. BridgeType = ?config(sqlserver_bridge_type, Config),
  170. Name = ?config(sqlserver_name, Config),
  171. SQLServerConfig0 = ?config(sqlserver_config, Config),
  172. SQLServerConfig = SQLServerConfig0#{
  173. <<"name">> => Name,
  174. <<"type">> => BridgeType
  175. },
  176. ?assertMatch(
  177. {ok, _},
  178. create_bridge_http(SQLServerConfig)
  179. ),
  180. Val = str(erlang:unique_integer()),
  181. SentData = sent_data(Val),
  182. ?check_trace(
  183. begin
  184. ?wait_async_action(
  185. ?assertEqual(ok, send_message(Config, SentData)),
  186. #{?snk_kind := sqlserver_connector_query_return},
  187. 10_000
  188. ),
  189. ?assertMatch(
  190. [{Val}],
  191. connect_and_get_payload(Config)
  192. ),
  193. ok
  194. end,
  195. fun(Trace0) ->
  196. Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
  197. ?assertMatch([#{result := ok}], Trace),
  198. ok
  199. end
  200. ),
  201. ok.
  202. t_get_status(Config) ->
  203. ?assertMatch(
  204. {ok, _},
  205. create_bridge(Config)
  206. ),
  207. ProxyPort = ?config(proxy_port, Config),
  208. ProxyHost = ?config(proxy_host, Config),
  209. ProxyName = ?config(proxy_name, Config),
  210. health_check_resource_ok(Config),
  211. emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
  212. health_check_resource_down(Config)
  213. end),
  214. ok.
  215. t_create_disconnected(Config) ->
  216. ProxyPort = ?config(proxy_port, Config),
  217. ProxyHost = ?config(proxy_host, Config),
  218. ProxyName = ?config(proxy_name, Config),
  219. emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
  220. ?assertMatch({ok, _}, create_bridge(Config)),
  221. health_check_resource_down(Config)
  222. end),
  223. health_check_resource_ok(Config),
  224. ok.
  225. t_create_with_invalid_password(Config) ->
  226. BridgeType = ?config(sqlserver_bridge_type, Config),
  227. Name = ?config(sqlserver_name, Config),
  228. SQLServerConfig0 = ?config(sqlserver_config, Config),
  229. SQLServerConfig = SQLServerConfig0#{
  230. <<"name">> => Name,
  231. <<"type">> => BridgeType,
  232. <<"password">> => <<"wrong_password">>
  233. },
  234. ?check_trace(
  235. begin
  236. ?assertMatch(
  237. {ok, _},
  238. create_bridge_http(SQLServerConfig)
  239. )
  240. end,
  241. fun(Trace) ->
  242. ?assertMatch(
  243. [#{error := {start_pool_failed, _, _}}],
  244. ?of_kind(sqlserver_connector_start_failed, Trace)
  245. ),
  246. ok
  247. end
  248. ),
  249. ok.
  250. t_write_failure(Config) ->
  251. ProxyName = ?config(proxy_name, Config),
  252. ProxyPort = ?config(proxy_port, Config),
  253. ProxyHost = ?config(proxy_host, Config),
  254. QueryMode = ?config(query_mode, Config),
  255. Val = str(erlang:unique_integer()),
  256. SentData = sent_data(Val),
  257. {{ok, _}, {ok, _}} =
  258. ?wait_async_action(
  259. create_bridge(Config),
  260. #{?snk_kind := resource_connected_enter},
  261. 20_000
  262. ),
  263. emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
  264. case QueryMode of
  265. sync ->
  266. ?assertMatch(
  267. {error, {resource_error, #{reason := timeout}}},
  268. send_message(Config, SentData)
  269. );
  270. async ->
  271. ?assertMatch(
  272. ok, send_message(Config, SentData)
  273. )
  274. end
  275. end),
  276. ok.
  277. t_write_timeout(_Config) ->
  278. %% msodbc driver handled all connection exceptions
  279. %% the case is same as t_write_failure/1
  280. ok.
  281. t_simple_query(Config) ->
  282. BatchSize = batch_size(Config),
  283. ?assertMatch(
  284. {ok, _},
  285. create_bridge(Config)
  286. ),
  287. {Requests, Vals} = gen_batch_req(BatchSize),
  288. ?check_trace(
  289. begin
  290. ?wait_async_action(
  291. begin
  292. [?assertEqual(ok, query_resource(Config, Request)) || Request <- Requests]
  293. end,
  294. #{?snk_kind := sqlserver_connector_query_return},
  295. 10_000
  296. ),
  297. %% just assert the data count is correct
  298. ?assertMatch(
  299. BatchSize,
  300. connect_and_get_count(Config)
  301. ),
  302. %% assert the data order is correct
  303. ?assertMatch(
  304. Vals,
  305. connect_and_get_payload(Config)
  306. )
  307. end,
  308. fun(Trace0) ->
  309. Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
  310. case BatchSize of
  311. 1 ->
  312. ?assertMatch([#{result := ok}], Trace);
  313. _ ->
  314. [?assertMatch(#{result := ok}, Trace1) || Trace1 <- Trace]
  315. end,
  316. ok
  317. end
  318. ),
  319. ok.
  320. -define(MISSING_TINYINT_ERROR,
  321. "[Microsoft][ODBC Driver 17 for SQL Server][SQL Server]"
  322. "Conversion failed when converting the varchar value 'undefined' to data type tinyint. SQLSTATE IS: 22018"
  323. ).
  324. t_missing_data(Config) ->
  325. QueryMode = ?config(query_mode, Config),
  326. ?assertMatch(
  327. {ok, _},
  328. create_bridge(Config)
  329. ),
  330. Result = send_message(Config, #{}),
  331. case QueryMode of
  332. sync ->
  333. ?assertMatch(
  334. {error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}},
  335. Result
  336. );
  337. async ->
  338. ?assertMatch(
  339. ok, send_message(Config, #{})
  340. )
  341. end,
  342. ok.
  343. t_bad_parameter(Config) ->
  344. QueryMode = ?config(query_mode, Config),
  345. ?assertMatch(
  346. {ok, _},
  347. create_bridge(Config)
  348. ),
  349. Result = send_message(Config, #{}),
  350. case QueryMode of
  351. sync ->
  352. ?assertMatch(
  353. {error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}},
  354. Result
  355. );
  356. async ->
  357. ?assertMatch(
  358. ok, send_message(Config, #{})
  359. )
  360. end,
  361. ok.
  362. %%------------------------------------------------------------------------------
  363. %% Helper fns
  364. %%------------------------------------------------------------------------------
  365. common_init(ConfigT) ->
  366. Host = os:getenv("SQLSERVER_HOST", "toxiproxy"),
  367. Port = list_to_integer(os:getenv("SQLSERVER_PORT", str(?SQLSERVER_DEFAULT_PORT))),
  368. Config0 = [
  369. {sqlserver_host, Host},
  370. {sqlserver_port, Port},
  371. %% see also for `proxy_name` : $PROJ_ROOT/.ci/docker-compose-file/toxiproxy.json
  372. {proxy_name, "sqlserver"},
  373. {batch_size, batch_size(ConfigT)}
  374. | ConfigT
  375. ],
  376. BridgeType = proplists:get_value(bridge_type, Config0, <<"sqlserver">>),
  377. case emqx_common_test_helpers:is_tcp_server_available(Host, Port) of
  378. true ->
  379. % Setup toxiproxy
  380. ProxyHost = os:getenv("PROXY_HOST", "toxiproxy"),
  381. ProxyPort = list_to_integer(os:getenv("PROXY_PORT", "8474")),
  382. emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
  383. % Ensure enterprise bridge module is loaded
  384. ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_bridge, odbc]),
  385. _ = emqx_bridge_enterprise:module_info(),
  386. emqx_mgmt_api_test_util:init_suite(),
  387. % Connect to sqlserver directly
  388. % drop old db and table, and then create new ones
  389. connect_and_create_db_and_table(Config0),
  390. {Name, SQLServerConf} = sqlserver_config(BridgeType, Config0),
  391. Config =
  392. [
  393. {sqlserver_config, SQLServerConf},
  394. {sqlserver_bridge_type, BridgeType},
  395. {sqlserver_name, Name},
  396. {proxy_host, ProxyHost},
  397. {proxy_port, ProxyPort}
  398. | Config0
  399. ],
  400. Config;
  401. false ->
  402. case os:getenv("IS_CI") of
  403. "yes" ->
  404. throw(no_sqlserver);
  405. _ ->
  406. {skip, no_sqlserver}
  407. end
  408. end.
  409. sqlserver_config(BridgeType, Config) ->
  410. Port = integer_to_list(?config(sqlserver_port, Config)),
  411. Server = ?config(sqlserver_host, Config) ++ ":" ++ Port,
  412. Name = atom_to_binary(?MODULE),
  413. BatchSize = batch_size(Config),
  414. QueryMode = ?config(query_mode, Config),
  415. ConfigString =
  416. io_lib:format(
  417. "bridges.~s.~s {\n"
  418. " enable = true\n"
  419. " server = ~p\n"
  420. " database = ~p\n"
  421. " username = ~p\n"
  422. " password = ~p\n"
  423. " sql = ~p\n"
  424. " driver = ~p\n"
  425. " resource_opts = {\n"
  426. " request_ttl = 500ms\n"
  427. " batch_size = ~b\n"
  428. " query_mode = ~s\n"
  429. " worker_pool_size = ~b\n"
  430. " }\n"
  431. "}",
  432. [
  433. BridgeType,
  434. Name,
  435. Server,
  436. ?SQL_SERVER_DATABASE,
  437. ?SQL_SERVER_USERNAME,
  438. ?SQL_SERVER_PASSWORD,
  439. ?SQL_BRIDGE,
  440. ?SQL_SERVER_DRIVER,
  441. BatchSize,
  442. QueryMode,
  443. ?WORKER_POOL_SIZE
  444. ]
  445. ),
  446. {Name, parse_and_check(ConfigString, BridgeType, Name)}.
  447. parse_and_check(ConfigString, BridgeType, Name) ->
  448. {ok, RawConf} = hocon:binary(ConfigString, #{format => map}),
  449. hocon_tconf:check_plain(emqx_bridge_schema, RawConf, #{required => false, atom_key => false}),
  450. #{<<"bridges">> := #{BridgeType := #{Name := Config}}} = RawConf,
  451. Config.
  452. create_bridge(Config) ->
  453. create_bridge(Config, _Overrides = #{}).
  454. create_bridge(Config, Overrides) ->
  455. BridgeType = ?config(sqlserver_bridge_type, Config),
  456. Name = ?config(sqlserver_name, Config),
  457. SSConfig0 = ?config(sqlserver_config, Config),
  458. SSConfig = emqx_utils_maps:deep_merge(SSConfig0, Overrides),
  459. emqx_bridge:create(BridgeType, Name, SSConfig).
  460. delete_bridge(Config) ->
  461. BridgeType = ?config(sqlserver_bridge_type, Config),
  462. Name = ?config(sqlserver_name, Config),
  463. emqx_bridge:remove(BridgeType, Name).
  464. create_bridge_http(Params) ->
  465. Path = emqx_mgmt_api_test_util:api_path(["bridges"]),
  466. AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
  467. case emqx_mgmt_api_test_util:request_api(post, Path, "", AuthHeader, Params) of
  468. {ok, Res} -> {ok, emqx_utils_json:decode(Res, [return_maps])};
  469. Error -> Error
  470. end.
  471. send_message(Config, Payload) ->
  472. Name = ?config(sqlserver_name, Config),
  473. BridgeType = ?config(sqlserver_bridge_type, Config),
  474. BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name),
  475. emqx_bridge:send_message(BridgeID, Payload).
  476. query_resource(Config, Request) ->
  477. Name = ?config(sqlserver_name, Config),
  478. BridgeType = ?config(sqlserver_bridge_type, Config),
  479. ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
  480. emqx_resource:query(ResourceID, Request, #{timeout => 1_000}).
  481. query_resource_async(Config, Request) ->
  482. Name = ?config(sqlserver_name, Config),
  483. BridgeType = ?config(sqlserver_bridge_type, Config),
  484. Ref = alias([reply]),
  485. AsyncReplyFun = fun(Result) -> Ref ! {result, Ref, Result} end,
  486. ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
  487. Return = emqx_resource:query(ResourceID, Request, #{
  488. timeout => 500, async_reply_fun => {AsyncReplyFun, []}
  489. }),
  490. {Return, Ref}.
  491. resource_id(Config) ->
  492. Name = ?config(sqlserver_name, Config),
  493. BridgeType = ?config(sqlserver_bridge_type, Config),
  494. _ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name).
  495. health_check_resource_ok(Config) ->
  496. ?assertEqual({ok, connected}, emqx_resource_manager:health_check(resource_id(Config))).
  497. health_check_resource_down(Config) ->
  498. case emqx_resource_manager:health_check(resource_id(Config)) of
  499. {ok, Status} when Status =:= disconnected orelse Status =:= connecting ->
  500. ok;
  501. {error, timeout} ->
  502. ok;
  503. Other ->
  504. ?assert(
  505. false, lists:flatten(io_lib:format("invalid health check result:~p~n", [Other]))
  506. )
  507. end.
  508. receive_result(Ref, Timeout) ->
  509. receive
  510. {result, Ref, Result} ->
  511. {ok, Result};
  512. {Ref, Result} ->
  513. {ok, Result}
  514. after Timeout ->
  515. timeout
  516. end.
  517. connect_direct_sqlserver(Config) ->
  518. Opts = [
  519. {host, ?config(sqlserver_host, Config)},
  520. {port, ?config(sqlserver_port, Config)},
  521. {username, ?SQL_SERVER_USERNAME},
  522. {password, ?SQL_SERVER_PASSWORD},
  523. {driver, ?SQL_SERVER_DRIVER},
  524. {pool_size, 8}
  525. ],
  526. {ok, Con} = connect(Opts),
  527. Con.
  528. connect(Options) ->
  529. ConnectStr = lists:concat(conn_str(Options, [])),
  530. Opts = proplists:get_value(options, Options, []),
  531. odbc:connect(ConnectStr, Opts).
  532. disconnect(Ref) ->
  533. odbc:disconnect(Ref).
  534. % These funs connect and then stop the sqlserver connection
  535. connect_and_create_db_and_table(Config) ->
  536. ?WITH_CON(begin
  537. {updated, undefined} = directly_query(Con, ?SQL_CREATE_DATABASE_IF_NOT_EXISTS),
  538. {updated, undefined} = directly_query(Con, ?SQL_CREATE_TABLE_IN_DB_MQTT)
  539. end).
  540. connect_and_drop_db(Config) ->
  541. ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_DB_MQTT)).
  542. connect_and_drop_table(Config) ->
  543. ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_TABLE)).
  544. connect_and_clear_table(Config) ->
  545. ?WITH_CON({updated, _} = directly_query(Con, ?SQL_DELETE)).
  546. connect_and_get_payload(Config) ->
  547. ?WITH_CON(
  548. {selected, ["payload"], Rows} = directly_query(Con, ?SQL_SELECT)
  549. ),
  550. Rows.
  551. connect_and_get_count(Config) ->
  552. ?WITH_CON(
  553. {selected, [[]], [{Count}]} = directly_query(Con, ?SQL_SELECT_COUNT)
  554. ),
  555. Count.
  556. directly_query(Con, Query) ->
  557. directly_query(Con, Query, ?REQUEST_TIMEOUT_MS).
  558. directly_query(Con, Query, Timeout) ->
  559. odbc:sql_query(Con, Query, Timeout).
  560. %%--------------------------------------------------------------------
  561. %% help functions
  562. %%--------------------------------------------------------------------
  563. batch_size(Config) ->
  564. case ?config(enable_batch, Config) of
  565. true -> ?BATCH_SIZE;
  566. false -> 1
  567. end.
  568. conn_str([], Acc) ->
  569. %% TODO: for msodbc 18+, we need to add "Encrypt=YES;TrustServerCertificate=YES"
  570. %% but havn't tested now
  571. %% we should use this for msodbcsql 18+
  572. %% lists:join(";", ["Encrypt=YES", "TrustServerCertificate=YES" | Acc]);
  573. lists:join(";", Acc);
  574. conn_str([{driver, Driver} | Opts], Acc) ->
  575. conn_str(Opts, ["Driver=" ++ str(Driver) | Acc]);
  576. conn_str([{host, Host} | Opts], Acc) ->
  577. Port = proplists:get_value(port, Opts, str(?SQLSERVER_DEFAULT_PORT)),
  578. NOpts = proplists:delete(port, Opts),
  579. conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
  580. conn_str([{port, Port} | Opts], Acc) ->
  581. Host = proplists:get_value(host, Opts, "localhost"),
  582. NOpts = proplists:delete(host, Opts),
  583. conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
  584. conn_str([{database, Database} | Opts], Acc) ->
  585. conn_str(Opts, ["Database=" ++ str(Database) | Acc]);
  586. conn_str([{username, Username} | Opts], Acc) ->
  587. conn_str(Opts, ["UID=" ++ str(Username) | Acc]);
  588. conn_str([{password, Password} | Opts], Acc) ->
  589. conn_str(Opts, ["PWD=" ++ str(Password) | Acc]);
  590. conn_str([{_, _} | Opts], Acc) ->
  591. conn_str(Opts, Acc).
  592. sent_data(Payload) ->
  593. #{
  594. payload => to_bin(Payload),
  595. id => <<"0005F8F84FFFAFB9F44200000D810002">>,
  596. topic => <<"test/topic">>,
  597. qos => 0
  598. }.
  599. gen_batch_req(Count) when
  600. is_integer(Count) andalso Count > 0
  601. ->
  602. Vals = [{str(erlang:unique_integer())} || _Seq <- lists:seq(1, Count)],
  603. Requests = [{send_message, sent_data(Payload)} || {Payload} <- Vals],
  604. {Requests, Vals};
  605. gen_batch_req(Count) ->
  606. ct:pal("Gen batch requests failed with unexpected Count: ~p", [Count]).
  607. str(List) when is_list(List) ->
  608. unicode:characters_to_list(List, utf8);
  609. str(Bin) when is_binary(Bin) ->
  610. unicode:characters_to_list(Bin, utf8);
  611. str(Num) when is_number(Num) ->
  612. number_to_list(Num).
  613. number_to_list(Int) when is_integer(Int) ->
  614. integer_to_list(Int);
  615. number_to_list(Float) when is_float(Float) ->
  616. float_to_list(Float, [{decimals, 10}, compact]).
  617. to_bin(List) when is_list(List) ->
  618. unicode:characters_to_binary(List, utf8);
  619. to_bin(Bin) when is_binary(Bin) ->
  620. Bin.