emqx_bridge_sqlserver_SUITE.erl 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693
  1. %%--------------------------------------------------------------------
  2. % Copyright (c) 2022-2024 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. Passfile = filename:join(?config(priv_dir, Config), "passfile"),
  118. ok = file:write_file(Passfile, <<?SQL_SERVER_PASSWORD>>),
  119. [{sqlserver_passfile, Passfile} | Config].
  120. end_per_suite(_Config) ->
  121. emqx_mgmt_api_test_util:end_suite(),
  122. ok = emqx_common_test_helpers:stop_apps([emqx_bridge, emqx_conf]),
  123. ok.
  124. init_per_testcase(_Testcase, Config) ->
  125. %% drop database and table
  126. %% connect_and_clear_table(Config),
  127. %% create a new one
  128. %% TODO: create a new database for each test case
  129. delete_bridge(Config),
  130. snabbkaffe:start_trace(),
  131. Config.
  132. end_per_testcase(_Testcase, Config) ->
  133. ProxyHost = ?config(proxy_host, Config),
  134. ProxyPort = ?config(proxy_port, Config),
  135. emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
  136. connect_and_clear_table(Config),
  137. ok = snabbkaffe:stop(),
  138. delete_bridge(Config),
  139. ok.
  140. %%------------------------------------------------------------------------------
  141. %% Testcases
  142. %%------------------------------------------------------------------------------
  143. t_setup_via_config_and_publish(Config) ->
  144. ?assertMatch(
  145. {ok, _},
  146. create_bridge(Config)
  147. ),
  148. Val = str(erlang:unique_integer()),
  149. SentData = sent_data(Val),
  150. ?check_trace(
  151. begin
  152. ?wait_async_action(
  153. ?assertEqual(ok, send_message(Config, SentData)),
  154. #{?snk_kind := sqlserver_connector_query_return},
  155. 10_000
  156. ),
  157. ?assertMatch(
  158. [{Val}],
  159. connect_and_get_payload(Config)
  160. ),
  161. ok
  162. end,
  163. fun(Trace0) ->
  164. Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
  165. ?assertMatch([#{result := ok}], Trace),
  166. ok
  167. end
  168. ),
  169. ok.
  170. t_setup_via_http_api_and_publish(Config) ->
  171. BridgeType = ?config(sqlserver_bridge_type, Config),
  172. Name = ?config(sqlserver_name, Config),
  173. SQLServerConfig0 = ?config(sqlserver_config, Config),
  174. SQLServerConfig = SQLServerConfig0#{
  175. <<"name">> => Name,
  176. <<"type">> => BridgeType,
  177. %% NOTE: using literal password with HTTP API requests.
  178. <<"password">> => <<?SQL_SERVER_PASSWORD>>
  179. },
  180. ?assertMatch(
  181. {ok, _},
  182. create_bridge_http(SQLServerConfig)
  183. ),
  184. Val = str(erlang:unique_integer()),
  185. SentData = sent_data(Val),
  186. ?check_trace(
  187. begin
  188. ?wait_async_action(
  189. ?assertEqual(ok, send_message(Config, SentData)),
  190. #{?snk_kind := sqlserver_connector_query_return},
  191. 10_000
  192. ),
  193. ?assertMatch(
  194. [{Val}],
  195. connect_and_get_payload(Config)
  196. ),
  197. ok
  198. end,
  199. fun(Trace0) ->
  200. Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
  201. ?assertMatch([#{result := ok}], Trace),
  202. ok
  203. end
  204. ),
  205. ok.
  206. t_get_status(Config) ->
  207. ?assertMatch(
  208. {ok, _},
  209. create_bridge(Config)
  210. ),
  211. ProxyPort = ?config(proxy_port, Config),
  212. ProxyHost = ?config(proxy_host, Config),
  213. ProxyName = ?config(proxy_name, Config),
  214. health_check_resource_ok(Config),
  215. emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
  216. health_check_resource_down(Config)
  217. end),
  218. ok.
  219. t_create_disconnected(Config) ->
  220. ProxyPort = ?config(proxy_port, Config),
  221. ProxyHost = ?config(proxy_host, Config),
  222. ProxyName = ?config(proxy_name, Config),
  223. emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
  224. ?assertMatch({ok, _}, create_bridge(Config)),
  225. health_check_resource_down(Config)
  226. end),
  227. health_check_resource_ok(Config),
  228. ok.
  229. t_create_with_invalid_password(Config) ->
  230. BridgeType = ?config(sqlserver_bridge_type, Config),
  231. Name = ?config(sqlserver_name, Config),
  232. SQLServerConfig0 = ?config(sqlserver_config, Config),
  233. SQLServerConfig = SQLServerConfig0#{
  234. <<"name">> => Name,
  235. <<"type">> => BridgeType,
  236. <<"password">> => <<"wrong_password">>
  237. },
  238. ?check_trace(
  239. begin
  240. ?assertMatch(
  241. {ok, _},
  242. create_bridge_http(SQLServerConfig)
  243. )
  244. end,
  245. fun(Trace) ->
  246. ?assertMatch(
  247. [#{error := {start_pool_failed, _, _}}],
  248. ?of_kind(sqlserver_connector_start_failed, Trace)
  249. ),
  250. ok
  251. end
  252. ),
  253. ok.
  254. t_write_failure(Config) ->
  255. ProxyName = ?config(proxy_name, Config),
  256. ProxyPort = ?config(proxy_port, Config),
  257. ProxyHost = ?config(proxy_host, Config),
  258. QueryMode = ?config(query_mode, Config),
  259. Val = str(erlang:unique_integer()),
  260. SentData = sent_data(Val),
  261. {{ok, _}, {ok, _}} =
  262. ?wait_async_action(
  263. create_bridge(Config),
  264. #{?snk_kind := resource_connected_enter},
  265. 20_000
  266. ),
  267. emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
  268. case QueryMode of
  269. sync ->
  270. ?assertMatch(
  271. {error, {resource_error, #{reason := timeout}}},
  272. send_message(Config, SentData)
  273. );
  274. async ->
  275. ?assertMatch(
  276. ok, send_message(Config, SentData)
  277. )
  278. end
  279. end),
  280. ok.
  281. t_write_timeout(_Config) ->
  282. %% msodbc driver handled all connection exceptions
  283. %% the case is same as t_write_failure/1
  284. ok.
  285. t_simple_query(Config) ->
  286. BatchSize = batch_size(Config),
  287. ?assertMatch(
  288. {ok, _},
  289. create_bridge(Config)
  290. ),
  291. {Requests, Vals} = gen_batch_req(BatchSize),
  292. ?check_trace(
  293. begin
  294. ?wait_async_action(
  295. begin
  296. [?assertEqual(ok, query_resource(Config, Request)) || Request <- Requests]
  297. end,
  298. #{?snk_kind := sqlserver_connector_query_return},
  299. 10_000
  300. ),
  301. %% just assert the data count is correct
  302. ?assertMatch(
  303. BatchSize,
  304. connect_and_get_count(Config)
  305. ),
  306. %% assert the data order is correct
  307. ?assertMatch(
  308. Vals,
  309. connect_and_get_payload(Config)
  310. )
  311. end,
  312. fun(Trace0) ->
  313. Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
  314. case BatchSize of
  315. 1 ->
  316. ?assertMatch([#{result := ok}], Trace);
  317. _ ->
  318. [?assertMatch(#{result := ok}, Trace1) || Trace1 <- Trace]
  319. end,
  320. ok
  321. end
  322. ),
  323. ok.
  324. -define(MISSING_TINYINT_ERROR,
  325. "[Microsoft][ODBC Driver 17 for SQL Server][SQL Server]"
  326. "Conversion failed when converting the varchar value 'undefined' to data type tinyint. SQLSTATE IS: 22018"
  327. ).
  328. t_missing_data(Config) ->
  329. QueryMode = ?config(query_mode, Config),
  330. ?assertMatch(
  331. {ok, _},
  332. create_bridge(Config)
  333. ),
  334. Result = send_message(Config, #{}),
  335. case QueryMode of
  336. sync ->
  337. ?assertMatch(
  338. {error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}},
  339. Result
  340. );
  341. async ->
  342. ?assertMatch(
  343. ok, send_message(Config, #{})
  344. )
  345. end,
  346. ok.
  347. t_bad_parameter(Config) ->
  348. QueryMode = ?config(query_mode, Config),
  349. ?assertMatch(
  350. {ok, _},
  351. create_bridge(Config)
  352. ),
  353. Result = send_message(Config, #{}),
  354. case QueryMode of
  355. sync ->
  356. ?assertMatch(
  357. {error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}},
  358. Result
  359. );
  360. async ->
  361. ?assertMatch(
  362. ok, send_message(Config, #{})
  363. )
  364. end,
  365. ok.
  366. %%------------------------------------------------------------------------------
  367. %% Helper fns
  368. %%------------------------------------------------------------------------------
  369. common_init(ConfigT) ->
  370. Host = os:getenv("SQLSERVER_HOST", "toxiproxy"),
  371. Port = list_to_integer(os:getenv("SQLSERVER_PORT", str(?SQLSERVER_DEFAULT_PORT))),
  372. Config0 = [
  373. {sqlserver_host, Host},
  374. {sqlserver_port, Port},
  375. %% see also for `proxy_name` : $PROJ_ROOT/.ci/docker-compose-file/toxiproxy.json
  376. {proxy_name, "sqlserver"},
  377. {batch_size, batch_size(ConfigT)}
  378. | ConfigT
  379. ],
  380. BridgeType = proplists:get_value(bridge_type, Config0, <<"sqlserver">>),
  381. case emqx_common_test_helpers:is_tcp_server_available(Host, Port) of
  382. true ->
  383. % Setup toxiproxy
  384. ProxyHost = os:getenv("PROXY_HOST", "toxiproxy"),
  385. ProxyPort = list_to_integer(os:getenv("PROXY_PORT", "8474")),
  386. emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
  387. % Ensure enterprise bridge module is loaded
  388. ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_bridge, odbc]),
  389. _ = emqx_bridge_enterprise:module_info(),
  390. emqx_mgmt_api_test_util:init_suite(),
  391. % Connect to sqlserver directly
  392. % drop old db and table, and then create new ones
  393. connect_and_create_db_and_table(Config0),
  394. {Name, SQLServerConf} = sqlserver_config(BridgeType, Config0),
  395. Config =
  396. [
  397. {sqlserver_config, SQLServerConf},
  398. {sqlserver_bridge_type, BridgeType},
  399. {sqlserver_name, Name},
  400. {proxy_host, ProxyHost},
  401. {proxy_port, ProxyPort}
  402. | Config0
  403. ],
  404. Config;
  405. false ->
  406. case os:getenv("IS_CI") of
  407. "yes" ->
  408. throw(no_sqlserver);
  409. _ ->
  410. {skip, no_sqlserver}
  411. end
  412. end.
  413. sqlserver_config(BridgeType, Config) ->
  414. Port = integer_to_list(?config(sqlserver_port, Config)),
  415. Server = ?config(sqlserver_host, Config) ++ ":" ++ Port,
  416. Name = atom_to_binary(?MODULE),
  417. BatchSize = batch_size(Config),
  418. QueryMode = ?config(query_mode, Config),
  419. Passfile = ?config(sqlserver_passfile, Config),
  420. ConfigString =
  421. io_lib:format(
  422. "bridges.~s.~s {\n"
  423. " enable = true\n"
  424. " server = ~p\n"
  425. " database = ~p\n"
  426. " username = ~p\n"
  427. " password = ~p\n"
  428. " sql = ~p\n"
  429. " driver = ~p\n"
  430. " resource_opts = {\n"
  431. " request_ttl = 500ms\n"
  432. " batch_size = ~b\n"
  433. " query_mode = ~s\n"
  434. " worker_pool_size = ~b\n"
  435. " }\n"
  436. "}",
  437. [
  438. BridgeType,
  439. Name,
  440. Server,
  441. ?SQL_SERVER_DATABASE,
  442. ?SQL_SERVER_USERNAME,
  443. "file://" ++ Passfile,
  444. ?SQL_BRIDGE,
  445. ?SQL_SERVER_DRIVER,
  446. BatchSize,
  447. QueryMode,
  448. ?WORKER_POOL_SIZE
  449. ]
  450. ),
  451. {Name, parse_and_check(ConfigString, BridgeType, Name)}.
  452. parse_and_check(ConfigString, BridgeType, Name) ->
  453. {ok, RawConf} = hocon:binary(ConfigString, #{format => map}),
  454. hocon_tconf:check_plain(emqx_bridge_schema, RawConf, #{required => false, atom_key => false}),
  455. #{<<"bridges">> := #{BridgeType := #{Name := Config}}} = RawConf,
  456. Config.
  457. create_bridge(Config) ->
  458. create_bridge(Config, _Overrides = #{}).
  459. create_bridge(Config, Overrides) ->
  460. BridgeType = ?config(sqlserver_bridge_type, Config),
  461. Name = ?config(sqlserver_name, Config),
  462. SSConfig0 = ?config(sqlserver_config, Config),
  463. SSConfig = emqx_utils_maps:deep_merge(SSConfig0, Overrides),
  464. emqx_bridge:create(BridgeType, Name, SSConfig).
  465. delete_bridge(Config) ->
  466. BridgeType = ?config(sqlserver_bridge_type, Config),
  467. Name = ?config(sqlserver_name, Config),
  468. emqx_bridge:remove(BridgeType, Name).
  469. create_bridge_http(Params) ->
  470. Path = emqx_mgmt_api_test_util:api_path(["bridges"]),
  471. AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
  472. case emqx_mgmt_api_test_util:request_api(post, Path, "", AuthHeader, Params) of
  473. {ok, Res} -> {ok, emqx_utils_json:decode(Res, [return_maps])};
  474. Error -> Error
  475. end.
  476. send_message(Config, Payload) ->
  477. Name = ?config(sqlserver_name, Config),
  478. BridgeType = ?config(sqlserver_bridge_type, Config),
  479. BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name),
  480. emqx_bridge:send_message(BridgeID, Payload).
  481. query_resource(Config, Request) ->
  482. Name = ?config(sqlserver_name, Config),
  483. BridgeType = ?config(sqlserver_bridge_type, Config),
  484. ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
  485. emqx_resource:query(ResourceID, Request, #{timeout => 1_000}).
  486. query_resource_async(Config, Request) ->
  487. Name = ?config(sqlserver_name, Config),
  488. BridgeType = ?config(sqlserver_bridge_type, Config),
  489. Ref = alias([reply]),
  490. AsyncReplyFun = fun(Result) -> Ref ! {result, Ref, Result} end,
  491. ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
  492. Return = emqx_resource:query(ResourceID, Request, #{
  493. timeout => 500, async_reply_fun => {AsyncReplyFun, []}
  494. }),
  495. {Return, Ref}.
  496. resource_id(Config) ->
  497. Name = ?config(sqlserver_name, Config),
  498. BridgeType = ?config(sqlserver_bridge_type, Config),
  499. _ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name).
  500. health_check_resource_ok(Config) ->
  501. ?assertEqual({ok, connected}, emqx_resource_manager:health_check(resource_id(Config))).
  502. health_check_resource_down(Config) ->
  503. case emqx_resource_manager:health_check(resource_id(Config)) of
  504. {ok, Status} when Status =:= disconnected orelse Status =:= connecting ->
  505. ok;
  506. {error, timeout} ->
  507. ok;
  508. Other ->
  509. ?assert(
  510. false, lists:flatten(io_lib:format("invalid health check result:~p~n", [Other]))
  511. )
  512. end.
  513. receive_result(Ref, Timeout) ->
  514. receive
  515. {result, Ref, Result} ->
  516. {ok, Result};
  517. {Ref, Result} ->
  518. {ok, Result}
  519. after Timeout ->
  520. timeout
  521. end.
  522. connect_direct_sqlserver(Config) ->
  523. Opts = [
  524. {host, ?config(sqlserver_host, Config)},
  525. {port, ?config(sqlserver_port, Config)},
  526. {username, ?SQL_SERVER_USERNAME},
  527. {password, ?SQL_SERVER_PASSWORD},
  528. {driver, ?SQL_SERVER_DRIVER},
  529. {pool_size, 8}
  530. ],
  531. {ok, Con} = connect(Opts),
  532. Con.
  533. connect(Options) ->
  534. ConnectStr = lists:concat(conn_str(Options, [])),
  535. Opts = proplists:get_value(options, Options, []),
  536. odbc:connect(ConnectStr, Opts).
  537. disconnect(Ref) ->
  538. odbc:disconnect(Ref).
  539. % These funs connect and then stop the sqlserver connection
  540. connect_and_create_db_and_table(Config) ->
  541. ?WITH_CON(begin
  542. {updated, undefined} = directly_query(Con, ?SQL_CREATE_DATABASE_IF_NOT_EXISTS),
  543. {updated, undefined} = directly_query(Con, ?SQL_CREATE_TABLE_IN_DB_MQTT)
  544. end).
  545. connect_and_drop_db(Config) ->
  546. ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_DB_MQTT)).
  547. connect_and_drop_table(Config) ->
  548. ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_TABLE)).
  549. connect_and_clear_table(Config) ->
  550. ?WITH_CON({updated, _} = directly_query(Con, ?SQL_DELETE)).
  551. connect_and_get_payload(Config) ->
  552. ?WITH_CON(
  553. {selected, ["payload"], Rows} = directly_query(Con, ?SQL_SELECT)
  554. ),
  555. Rows.
  556. connect_and_get_count(Config) ->
  557. ?WITH_CON(
  558. {selected, [[]], [{Count}]} = directly_query(Con, ?SQL_SELECT_COUNT)
  559. ),
  560. Count.
  561. directly_query(Con, Query) ->
  562. directly_query(Con, Query, ?REQUEST_TIMEOUT_MS).
  563. directly_query(Con, Query, Timeout) ->
  564. odbc:sql_query(Con, Query, Timeout).
  565. %%--------------------------------------------------------------------
  566. %% help functions
  567. %%--------------------------------------------------------------------
  568. batch_size(Config) ->
  569. case ?config(enable_batch, Config) of
  570. true -> ?BATCH_SIZE;
  571. false -> 1
  572. end.
  573. conn_str([], Acc) ->
  574. %% TODO: for msodbc 18+, we need to add "Encrypt=YES;TrustServerCertificate=YES"
  575. %% but havn't tested now
  576. %% we should use this for msodbcsql 18+
  577. %% lists:join(";", ["Encrypt=YES", "TrustServerCertificate=YES" | Acc]);
  578. lists:join(";", Acc);
  579. conn_str([{driver, Driver} | Opts], Acc) ->
  580. conn_str(Opts, ["Driver=" ++ str(Driver) | Acc]);
  581. conn_str([{host, Host} | Opts], Acc) ->
  582. Port = proplists:get_value(port, Opts, str(?SQLSERVER_DEFAULT_PORT)),
  583. NOpts = proplists:delete(port, Opts),
  584. conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
  585. conn_str([{port, Port} | Opts], Acc) ->
  586. Host = proplists:get_value(host, Opts, "localhost"),
  587. NOpts = proplists:delete(host, Opts),
  588. conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
  589. conn_str([{database, Database} | Opts], Acc) ->
  590. conn_str(Opts, ["Database=" ++ str(Database) | Acc]);
  591. conn_str([{username, Username} | Opts], Acc) ->
  592. conn_str(Opts, ["UID=" ++ str(Username) | Acc]);
  593. conn_str([{password, Password} | Opts], Acc) ->
  594. conn_str(Opts, ["PWD=" ++ str(Password) | Acc]);
  595. conn_str([{_, _} | Opts], Acc) ->
  596. conn_str(Opts, Acc).
  597. sent_data(Payload) ->
  598. #{
  599. payload => to_bin(Payload),
  600. id => <<"0005F8F84FFFAFB9F44200000D810002">>,
  601. topic => <<"test/topic">>,
  602. qos => 0
  603. }.
  604. gen_batch_req(Count) when
  605. is_integer(Count) andalso Count > 0
  606. ->
  607. Vals = [{str(erlang:unique_integer())} || _Seq <- lists:seq(1, Count)],
  608. Requests = [{send_message, sent_data(Payload)} || {Payload} <- Vals],
  609. {Requests, Vals};
  610. gen_batch_req(Count) ->
  611. ct:pal("Gen batch requests failed with unexpected Count: ~p", [Count]).
  612. str(List) when is_list(List) ->
  613. unicode:characters_to_list(List, utf8);
  614. str(Bin) when is_binary(Bin) ->
  615. unicode:characters_to_list(Bin, utf8);
  616. str(Num) when is_number(Num) ->
  617. number_to_list(Num).
  618. number_to_list(Int) when is_integer(Int) ->
  619. integer_to_list(Int);
  620. number_to_list(Float) when is_float(Float) ->
  621. float_to_list(Float, [{decimals, 10}, compact]).
  622. to_bin(List) when is_list(List) ->
  623. unicode:characters_to_binary(List, utf8);
  624. to_bin(Bin) when is_binary(Bin) ->
  625. Bin.