emqx_bridge_sqlserver_SUITE.erl 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689
  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 EE bridge module is loaded
  384. _ = application:load(emqx_ee_bridge),
  385. _ = emqx_ee_bridge:module_info(),
  386. ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_bridge]),
  387. emqx_mgmt_api_test_util:init_suite(),
  388. % Connect to sqlserver directly
  389. % drop old db and table, and then create new ones
  390. connect_and_create_db_and_table(Config0),
  391. {Name, SQLServerConf} = sqlserver_config(BridgeType, Config0),
  392. Config =
  393. [
  394. {sqlserver_config, SQLServerConf},
  395. {sqlserver_bridge_type, BridgeType},
  396. {sqlserver_name, Name},
  397. {proxy_host, ProxyHost},
  398. {proxy_port, ProxyPort}
  399. | Config0
  400. ],
  401. Config;
  402. false ->
  403. case os:getenv("IS_CI") of
  404. "yes" ->
  405. throw(no_sqlserver);
  406. _ ->
  407. {skip, no_sqlserver}
  408. end
  409. end.
  410. sqlserver_config(BridgeType, Config) ->
  411. Port = integer_to_list(?config(sqlserver_port, Config)),
  412. Server = ?config(sqlserver_host, Config) ++ ":" ++ Port,
  413. Name = atom_to_binary(?MODULE),
  414. BatchSize = batch_size(Config),
  415. QueryMode = ?config(query_mode, Config),
  416. ConfigString =
  417. io_lib:format(
  418. "bridges.~s.~s {\n"
  419. " enable = true\n"
  420. " server = ~p\n"
  421. " database = ~p\n"
  422. " username = ~p\n"
  423. " password = ~p\n"
  424. " sql = ~p\n"
  425. " driver = ~p\n"
  426. " resource_opts = {\n"
  427. " request_ttl = 500ms\n"
  428. " batch_size = ~b\n"
  429. " query_mode = ~s\n"
  430. " worker_pool_size = ~b\n"
  431. " }\n"
  432. "}",
  433. [
  434. BridgeType,
  435. Name,
  436. Server,
  437. ?SQL_SERVER_DATABASE,
  438. ?SQL_SERVER_USERNAME,
  439. ?SQL_SERVER_PASSWORD,
  440. ?SQL_BRIDGE,
  441. ?SQL_SERVER_DRIVER,
  442. BatchSize,
  443. QueryMode,
  444. ?WORKER_POOL_SIZE
  445. ]
  446. ),
  447. {Name, parse_and_check(ConfigString, BridgeType, Name)}.
  448. parse_and_check(ConfigString, BridgeType, Name) ->
  449. {ok, RawConf} = hocon:binary(ConfigString, #{format => map}),
  450. hocon_tconf:check_plain(emqx_bridge_schema, RawConf, #{required => false, atom_key => false}),
  451. #{<<"bridges">> := #{BridgeType := #{Name := Config}}} = RawConf,
  452. Config.
  453. create_bridge(Config) ->
  454. create_bridge(Config, _Overrides = #{}).
  455. create_bridge(Config, Overrides) ->
  456. BridgeType = ?config(sqlserver_bridge_type, Config),
  457. Name = ?config(sqlserver_name, Config),
  458. SSConfig0 = ?config(sqlserver_config, Config),
  459. SSConfig = emqx_utils_maps:deep_merge(SSConfig0, Overrides),
  460. emqx_bridge:create(BridgeType, Name, SSConfig).
  461. delete_bridge(Config) ->
  462. BridgeType = ?config(sqlserver_bridge_type, Config),
  463. Name = ?config(sqlserver_name, Config),
  464. emqx_bridge:remove(BridgeType, Name).
  465. create_bridge_http(Params) ->
  466. Path = emqx_mgmt_api_test_util:api_path(["bridges"]),
  467. AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
  468. case emqx_mgmt_api_test_util:request_api(post, Path, "", AuthHeader, Params) of
  469. {ok, Res} -> {ok, emqx_utils_json:decode(Res, [return_maps])};
  470. Error -> Error
  471. end.
  472. send_message(Config, Payload) ->
  473. Name = ?config(sqlserver_name, Config),
  474. BridgeType = ?config(sqlserver_bridge_type, Config),
  475. BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name),
  476. emqx_bridge:send_message(BridgeID, Payload).
  477. query_resource(Config, Request) ->
  478. Name = ?config(sqlserver_name, Config),
  479. BridgeType = ?config(sqlserver_bridge_type, Config),
  480. ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
  481. emqx_resource:query(ResourceID, Request, #{timeout => 1_000}).
  482. query_resource_async(Config, Request) ->
  483. Name = ?config(sqlserver_name, Config),
  484. BridgeType = ?config(sqlserver_bridge_type, Config),
  485. Ref = alias([reply]),
  486. AsyncReplyFun = fun(Result) -> Ref ! {result, Ref, Result} end,
  487. ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
  488. Return = emqx_resource:query(ResourceID, Request, #{
  489. timeout => 500, async_reply_fun => {AsyncReplyFun, []}
  490. }),
  491. {Return, Ref}.
  492. resource_id(Config) ->
  493. Name = ?config(sqlserver_name, Config),
  494. BridgeType = ?config(sqlserver_bridge_type, Config),
  495. _ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name).
  496. health_check_resource_ok(Config) ->
  497. ?assertEqual({ok, connected}, emqx_resource_manager:health_check(resource_id(Config))).
  498. health_check_resource_down(Config) ->
  499. case emqx_resource_manager:health_check(resource_id(Config)) of
  500. {ok, Status} when Status =:= disconnected orelse Status =:= connecting ->
  501. ok;
  502. {error, timeout} ->
  503. ok;
  504. Other ->
  505. ?assert(
  506. false, lists:flatten(io_lib:format("invalid health check result:~p~n", [Other]))
  507. )
  508. end.
  509. receive_result(Ref, Timeout) ->
  510. receive
  511. {result, Ref, Result} ->
  512. {ok, Result};
  513. {Ref, Result} ->
  514. {ok, Result}
  515. after Timeout ->
  516. timeout
  517. end.
  518. connect_direct_sqlserver(Config) ->
  519. Opts = [
  520. {host, ?config(sqlserver_host, Config)},
  521. {port, ?config(sqlserver_port, Config)},
  522. {username, ?SQL_SERVER_USERNAME},
  523. {password, ?SQL_SERVER_PASSWORD},
  524. {driver, ?SQL_SERVER_DRIVER},
  525. {pool_size, 8}
  526. ],
  527. {ok, Con} = connect(Opts),
  528. Con.
  529. connect(Options) ->
  530. ConnectStr = lists:concat(conn_str(Options, [])),
  531. Opts = proplists:get_value(options, Options, []),
  532. odbc:connect(ConnectStr, Opts).
  533. disconnect(Ref) ->
  534. odbc:disconnect(Ref).
  535. % These funs connect and then stop the sqlserver connection
  536. connect_and_create_db_and_table(Config) ->
  537. ?WITH_CON(begin
  538. {updated, undefined} = directly_query(Con, ?SQL_CREATE_DATABASE_IF_NOT_EXISTS),
  539. {updated, undefined} = directly_query(Con, ?SQL_CREATE_TABLE_IN_DB_MQTT)
  540. end).
  541. connect_and_drop_db(Config) ->
  542. ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_DB_MQTT)).
  543. connect_and_drop_table(Config) ->
  544. ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_TABLE)).
  545. connect_and_clear_table(Config) ->
  546. ?WITH_CON({updated, _} = directly_query(Con, ?SQL_DELETE)).
  547. connect_and_get_payload(Config) ->
  548. ?WITH_CON(
  549. {selected, ["payload"], Rows} = directly_query(Con, ?SQL_SELECT)
  550. ),
  551. Rows.
  552. connect_and_get_count(Config) ->
  553. ?WITH_CON(
  554. {selected, [[]], [{Count}]} = directly_query(Con, ?SQL_SELECT_COUNT)
  555. ),
  556. Count.
  557. directly_query(Con, Query) ->
  558. directly_query(Con, Query, ?REQUEST_TIMEOUT_MS).
  559. directly_query(Con, Query, Timeout) ->
  560. odbc:sql_query(Con, Query, Timeout).
  561. %%--------------------------------------------------------------------
  562. %% help functions
  563. %%--------------------------------------------------------------------
  564. batch_size(Config) ->
  565. case ?config(enable_batch, Config) of
  566. true -> ?BATCH_SIZE;
  567. false -> 1
  568. end.
  569. conn_str([], Acc) ->
  570. %% TODO: for msodbc 18+, we need to add "Encrypt=YES;TrustServerCertificate=YES"
  571. %% but havn't tested now
  572. %% we should use this for msodbcsql 18+
  573. %% lists:join(";", ["Encrypt=YES", "TrustServerCertificate=YES" | Acc]);
  574. lists:join(";", Acc);
  575. conn_str([{driver, Driver} | Opts], Acc) ->
  576. conn_str(Opts, ["Driver=" ++ str(Driver) | Acc]);
  577. conn_str([{host, Host} | Opts], Acc) ->
  578. Port = proplists:get_value(port, Opts, str(?SQLSERVER_DEFAULT_PORT)),
  579. NOpts = proplists:delete(port, Opts),
  580. conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
  581. conn_str([{port, Port} | Opts], Acc) ->
  582. Host = proplists:get_value(host, Opts, "localhost"),
  583. NOpts = proplists:delete(host, Opts),
  584. conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
  585. conn_str([{database, Database} | Opts], Acc) ->
  586. conn_str(Opts, ["Database=" ++ str(Database) | Acc]);
  587. conn_str([{username, Username} | Opts], Acc) ->
  588. conn_str(Opts, ["UID=" ++ str(Username) | Acc]);
  589. conn_str([{password, Password} | Opts], Acc) ->
  590. conn_str(Opts, ["PWD=" ++ str(Password) | Acc]);
  591. conn_str([{_, _} | Opts], Acc) ->
  592. conn_str(Opts, Acc).
  593. sent_data(Payload) ->
  594. #{
  595. payload => to_bin(Payload),
  596. id => <<"0005F8F84FFFAFB9F44200000D810002">>,
  597. topic => <<"test/topic">>,
  598. qos => 0
  599. }.
  600. gen_batch_req(Count) when
  601. is_integer(Count) andalso Count > 0
  602. ->
  603. Vals = [{str(erlang:unique_integer())} || _Seq <- lists:seq(1, Count)],
  604. Requests = [{send_message, sent_data(Payload)} || {Payload} <- Vals],
  605. {Requests, Vals};
  606. gen_batch_req(Count) ->
  607. ct:pal("Gen batch requests failed with unexpected Count: ~p", [Count]).
  608. str(List) when is_list(List) ->
  609. unicode:characters_to_list(List, utf8);
  610. str(Bin) when is_binary(Bin) ->
  611. unicode:characters_to_list(Bin, utf8);
  612. str(Num) when is_number(Num) ->
  613. number_to_list(Num).
  614. number_to_list(Int) when is_integer(Int) ->
  615. integer_to_list(Int);
  616. number_to_list(Float) when is_float(Float) ->
  617. float_to_list(Float, [{decimals, 10}, compact]).
  618. to_bin(List) when is_list(List) ->
  619. unicode:characters_to_binary(List, utf8);
  620. to_bin(Bin) when is_binary(Bin) ->
  621. Bin.