diff --git a/.ci/docker-compose-file/.env b/.ci/docker-compose-file/.env
index 956750e00..d33637ea0 100644
--- a/.ci/docker-compose-file/.env
+++ b/.ci/docker-compose-file/.env
@@ -8,4 +8,7 @@ TDENGINE_TAG=3.0.2.4
DYNAMO_TAG=1.21.0
CASSANDRA_TAG=3.11.6
+MS_IMAGE_ADDR=mcr.microsoft.com/mssql/server
+SQLSERVER_TAG=2019-CU19-ubuntu-20.04
+
TARGET=emqx/emqx
diff --git a/.ci/docker-compose-file/docker-compose-sqlserver.yaml b/.ci/docker-compose-file/docker-compose-sqlserver.yaml
new file mode 100644
index 000000000..63fcfeecd
--- /dev/null
+++ b/.ci/docker-compose-file/docker-compose-sqlserver.yaml
@@ -0,0 +1,19 @@
+version: '3.9'
+
+services:
+ sql_server:
+ container_name: sqlserver
+ # See also:
+ # https://mcr.microsoft.com/en-us/product/mssql/server/about
+ # https://hub.docker.com/_/microsoft-mssql-server
+ image: ${MS_IMAGE_ADDR}:${SQLSERVER_TAG}
+ environment:
+ # See also:
+ # https://learn.microsoft.com/en-us/sql/linux/sql-server-linux-configure-environment-variables
+ ACCEPT_EULA: "Y"
+ MSSQL_SA_PASSWORD: "mqtt_public1"
+ restart: always
+ # ports:
+ # - "1433:1433"
+ networks:
+ - emqx_bridge
diff --git a/.ci/docker-compose-file/docker-compose-toxiproxy.yaml b/.ci/docker-compose-file/docker-compose-toxiproxy.yaml
index 9a1d08ba6..ba5e831a5 100644
--- a/.ci/docker-compose-file/docker-compose-toxiproxy.yaml
+++ b/.ci/docker-compose-file/docker-compose-toxiproxy.yaml
@@ -16,6 +16,7 @@ services:
- 8474:8474
- 8086:8086
- 8087:8087
+ - 11433:1433
- 13306:3306
- 13307:3307
- 15432:5432
diff --git a/.ci/docker-compose-file/docker-compose.yaml b/.ci/docker-compose-file/docker-compose.yaml
index 48d900400..6f4b7c04b 100644
--- a/.ci/docker-compose-file/docker-compose.yaml
+++ b/.ci/docker-compose-file/docker-compose.yaml
@@ -24,6 +24,7 @@ services:
- /tmp/emqx-ci/emqx-shared-secret:/var/lib/secret
- ./kerberos/krb5.conf:/etc/kdc/krb5.conf
- ./kerberos/krb5.conf:/etc/krb5.conf
+ # - ./odbc/odbcinst.ini:/etc/odbcinst.ini
working_dir: /emqx
tty: true
user: "${DOCKER_USER:-root}"
diff --git a/.ci/docker-compose-file/odbc/odbcinst.ini b/.ci/docker-compose-file/odbc/odbcinst.ini
new file mode 100644
index 000000000..dd0241543
--- /dev/null
+++ b/.ci/docker-compose-file/odbc/odbcinst.ini
@@ -0,0 +1,9 @@
+[ms-sql]
+Description=Microsoft ODBC Driver 17 for SQL Server
+Driver=/opt/microsoft/msodbcsql17/lib64/libmsodbcsql-17.10.so.2.1
+UsageCount=1
+
+[ODBC Driver 17 for SQL Server]
+Description=Microsoft ODBC Driver 17 for SQL Server
+Driver=/opt/microsoft/msodbcsql17/lib64/libmsodbcsql-17.10.so.2.1
+UsageCount=1
diff --git a/.ci/docker-compose-file/toxiproxy.json b/.ci/docker-compose-file/toxiproxy.json
index 708cbf1ef..da2dff763 100644
--- a/.ci/docker-compose-file/toxiproxy.json
+++ b/.ci/docker-compose-file/toxiproxy.json
@@ -95,5 +95,11 @@
"listen": "0.0.0.0:9142",
"upstream": "cassandra:9142",
"enabled": true
+ },
+ {
+ "name": "sqlserver",
+ "listen": "0.0.0.0:1433",
+ "upstream": "sqlserver:1433",
+ "enabled": true
}
]
diff --git a/apps/emqx_bridge/src/emqx_bridge.erl b/apps/emqx_bridge/src/emqx_bridge.erl
index 6ff81a10b..2a20c5994 100644
--- a/apps/emqx_bridge/src/emqx_bridge.erl
+++ b/apps/emqx_bridge/src/emqx_bridge.erl
@@ -69,7 +69,8 @@
T == tdengine;
T == dynamo;
T == rocketmq;
- T == cassandra
+ T == cassandra;
+ T == sqlserver
).
load() ->
diff --git a/apps/emqx_connector/src/emqx_connector_mysql.erl b/apps/emqx_connector/src/emqx_connector_mysql.erl
index fe495252a..c69cf5ca0 100644
--- a/apps/emqx_connector/src/emqx_connector_mysql.erl
+++ b/apps/emqx_connector/src/emqx_connector_mysql.erl
@@ -437,7 +437,7 @@ do_sql_query(SQLFunc, Conn, SQLOrKey, Params, Timeout, LogMeta) ->
error,
LogMeta#{msg => "mysql_connector_do_sql_query_failed", reason => disconnected}
),
- %% kill the poll worker to trigger reconnection
+ %% kill the pool worker to trigger reconnection
_ = exit(Conn, restart),
{error, {recoverable_error, disconnected}};
{error, not_prepared} = Error ->
diff --git a/changes/ee/feat-10363.en.md b/changes/ee/feat-10363.en.md
new file mode 100644
index 000000000..84816031d
--- /dev/null
+++ b/changes/ee/feat-10363.en.md
@@ -0,0 +1 @@
+Implement Microsoft SQL Server bridge.
diff --git a/lib-ee/emqx_ee_bridge/docker-ct b/lib-ee/emqx_ee_bridge/docker-ct
index 0e947d89e..35d6b9d5b 100644
--- a/lib-ee/emqx_ee_bridge/docker-ct
+++ b/lib-ee/emqx_ee_bridge/docker-ct
@@ -11,3 +11,4 @@ clickhouse
dynamo
rocketmq
cassandra
+sqlserver
diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl
index 84b0b98b0..3ad5cbbb4 100644
--- a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl
+++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl
@@ -34,7 +34,8 @@ api_schemas(Method) ->
ref(emqx_ee_bridge_clickhouse, Method),
ref(emqx_ee_bridge_dynamo, Method),
ref(emqx_ee_bridge_rocketmq, Method),
- ref(emqx_ee_bridge_cassa, Method)
+ ref(emqx_ee_bridge_cassa, Method),
+ ref(emqx_ee_bridge_sqlserver, Method)
].
schema_modules() ->
@@ -53,7 +54,8 @@ schema_modules() ->
emqx_ee_bridge_clickhouse,
emqx_ee_bridge_dynamo,
emqx_ee_bridge_rocketmq,
- emqx_ee_bridge_cassa
+ emqx_ee_bridge_cassa,
+ emqx_ee_bridge_sqlserver
].
examples(Method) ->
@@ -91,7 +93,8 @@ resource_type(tdengine) -> emqx_ee_connector_tdengine;
resource_type(clickhouse) -> emqx_ee_connector_clickhouse;
resource_type(dynamo) -> emqx_ee_connector_dynamo;
resource_type(rocketmq) -> emqx_ee_connector_rocketmq;
-resource_type(cassandra) -> emqx_ee_connector_cassa.
+resource_type(cassandra) -> emqx_ee_connector_cassa;
+resource_type(sqlserver) -> emqx_ee_connector_sqlserver.
fields(bridges) ->
[
@@ -152,7 +155,7 @@ fields(bridges) ->
}
)}
] ++ kafka_structs() ++ mongodb_structs() ++ influxdb_structs() ++ redis_structs() ++
- pgsql_structs() ++ clickhouse_structs().
+ pgsql_structs() ++ clickhouse_structs() ++ sqlserver_structs().
mongodb_structs() ->
[
@@ -249,3 +252,15 @@ clickhouse_structs() ->
}
)}
].
+
+sqlserver_structs() ->
+ [
+ {sqlserver,
+ mk(
+ hoconsc:map(name, ref(emqx_ee_bridge_sqlserver, "config")),
+ #{
+ desc => <<"Microsoft SQL Server Bridge Config">>,
+ required => false
+ }
+ )}
+ ].
diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_sqlserver.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_sqlserver.erl
new file mode 100644
index 000000000..49a5ed0ce
--- /dev/null
+++ b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_sqlserver.erl
@@ -0,0 +1,128 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+-module(emqx_ee_bridge_sqlserver).
+
+-include_lib("typerefl/include/types.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+-include_lib("emqx_bridge/include/emqx_bridge.hrl").
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+-export([
+ conn_bridge_examples/1
+]).
+
+-export([
+ namespace/0,
+ roots/0,
+ fields/1,
+ desc/1
+]).
+
+-define(DEFAULT_SQL, <<
+ "insert into t_mqtt_msg(msgid, topic, qos, payload)"
+ "values (${id}, ${topic}, ${qos}, ${payload})"
+>>).
+
+-define(DEFAULT_DRIVER, <<"ms-sqlserver-18">>).
+
+conn_bridge_examples(Method) ->
+ [
+ #{
+ <<"sqlserver">> => #{
+ summary => <<"Microsoft SQL Server Bridge">>,
+ value => values(Method)
+ }
+ }
+ ].
+
+values(get) ->
+ values(post);
+values(post) ->
+ #{
+ enable => true,
+ type => sqlserver,
+ name => <<"bar">>,
+ server => <<"127.0.0.1:1433">>,
+ database => <<"test">>,
+ pool_size => 8,
+ username => <<"sa">>,
+ password => <<"******">>,
+ sql => ?DEFAULT_SQL,
+ driver => ?DEFAULT_DRIVER,
+ local_topic => <<"local/topic/#">>,
+ resource_opts => #{
+ worker_pool_size => 1,
+ health_check_interval => ?HEALTHCHECK_INTERVAL_RAW,
+ auto_restart_interval => ?AUTO_RESTART_INTERVAL_RAW,
+ batch_size => ?DEFAULT_BATCH_SIZE,
+ batch_time => ?DEFAULT_BATCH_TIME,
+ query_mode => async,
+ max_queue_bytes => ?DEFAULT_QUEUE_SIZE
+ }
+ };
+values(put) ->
+ values(post).
+
+%% -------------------------------------------------------------------------------------------------
+%% Hocon Schema Definitions
+namespace() -> "bridge_sqlserver".
+
+roots() -> [].
+
+fields("config") ->
+ [
+ {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
+ {sql,
+ mk(
+ binary(),
+ #{desc => ?DESC("sql_template"), default => ?DEFAULT_SQL, format => <<"sql">>}
+ )},
+ {driver, mk(binary(), #{desc => ?DESC("driver"), default => ?DEFAULT_DRIVER})},
+ {local_topic,
+ mk(
+ binary(),
+ #{desc => ?DESC("local_topic"), default => undefined}
+ )},
+ {resource_opts,
+ mk(
+ ref(?MODULE, "creation_opts"),
+ #{
+ required => false,
+ default => #{},
+ desc => ?DESC(emqx_resource_schema, <<"resource_opts">>)
+ }
+ )}
+ ] ++
+ (emqx_ee_connector_sqlserver:fields(config) --
+ emqx_connector_schema_lib:prepare_statement_fields());
+fields("creation_opts") ->
+ emqx_resource_schema:fields("creation_opts");
+fields("post") ->
+ fields("post", sqlserver);
+fields("put") ->
+ fields("config");
+fields("get") ->
+ emqx_bridge_schema:status_fields() ++ fields("post").
+
+fields("post", Type) ->
+ [type_field(Type), name_field() | fields("config")].
+
+desc("config") ->
+ ?DESC("desc_config");
+desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
+ ["Configuration for Microsoft SQL Server using `", string:to_upper(Method), "` method."];
+desc("creation_opts" = Name) ->
+ emqx_resource_schema:desc(Name);
+desc(_) ->
+ undefined.
+
+%% -------------------------------------------------------------------------------------------------
+
+type_field(Type) ->
+ {type, mk(enum([Type]), #{required => true, desc => ?DESC("desc_type")})}.
+
+name_field() ->
+ {name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}.
diff --git a/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl
new file mode 100644
index 000000000..c1b33c3c9
--- /dev/null
+++ b/lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl
@@ -0,0 +1,682 @@
+%%--------------------------------------------------------------------
+% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_ee_bridge_sqlserver_SUITE).
+
+-compile(nowarn_export_all).
+-compile(export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+
+% SQL definitions
+-define(SQL_BRIDGE,
+ "insert into t_mqtt_msg(msgid, topic, qos, payload) values ( ${id}, ${topic}, ${qos}, ${payload})"
+).
+-define(SQL_SERVER_DRIVER, "ms-sql").
+
+-define(SQL_CREATE_DATABASE_IF_NOT_EXISTS,
+ " IF NOT EXISTS(SELECT name FROM sys.databases WHERE name = 'mqtt')"
+ " BEGIN"
+ " CREATE DATABASE mqtt;"
+ " END"
+).
+
+-define(SQL_CREATE_TABLE_IN_DB_MQTT,
+ " CREATE TABLE mqtt.dbo.t_mqtt_msg"
+ " (id int PRIMARY KEY IDENTITY(1000000001,1) NOT NULL,"
+ " msgid VARCHAR(64) NULL,"
+ " topic VARCHAR(100) NULL,"
+ " qos tinyint NOT NULL DEFAULT 0,"
+ %% use VARCHAR to use utf8 encoding
+ %% for default, sqlserver use utf16 encoding NVARCHAR()
+ " payload VARCHAR(100) NULL,"
+ " arrived DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP)"
+).
+
+-define(SQL_DROP_DB_MQTT, "DROP DATABASE mqtt").
+-define(SQL_DROP_TABLE, "DROP TABLE mqtt.dbo.t_mqtt_msg").
+-define(SQL_DELETE, "DELETE from mqtt.dbo.t_mqtt_msg").
+-define(SQL_SELECT, "SELECT payload FROM mqtt.dbo.t_mqtt_msg").
+-define(SQL_SELECT_COUNT, "SELECT COUNT(*) FROM mqtt.dbo.t_mqtt_msg").
+% DB defaults
+-define(SQL_SERVER_DATABASE, "mqtt").
+-define(SQL_SERVER_USERNAME, "sa").
+-define(SQL_SERVER_PASSWORD, "mqtt_public1").
+-define(BATCH_SIZE, 10).
+-define(REQUEST_TIMEOUT_MS, 500).
+
+-define(WORKER_POOL_SIZE, 4).
+
+-define(WITH_CON(Process),
+ Con = connect_direct_sqlserver(Config),
+ Process,
+ ok = disconnect(Con)
+).
+
+%% How to run it locally (all commands are run in $PROJ_ROOT dir):
+%% A: run ct on host
+%% 1. Start all deps services
+%% sudo docker compose -f .ci/docker-compose-file/docker-compose.yaml \
+%% -f .ci/docker-compose-file/docker-compose-sqlserver.yaml \
+%% -f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \
+%% up --build
+%%
+%% 2. Run use cases with special environment variables
+%% 11433 is toxiproxy exported port.
+%% Local:
+%% ```
+%% SQLSERVER_HOST=toxiproxy SQLSERVER_PORT=11433 \
+%% PROXY_HOST=toxiproxy PROXY_PORT=1433 \
+%% ./rebar3 as test ct -c -v --readable true --name ct@127.0.0.1 --suite lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl
+%% ```
+%%
+%% B: run ct in docker container
+%% run script:
+%% ./scripts/ct/run.sh --ci --app lib-ee/emqx_ee_bridge/ \
+%% -- --name 'test@127.0.0.1' -c -v --readable true --suite lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_sqlserver_SUITE.erl
+
+%%------------------------------------------------------------------------------
+%% CT boilerplate
+%%------------------------------------------------------------------------------
+
+all() ->
+ [
+ {group, async},
+ {group, sync}
+ ].
+
+groups() ->
+ TCs = emqx_common_test_helpers:all(?MODULE),
+ NonBatchCases = [t_write_timeout],
+ BatchingGroups = [{group, with_batch}, {group, without_batch}],
+ [
+ {async, BatchingGroups},
+ {sync, BatchingGroups},
+ {with_batch, TCs -- NonBatchCases},
+ {without_batch, TCs}
+ ].
+
+init_per_group(async, Config) ->
+ [{query_mode, async} | Config];
+init_per_group(sync, Config) ->
+ [{query_mode, sync} | Config];
+init_per_group(with_batch, Config0) ->
+ Config = [{enable_batch, true} | Config0],
+ common_init(Config);
+init_per_group(without_batch, Config0) ->
+ Config = [{enable_batch, false} | Config0],
+ common_init(Config);
+init_per_group(_Group, Config) ->
+ Config.
+
+end_per_group(Group, Config) when Group =:= with_batch; Group =:= without_batch ->
+ connect_and_drop_table(Config),
+ connect_and_drop_db(Config),
+ ProxyHost = ?config(proxy_host, Config),
+ ProxyPort = ?config(proxy_port, Config),
+ emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
+ ok;
+end_per_group(_Group, _Config) ->
+ ok.
+
+init_per_suite(Config) ->
+ Config.
+
+end_per_suite(_Config) ->
+ emqx_mgmt_api_test_util:end_suite(),
+ ok = emqx_common_test_helpers:stop_apps([emqx_bridge, emqx_conf]),
+ ok.
+
+init_per_testcase(_Testcase, Config) ->
+ %% drop database and table
+ %% connect_and_clear_table(Config),
+ %% create a new one
+ %% TODO: create a new database for each test case
+ delete_bridge(Config),
+ snabbkaffe:start_trace(),
+ Config.
+
+end_per_testcase(_Testcase, Config) ->
+ ProxyHost = ?config(proxy_host, Config),
+ ProxyPort = ?config(proxy_port, Config),
+ emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
+ connect_and_clear_table(Config),
+ ok = snabbkaffe:stop(),
+ delete_bridge(Config),
+ ok.
+
+%%------------------------------------------------------------------------------
+%% Testcases
+%%------------------------------------------------------------------------------
+
+t_setup_via_config_and_publish(Config) ->
+ ?assertMatch(
+ {ok, _},
+ create_bridge(Config)
+ ),
+ Val = str(erlang:unique_integer()),
+ SentData = sent_data(Val),
+ ?check_trace(
+ begin
+ ?wait_async_action(
+ ?assertEqual(ok, send_message(Config, SentData)),
+ #{?snk_kind := sqlserver_connector_query_return},
+ 10_000
+ ),
+ ?assertMatch(
+ [{Val}],
+ connect_and_get_payload(Config)
+ ),
+ ok
+ end,
+ fun(Trace0) ->
+ Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
+ ?assertMatch([#{result := ok}], Trace),
+ ok
+ end
+ ),
+ ok.
+
+t_setup_via_http_api_and_publish(Config) ->
+ BridgeType = ?config(sqlserver_bridge_type, Config),
+ Name = ?config(sqlserver_name, Config),
+ SQLServerConfig0 = ?config(sqlserver_config, Config),
+ SQLServerConfig = SQLServerConfig0#{
+ <<"name">> => Name,
+ <<"type">> => BridgeType
+ },
+ ?assertMatch(
+ {ok, _},
+ create_bridge_http(SQLServerConfig)
+ ),
+ Val = str(erlang:unique_integer()),
+ SentData = sent_data(Val),
+ ?check_trace(
+ begin
+ ?wait_async_action(
+ ?assertEqual(ok, send_message(Config, SentData)),
+ #{?snk_kind := sqlserver_connector_query_return},
+ 10_000
+ ),
+ ?assertMatch(
+ [{Val}],
+ connect_and_get_payload(Config)
+ ),
+ ok
+ end,
+ fun(Trace0) ->
+ Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
+ ?assertMatch([#{result := ok}], Trace),
+ ok
+ end
+ ),
+ ok.
+
+t_get_status(Config) ->
+ ?assertMatch(
+ {ok, _},
+ create_bridge(Config)
+ ),
+ ProxyPort = ?config(proxy_port, Config),
+ ProxyHost = ?config(proxy_host, Config),
+ ProxyName = ?config(proxy_name, Config),
+
+ health_check_resource_ok(Config),
+ emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
+ health_check_resource_down(Config)
+ end),
+ ok.
+
+t_create_disconnected(Config) ->
+ ProxyPort = ?config(proxy_port, Config),
+ ProxyHost = ?config(proxy_host, Config),
+ ProxyName = ?config(proxy_name, Config),
+ emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
+ ?assertMatch({ok, _}, create_bridge(Config)),
+ health_check_resource_down(Config)
+ end),
+ health_check_resource_ok(Config),
+ ok.
+
+t_create_with_invalid_password(Config) ->
+ BridgeType = ?config(sqlserver_bridge_type, Config),
+ Name = ?config(sqlserver_name, Config),
+ SQLServerConfig0 = ?config(sqlserver_config, Config),
+ SQLServerConfig = SQLServerConfig0#{
+ <<"name">> => Name,
+ <<"type">> => BridgeType,
+ <<"password">> => <<"wrong_password">>
+ },
+ ?check_trace(
+ begin
+ ?assertMatch(
+ {ok, _},
+ create_bridge_http(SQLServerConfig)
+ )
+ end,
+ fun(Trace) ->
+ ?assertMatch(
+ [#{error := {start_pool_failed, _, _}}],
+ ?of_kind(sqlserver_connector_start_failed, Trace)
+ ),
+ ok
+ end
+ ),
+ ok.
+
+t_write_failure(Config) ->
+ ProxyName = ?config(proxy_name, Config),
+ ProxyPort = ?config(proxy_port, Config),
+ ProxyHost = ?config(proxy_host, Config),
+ QueryMode = ?config(query_mode, Config),
+ Val = str(erlang:unique_integer()),
+ SentData = sent_data(Val),
+ {{ok, _}, {ok, _}} =
+ ?wait_async_action(
+ create_bridge(Config),
+ #{?snk_kind := resource_connected_enter},
+ 20_000
+ ),
+ emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
+ case QueryMode of
+ sync ->
+ ?assertMatch(
+ {error, {resource_error, #{reason := timeout}}},
+ send_message(Config, SentData)
+ );
+ async ->
+ ?assertMatch(
+ ok, send_message(Config, SentData)
+ )
+ end
+ end),
+ ok.
+
+t_write_timeout(_Config) ->
+ %% msodbc driver handled all connection exceptions
+ %% the case is same as t_write_failure/1
+ ok.
+
+t_simple_query(Config) ->
+ BatchSize = batch_size(Config),
+ ?assertMatch(
+ {ok, _},
+ create_bridge(Config)
+ ),
+ {Requests, Vals} = gen_batch_req(BatchSize),
+ ?check_trace(
+ begin
+ ?wait_async_action(
+ begin
+ [?assertEqual(ok, query_resource(Config, Request)) || Request <- Requests]
+ end,
+ #{?snk_kind := sqlserver_connector_query_return},
+ 10_000
+ ),
+ %% just assert the data count is correct
+ ?assertMatch(
+ BatchSize,
+ connect_and_get_count(Config)
+ ),
+ %% assert the data order is correct
+ ?assertMatch(
+ Vals,
+ connect_and_get_payload(Config)
+ )
+ end,
+ fun(Trace0) ->
+ Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
+ case BatchSize of
+ 1 ->
+ ?assertMatch([#{result := ok}], Trace);
+ _ ->
+ [?assertMatch(#{result := ok}, Trace1) || Trace1 <- Trace]
+ end,
+ ok
+ end
+ ),
+ ok.
+
+-define(MISSING_TINYINT_ERROR,
+ "[Microsoft][ODBC Driver 17 for SQL Server][SQL Server]"
+ "Conversion failed when converting the varchar value 'undefined' to data type tinyint. SQLSTATE IS: 22018"
+).
+
+t_missing_data(Config) ->
+ QueryMode = ?config(query_mode, Config),
+ ?assertMatch(
+ {ok, _},
+ create_bridge(Config)
+ ),
+ Result = send_message(Config, #{}),
+ case QueryMode of
+ sync ->
+ ?assertMatch(
+ {error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}},
+ Result
+ );
+ async ->
+ ?assertMatch(
+ ok, send_message(Config, #{})
+ )
+ end,
+ ok.
+
+t_bad_parameter(Config) ->
+ QueryMode = ?config(query_mode, Config),
+ ?assertMatch(
+ {ok, _},
+ create_bridge(Config)
+ ),
+ Result = send_message(Config, #{}),
+ case QueryMode of
+ sync ->
+ ?assertMatch(
+ {error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}},
+ Result
+ );
+ async ->
+ ?assertMatch(
+ ok, send_message(Config, #{})
+ )
+ end,
+ ok.
+
+%%------------------------------------------------------------------------------
+%% Helper fns
+%%------------------------------------------------------------------------------
+
+common_init(ConfigT) ->
+ Host = os:getenv("SQLSERVER_HOST", "toxiproxy"),
+ Port = list_to_integer(os:getenv("SQLSERVER_PORT", "1433")),
+
+ Config0 = [
+ {sqlserver_host, Host},
+ {sqlserver_port, Port},
+ %% see also for `proxy_name` : $PROJ_ROOT/.ci/docker-compose-file/toxiproxy.json
+ {proxy_name, "sqlserver"},
+ {batch_size, batch_size(ConfigT)}
+ | ConfigT
+ ],
+
+ BridgeType = proplists:get_value(bridge_type, Config0, <<"sqlserver">>),
+ case emqx_common_test_helpers:is_tcp_server_available(Host, Port) of
+ true ->
+ % Setup toxiproxy
+ ProxyHost = os:getenv("PROXY_HOST", "toxiproxy"),
+ ProxyPort = list_to_integer(os:getenv("PROXY_PORT", "8474")),
+ emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
+ % Ensure EE bridge module is loaded
+ _ = application:load(emqx_ee_bridge),
+ _ = emqx_ee_bridge:module_info(),
+ ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_bridge]),
+ emqx_mgmt_api_test_util:init_suite(),
+ % Connect to sqlserver directly
+ % drop old db and table, and then create new ones
+ connect_and_create_db_and_table(Config0),
+ {Name, SQLServerConf} = sqlserver_config(BridgeType, Config0),
+ Config =
+ [
+ {sqlserver_config, SQLServerConf},
+ {sqlserver_bridge_type, BridgeType},
+ {sqlserver_name, Name},
+ {proxy_host, ProxyHost},
+ {proxy_port, ProxyPort}
+ | Config0
+ ],
+ Config;
+ false ->
+ case os:getenv("IS_CI") of
+ "yes" ->
+ throw(no_sqlserver);
+ _ ->
+ {skip, no_sqlserver}
+ end
+ end.
+
+sqlserver_config(BridgeType, Config) ->
+ Port = integer_to_list(?config(sqlserver_port, Config)),
+ Server = ?config(sqlserver_host, Config) ++ ":" ++ Port,
+ Name = atom_to_binary(?MODULE),
+ BatchSize = batch_size(Config),
+ QueryMode = ?config(query_mode, Config),
+ ConfigString =
+ io_lib:format(
+ "bridges.~s.~s {\n"
+ " enable = true\n"
+ " server = ~p\n"
+ " database = ~p\n"
+ " username = ~p\n"
+ " password = ~p\n"
+ " sql = ~p\n"
+ " driver = ~p\n"
+ " resource_opts = {\n"
+ " request_timeout = 500ms\n"
+ " batch_size = ~b\n"
+ " query_mode = ~s\n"
+ " worker_pool_size = ~b\n"
+ " }\n"
+ "}",
+ [
+ BridgeType,
+ Name,
+ Server,
+ ?SQL_SERVER_DATABASE,
+ ?SQL_SERVER_USERNAME,
+ ?SQL_SERVER_PASSWORD,
+ ?SQL_BRIDGE,
+ ?SQL_SERVER_DRIVER,
+ BatchSize,
+ QueryMode,
+ ?WORKER_POOL_SIZE
+ ]
+ ),
+ {Name, parse_and_check(ConfigString, BridgeType, Name)}.
+
+parse_and_check(ConfigString, BridgeType, Name) ->
+ {ok, RawConf} = hocon:binary(ConfigString, #{format => map}),
+ hocon_tconf:check_plain(emqx_bridge_schema, RawConf, #{required => false, atom_key => false}),
+ #{<<"bridges">> := #{BridgeType := #{Name := Config}}} = RawConf,
+ Config.
+
+create_bridge(Config) ->
+ create_bridge(Config, _Overrides = #{}).
+
+create_bridge(Config, Overrides) ->
+ BridgeType = ?config(sqlserver_bridge_type, Config),
+ Name = ?config(sqlserver_name, Config),
+ SSConfig0 = ?config(sqlserver_config, Config),
+ SSConfig = emqx_map_lib:deep_merge(SSConfig0, Overrides),
+ emqx_bridge:create(BridgeType, Name, SSConfig).
+
+delete_bridge(Config) ->
+ BridgeType = ?config(sqlserver_bridge_type, Config),
+ Name = ?config(sqlserver_name, Config),
+ emqx_bridge:remove(BridgeType, Name).
+
+create_bridge_http(Params) ->
+ Path = emqx_mgmt_api_test_util:api_path(["bridges"]),
+ AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
+ case emqx_mgmt_api_test_util:request_api(post, Path, "", AuthHeader, Params) of
+ {ok, Res} -> {ok, emqx_json:decode(Res, [return_maps])};
+ Error -> Error
+ end.
+
+send_message(Config, Payload) ->
+ Name = ?config(sqlserver_name, Config),
+ BridgeType = ?config(sqlserver_bridge_type, Config),
+ BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name),
+ emqx_bridge:send_message(BridgeID, Payload).
+
+query_resource(Config, Request) ->
+ Name = ?config(sqlserver_name, Config),
+ BridgeType = ?config(sqlserver_bridge_type, Config),
+ ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
+ emqx_resource:query(ResourceID, Request, #{timeout => 1_000}).
+
+query_resource_async(Config, Request) ->
+ Name = ?config(sqlserver_name, Config),
+ BridgeType = ?config(sqlserver_bridge_type, Config),
+ Ref = alias([reply]),
+ AsyncReplyFun = fun(Result) -> Ref ! {result, Ref, Result} end,
+ ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
+ Return = emqx_resource:query(ResourceID, Request, #{
+ timeout => 500, async_reply_fun => {AsyncReplyFun, []}
+ }),
+ {Return, Ref}.
+
+resource_id(Config) ->
+ Name = ?config(sqlserver_name, Config),
+ BridgeType = ?config(sqlserver_bridge_type, Config),
+ _ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name).
+
+health_check_resource_ok(Config) ->
+ ?assertEqual({ok, connected}, emqx_resource_manager:health_check(resource_id(Config))).
+
+health_check_resource_down(Config) ->
+ case emqx_resource_manager:health_check(resource_id(Config)) of
+ {ok, Status} when Status =:= disconnected orelse Status =:= connecting ->
+ ok;
+ {error, timeout} ->
+ ok;
+ Other ->
+ ?assert(
+ false, lists:flatten(io_lib:format("invalid health check result:~p~n", [Other]))
+ )
+ end.
+
+receive_result(Ref, Timeout) ->
+ receive
+ {result, Ref, Result} ->
+ {ok, Result};
+ {Ref, Result} ->
+ {ok, Result}
+ after Timeout ->
+ timeout
+ end.
+
+connect_direct_sqlserver(Config) ->
+ Opts = [
+ {host, ?config(sqlserver_host, Config)},
+ {port, ?config(sqlserver_port, Config)},
+ {username, ?SQL_SERVER_USERNAME},
+ {password, ?SQL_SERVER_PASSWORD},
+ {driver, ?SQL_SERVER_DRIVER},
+ {pool_size, 8}
+ ],
+ {ok, Con} = connect(Opts),
+ Con.
+
+connect(Options) ->
+ ConnectStr = lists:concat(conn_str(Options, [])),
+ Opts = proplists:get_value(options, Options, []),
+ odbc:connect(ConnectStr, Opts).
+
+disconnect(Ref) ->
+ odbc:disconnect(Ref).
+
+% These funs connect and then stop the sqlserver connection
+connect_and_create_db_and_table(Config) ->
+ ?WITH_CON(begin
+ {updated, undefined} = directly_query(Con, ?SQL_CREATE_DATABASE_IF_NOT_EXISTS),
+ {updated, undefined} = directly_query(Con, ?SQL_CREATE_TABLE_IN_DB_MQTT)
+ end).
+
+connect_and_drop_db(Config) ->
+ ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_DB_MQTT)).
+
+connect_and_drop_table(Config) ->
+ ?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_TABLE)).
+
+connect_and_clear_table(Config) ->
+ ?WITH_CON({updated, _} = directly_query(Con, ?SQL_DELETE)).
+
+connect_and_get_payload(Config) ->
+ ?WITH_CON(
+ {selected, ["payload"], Rows} = directly_query(Con, ?SQL_SELECT)
+ ),
+ Rows.
+
+connect_and_get_count(Config) ->
+ ?WITH_CON(
+ {selected, [[]], [{Count}]} = directly_query(Con, ?SQL_SELECT_COUNT)
+ ),
+ Count.
+
+directly_query(Con, Query) ->
+ directly_query(Con, Query, ?REQUEST_TIMEOUT_MS).
+
+directly_query(Con, Query, Timeout) ->
+ odbc:sql_query(Con, Query, Timeout).
+
+%%--------------------------------------------------------------------
+%% help functions
+%%--------------------------------------------------------------------
+
+batch_size(Config) ->
+ case ?config(enable_batch, Config) of
+ true -> ?BATCH_SIZE;
+ false -> 1
+ end.
+
+conn_str([], Acc) ->
+ %% TODO: for msodbc 18+, we need to add "Encrypt=YES;TrustServerCertificate=YES"
+ %% but havn't tested now
+ %% we should use this for msodbcsql 18+
+ %% lists:join(";", ["Encrypt=YES", "TrustServerCertificate=YES" | Acc]);
+ lists:join(";", Acc);
+conn_str([{driver, Driver} | Opts], Acc) ->
+ conn_str(Opts, ["Driver=" ++ str(Driver) | Acc]);
+conn_str([{host, Host} | Opts], Acc) ->
+ Port = proplists:get_value(port, Opts, "1433"),
+ NOpts = proplists:delete(port, Opts),
+ conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
+conn_str([{port, Port} | Opts], Acc) ->
+ Host = proplists:get_value(host, Opts, "localhost"),
+ NOpts = proplists:delete(host, Opts),
+ conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
+conn_str([{database, Database} | Opts], Acc) ->
+ conn_str(Opts, ["Database=" ++ str(Database) | Acc]);
+conn_str([{username, Username} | Opts], Acc) ->
+ conn_str(Opts, ["UID=" ++ str(Username) | Acc]);
+conn_str([{password, Password} | Opts], Acc) ->
+ conn_str(Opts, ["PWD=" ++ str(Password) | Acc]);
+conn_str([{_, _} | Opts], Acc) ->
+ conn_str(Opts, Acc).
+
+sent_data(Payload) ->
+ #{
+ payload => to_bin(Payload),
+ id => <<"0005F8F84FFFAFB9F44200000D810002">>,
+ topic => <<"test/topic">>,
+ qos => 0
+ }.
+
+gen_batch_req(Count) when
+ is_integer(Count) andalso Count > 0
+->
+ Vals = [{str(erlang:unique_integer())} || _Seq <- lists:seq(1, Count)],
+ Requests = [{send_message, sent_data(Payload)} || {Payload} <- Vals],
+ {Requests, Vals};
+gen_batch_req(Count) ->
+ ct:pal("Gen batch requests failed with unexpected Count: ~p", [Count]).
+
+str(List) when is_list(List) ->
+ unicode:characters_to_list(List, utf8);
+str(Bin) when is_binary(Bin) ->
+ unicode:characters_to_list(Bin, utf8);
+str(Num) when is_number(Num) ->
+ number_to_list(Num).
+
+number_to_list(Int) when is_integer(Int) ->
+ integer_to_list(Int);
+number_to_list(Float) when is_float(Float) ->
+ float_to_list(Float, [{decimals, 10}, compact]).
+
+to_bin(List) when is_list(List) ->
+ unicode:characters_to_binary(List, utf8);
+to_bin(Bin) when is_binary(Bin) ->
+ Bin.
diff --git a/lib-ee/emqx_ee_connector/docker-ct b/lib-ee/emqx_ee_connector/docker-ct
index 446c8ee4e..fc8e75e68 100644
--- a/lib-ee/emqx_ee_connector/docker-ct
+++ b/lib-ee/emqx_ee_connector/docker-ct
@@ -2,3 +2,4 @@ toxiproxy
influxdb
clickhouse
cassandra
+sqlserver
diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src b/lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src
index 2e6406d70..2e2ff2a62 100644
--- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src
+++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector.app.src
@@ -5,13 +5,15 @@
{applications, [
kernel,
stdlib,
+ ecpool,
hstreamdb_erl,
influxdb,
tdengine,
clickhouse,
erlcloud,
rocketmq,
- ecql
+ ecql,
+ odbc
]},
{env, []},
{modules, []},
diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_cassa.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_cassa.erl
index 1e1882a1f..ce964dd17 100644
--- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_cassa.erl
+++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_cassa.erl
@@ -1,18 +1,7 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2023 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_ee_connector_cassa).
-behaviour(emqx_resource).
diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_clickhouse.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_clickhouse.erl
index 5d68687dc..3749ecc6e 100644
--- a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_clickhouse.erl
+++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_clickhouse.erl
@@ -1,17 +1,5 @@
%%--------------------------------------------------------------------
-%% Copyright (c) 2020-2023 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.
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_ee_connector_clickhouse).
diff --git a/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl
new file mode 100644
index 000000000..dc9759341
--- /dev/null
+++ b/lib-ee/emqx_ee_connector/src/emqx_ee_connector_sqlserver.erl
@@ -0,0 +1,528 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_ee_connector_sqlserver).
+
+-behaviour(emqx_resource).
+
+-include_lib("kernel/include/file.hrl").
+-include_lib("emqx/include/logger.hrl").
+-include_lib("emqx_resource/include/emqx_resource.hrl").
+-include_lib("emqx_ee_connector/include/emqx_ee_connector.hrl").
+
+-include_lib("typerefl/include/types.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+
+%%====================================================================
+%% Exports
+%%====================================================================
+
+%% Hocon config schema exports
+-export([
+ roots/0,
+ fields/1
+]).
+
+%% callbacks for behaviour emqx_resource
+-export([
+ callback_mode/0,
+ is_buffer_supported/0,
+ on_start/2,
+ on_stop/2,
+ on_query/3,
+ on_batch_query/3,
+ on_query_async/4,
+ on_batch_query_async/4,
+ on_get_status/2
+]).
+
+%% callbacks for ecpool
+-export([connect/1]).
+
+%% Internal exports used to execute code with ecpool worker
+-export([do_get_status/2, worker_do_insert/3, do_async_reply/2]).
+
+-import(emqx_plugin_libs_rule, [str/1]).
+-import(hoconsc, [mk/2, enum/1, ref/2]).
+
+-define(ACTION_SEND_MESSAGE, send_message).
+
+-define(SYNC_QUERY_MODE, handover).
+-define(ASYNC_QUERY_MODE(REPLY), {handover_async, {?MODULE, do_async_reply, [REPLY]}}).
+
+-define(SQLSERVER_HOST_OPTIONS, #{
+ default_port => 1433
+}).
+
+-define(REQUEST_TIMEOUT(RESOURCE_OPTS),
+ maps:get(request_timeout, RESOURCE_OPTS, ?DEFAULT_REQUEST_TIMEOUT)
+).
+
+-define(BATCH_INSERT_TEMP, batch_insert_temp).
+
+-define(BATCH_INSERT_PART, batch_insert_part).
+-define(BATCH_PARAMS_TOKENS, batch_insert_tks).
+
+-define(FILE_MODE_755, 33261).
+%% 32768 + 8#00400 + 8#00200 + 8#00100 + 8#00040 + 8#00010 + 8#00004 + 8#00001
+%% See also
+%% https://www.erlang.org/doc/man/file.html#read_file_info-2
+
+%% Copied from odbc reference page
+%% https://www.erlang.org/doc/man/odbc.html
+
+%% as returned by connect/2
+-type connection_reference() :: pid().
+-type time_out() :: milliseconds() | infinity.
+-type sql() :: string() | binary().
+-type milliseconds() :: pos_integer().
+%% Tuple of column values e.g. one row of the result set.
+%% it's a variable size tuple of column values.
+-type row() :: tuple().
+%% Some kind of explanation of what went wrong
+-type common_reason() :: connection_closed | extended_error() | term().
+%% extended error type with ODBC
+%% and native database error codes, as well as the base reason that would have been
+%% returned had extended_errors not been enabled.
+-type extended_error() :: {string(), integer(), _Reason :: term()}.
+%% Name of column in the result set
+-type col_name() :: string().
+%% e.g. a list of the names of the selected columns in the result set.
+-type col_names() :: [col_name()].
+%% A list of rows from the result set.
+-type rows() :: list(row()).
+
+%% -type result_tuple() :: {updated, n_rows()} | {selected, col_names(), rows()}.
+-type updated_tuple() :: {updated, n_rows()}.
+-type selected_tuple() :: {selected, col_names(), rows()}.
+%% The number of affected rows for UPDATE,
+%% INSERT, or DELETE queries. For other query types the value
+%% is driver defined, and hence should be ignored.
+-type n_rows() :: integer().
+
+%% These type was not used in this module, but we may use it later
+%% -type odbc_data_type() ::
+%% sql_integer
+%% | sql_smallint
+%% | sql_tinyint
+%% | {sql_decimal, precision(), scale()}
+%% | {sql_numeric, precision(), scale()}
+%% | {sql_char, size()}
+%% | {sql_wchar, size()}
+%% | {sql_varchar, size()}
+%% | {sql_wvarchar, size()}
+%% | {sql_float, precision()}
+%% | {sql_wlongvarchar, size()}
+%% | {sql_float, precision()}
+%% | sql_real
+%% | sql_double
+%% | sql_bit
+%% | atom().
+%% -type precision() :: integer().
+%% -type scale() :: integer().
+%% -type size() :: integer().
+
+-type state() :: #{
+ poolname := binary(),
+ resource_opts := map(),
+ sql_templates := map()
+}.
+
+%%====================================================================
+%% Configuration and default values
+%%====================================================================
+
+roots() ->
+ [{config, #{type => hoconsc:ref(?MODULE, config)}}].
+
+fields(config) ->
+ [
+ {server, server()}
+ | add_default_username(emqx_connector_schema_lib:relational_db_fields())
+ ].
+
+add_default_username(Fields) ->
+ lists:map(
+ fun
+ ({username, OrigUsernameFn}) ->
+ {username, add_default_fn(OrigUsernameFn, <<"sa">>)};
+ (Field) ->
+ Field
+ end,
+ Fields
+ ).
+
+add_default_fn(OrigFn, Default) ->
+ fun
+ (default) -> Default;
+ (Field) -> OrigFn(Field)
+ end.
+
+server() ->
+ Meta = #{desc => ?DESC("server")},
+ emqx_schema:servers_sc(Meta, ?SQLSERVER_HOST_OPTIONS).
+
+%%====================================================================
+%% Callbacks defined in emqx_resource
+%%====================================================================
+
+callback_mode() -> async_if_possible.
+
+is_buffer_supported() -> false.
+
+on_start(
+ InstanceId = PoolName,
+ #{
+ server := Server,
+ username := Username,
+ password := Password,
+ driver := Driver,
+ database := Database,
+ pool_size := PoolSize,
+ resource_opts := ResourceOpts
+ } = Config
+) ->
+ ?SLOG(info, #{
+ msg => "starting_sqlserver_connector",
+ connector => InstanceId,
+ config => emqx_misc:redact(Config)
+ }),
+
+ ODBCDir = code:priv_dir(odbc),
+ OdbcserverDir = filename:join(ODBCDir, "bin/odbcserver"),
+ {ok, Info = #file_info{mode = Mode}} = file:read_file_info(OdbcserverDir),
+ case ?FILE_MODE_755 =:= Mode of
+ true ->
+ ok;
+ false ->
+ _ = file:write_file_info(OdbcserverDir, Info#file_info{mode = ?FILE_MODE_755}),
+ ok
+ end,
+
+ Options = [
+ {server, to_bin(Server)},
+ {username, Username},
+ {password, Password},
+ {driver, Driver},
+ {database, Database},
+ {pool_size, PoolSize},
+ {poolname, PoolName}
+ ],
+
+ State = #{
+ %% also InstanceId
+ poolname => PoolName,
+ sql_templates => parse_sql_template(Config),
+ resource_opts => ResourceOpts
+ },
+ case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Options) of
+ ok ->
+ {ok, State};
+ {error, Reason} ->
+ ?tp(
+ sqlserver_connector_start_failed,
+ #{error => Reason}
+ ),
+ {error, Reason}
+ end.
+
+on_stop(InstanceId, #{poolname := PoolName} = _State) ->
+ ?SLOG(info, #{
+ msg => "stopping_sqlserver_connector",
+ connector => InstanceId
+ }),
+ emqx_plugin_libs_pool:stop_pool(PoolName).
+
+-spec on_query(
+ manager_id(),
+ {?ACTION_SEND_MESSAGE, map()},
+ state()
+) ->
+ ok
+ | {ok, list()}
+ | {error, {recoverable_error, term()}}
+ | {error, term()}.
+on_query(InstanceId, {?ACTION_SEND_MESSAGE, _Msg} = Query, State) ->
+ ?TRACE(
+ "SINGLE_QUERY_SYNC",
+ "bridge_sqlserver_received",
+ #{requests => Query, connector => InstanceId, state => State}
+ ),
+ do_query(InstanceId, Query, ?SYNC_QUERY_MODE, State).
+
+-spec on_query_async(
+ manager_id(),
+ {?ACTION_SEND_MESSAGE, map()},
+ {ReplyFun :: function(), Args :: list()},
+ state()
+) ->
+ {ok, any()}
+ | {error, term()}.
+on_query_async(
+ InstanceId,
+ {?ACTION_SEND_MESSAGE, _Msg} = Query,
+ ReplyFunAndArgs,
+ %% #{poolname := PoolName, sql_templates := Templates} = State
+ State
+) ->
+ ?TRACE(
+ "SINGLE_QUERY_ASYNC",
+ "bridge_sqlserver_received",
+ #{requests => Query, connector => InstanceId, state => State}
+ ),
+ do_query(InstanceId, Query, ?ASYNC_QUERY_MODE(ReplyFunAndArgs), State).
+
+-spec on_batch_query(
+ manager_id(),
+ [{?ACTION_SEND_MESSAGE, map()}],
+ state()
+) ->
+ ok
+ | {ok, list()}
+ | {error, {recoverable_error, term()}}
+ | {error, term()}.
+on_batch_query(InstanceId, BatchRequests, State) ->
+ ?TRACE(
+ "BATCH_QUERY_SYNC",
+ "bridge_sqlserver_received",
+ #{requests => BatchRequests, connector => InstanceId, state => State}
+ ),
+ do_query(InstanceId, BatchRequests, ?SYNC_QUERY_MODE, State).
+
+-spec on_batch_query_async(
+ manager_id(),
+ [{?ACTION_SEND_MESSAGE, map()}],
+ {ReplyFun :: function(), Args :: list()},
+ state()
+) -> {ok, any()}.
+on_batch_query_async(InstanceId, Requests, ReplyFunAndArgs, State) ->
+ ?TRACE(
+ "BATCH_QUERY_ASYNC",
+ "bridge_sqlserver_received",
+ #{requests => Requests, connector => InstanceId, state => State}
+ ),
+ do_query(InstanceId, Requests, ?ASYNC_QUERY_MODE(ReplyFunAndArgs), State).
+
+on_get_status(_InstanceId, #{poolname := Pool, resource_opts := ResourceOpts} = _State) ->
+ RequestTimeout = ?REQUEST_TIMEOUT(ResourceOpts),
+ Health = emqx_plugin_libs_pool:health_check_ecpool_workers(
+ Pool, {?MODULE, do_get_status, [RequestTimeout]}, RequestTimeout
+ ),
+ status_result(Health).
+
+status_result(_Status = true) -> connected;
+status_result(_Status = false) -> connecting.
+%% TODO:
+%% case for disconnected
+
+%%====================================================================
+%% ecpool callback fns
+%%====================================================================
+
+-spec connect(Options :: list()) -> {ok, connection_reference()} | {error, term()}.
+connect(Options) ->
+ ConnectStr = lists:concat(conn_str(Options, [])),
+ Opts = proplists:get_value(options, Options, []),
+ odbc:connect(ConnectStr, Opts).
+
+-spec do_get_status(connection_reference(), time_out()) -> Result :: boolean().
+do_get_status(Conn, RequestTimeout) ->
+ case execute(Conn, <<"SELECT 1">>, RequestTimeout) of
+ {selected, [[]], [{1}]} -> true;
+ _ -> false
+ end.
+
+%%====================================================================
+%% Internal Helper fns
+%%====================================================================
+
+%% TODO && FIXME:
+%% About the connection string attribute `Encrypt`:
+%% The default value is `yes` in odbc version 18.0+ and `no` in previous versions.
+%% And encrypted connections always verify the server's certificate.
+%% So `Encrypt=YES;TrustServerCertificate=YES` must be set in the connection string
+%% when connecting to a server that has a self-signed certificate.
+%% See also:
+%% 'https://learn.microsoft.com/en-us/sql/connect/odbc/
+%% dsn-connection-string-attribute?source=recommendations&view=sql-server-ver16#encrypt'
+conn_str([], Acc) ->
+ %% we should use this for msodbcsql 18+
+ %% lists:join(";", ["Encrypt=YES", "TrustServerCertificate=YES" | Acc]);
+ lists:join(";", Acc);
+conn_str([{driver, Driver} | Opts], Acc) ->
+ conn_str(Opts, ["Driver=" ++ str(Driver) | Acc]);
+conn_str([{server, Server} | Opts], Acc) ->
+ {Host, Port} = emqx_schema:parse_server(Server, ?SQLSERVER_HOST_OPTIONS),
+ conn_str(Opts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
+conn_str([{database, Database} | Opts], Acc) ->
+ conn_str(Opts, ["Database=" ++ str(Database) | Acc]);
+conn_str([{username, Username} | Opts], Acc) ->
+ conn_str(Opts, ["UID=" ++ str(Username) | Acc]);
+conn_str([{password, Password} | Opts], Acc) ->
+ conn_str(Opts, ["PWD=" ++ str(Password) | Acc]);
+conn_str([{_, _} | Opts], Acc) ->
+ conn_str(Opts, Acc).
+
+%% Sync & Async query with singe & batch sql statement
+-spec do_query(
+ manager_id(),
+ Query :: {?ACTION_SEND_MESSAGE, map()} | [{?ACTION_SEND_MESSAGE, map()}],
+ ApplyMode ::
+ handover
+ | {handover_async, {?MODULE, do_async_reply, [{ReplyFun :: function(), Args :: list()}]}},
+ state()
+) ->
+ {ok, list()}
+ | {error, {recoverable_error, term()}}
+ | {error, term()}.
+do_query(
+ InstanceId,
+ Query,
+ ApplyMode,
+ #{poolname := PoolName, sql_templates := Templates} = State
+) ->
+ ?TRACE(
+ "SINGLE_QUERY_SYNC",
+ "sqlserver_connector_received",
+ #{query => Query, connector => InstanceId, state => State}
+ ),
+
+ %% only insert sql statement for single query and batch query
+ case apply_template(Query, Templates) of
+ {?ACTION_SEND_MESSAGE, SQL} ->
+ Result = ecpool:pick_and_do(
+ PoolName,
+ {?MODULE, worker_do_insert, [SQL, State]},
+ ApplyMode
+ );
+ Query ->
+ Result = {error, {unrecoverable_error, invalid_query}};
+ _ ->
+ Result = {error, {unrecoverable_error, failed_to_apply_sql_template}}
+ end,
+ case Result of
+ {error, Reason} ->
+ ?tp(
+ sqlserver_connector_query_return,
+ #{error => Reason}
+ ),
+ ?SLOG(error, #{
+ msg => "sqlserver_connector_do_query_failed",
+ connector => InstanceId,
+ query => Query,
+ reason => Reason
+ }),
+ Result;
+ _ ->
+ ?tp(
+ sqlserver_connector_query_return,
+ #{result => Result}
+ ),
+ Result
+ end.
+
+worker_do_insert(
+ Conn, SQL, #{resource_opts := ResourceOpts, poolname := InstanceId} = State
+) ->
+ LogMeta = #{connector => InstanceId, state => State},
+ try
+ case execute(Conn, SQL, ?REQUEST_TIMEOUT(ResourceOpts)) of
+ {selected, Rows, _} ->
+ {ok, Rows};
+ {updated, _} ->
+ ok;
+ {error, ErrStr} ->
+ ?SLOG(error, LogMeta#{msg => "invalid_request", reason => ErrStr}),
+ {error, {unrecoverable_error, {invalid_request, ErrStr}}}
+ end
+ catch
+ _Type:Reason ->
+ ?SLOG(error, LogMeta#{msg => "invalid_request", reason => Reason}),
+ {error, {unrecoverable_error, {invalid_request, Reason}}}
+ end.
+
+-spec execute(pid(), sql(), time_out()) ->
+ updated_tuple()
+ | selected_tuple()
+ | [updated_tuple()]
+ | [selected_tuple()]
+ | {error, common_reason()}.
+execute(Conn, SQL, Timeout) ->
+ odbc:sql_query(Conn, str(SQL), Timeout).
+
+to_bin(List) when is_list(List) ->
+ unicode:characters_to_binary(List, utf8).
+
+%% for bridge data to sql server
+parse_sql_template(Config) ->
+ RawSQLTemplates =
+ case maps:get(sql, Config, undefined) of
+ undefined -> #{};
+ <<>> -> #{};
+ SQLTemplate -> #{?ACTION_SEND_MESSAGE => SQLTemplate}
+ end,
+
+ BatchInsertTks = #{},
+ parse_sql_template(maps:to_list(RawSQLTemplates), BatchInsertTks).
+
+parse_sql_template([{Key, H} | T], BatchInsertTks) ->
+ case emqx_plugin_libs_rule:detect_sql_type(H) of
+ {ok, select} ->
+ parse_sql_template(T, BatchInsertTks);
+ {ok, insert} ->
+ case emqx_plugin_libs_rule:split_insert_sql(H) of
+ {ok, {InsertSQL, Params}} ->
+ parse_sql_template(
+ T,
+ BatchInsertTks#{
+ Key =>
+ #{
+ ?BATCH_INSERT_PART => InsertSQL,
+ ?BATCH_PARAMS_TOKENS => emqx_plugin_libs_rule:preproc_tmpl(
+ Params
+ )
+ }
+ }
+ );
+ {error, Reason} ->
+ ?SLOG(error, #{msg => "split sql failed", sql => H, reason => Reason}),
+ parse_sql_template(T, BatchInsertTks)
+ end;
+ {error, Reason} ->
+ ?SLOG(error, #{msg => "detect sql type failed", sql => H, reason => Reason}),
+ parse_sql_template(T, BatchInsertTks)
+ end;
+parse_sql_template([], BatchInsertTks) ->
+ #{
+ ?BATCH_INSERT_TEMP => BatchInsertTks
+ }.
+
+%% single insert
+apply_template(
+ {?ACTION_SEND_MESSAGE = _Key, _Msg} = Query, Templates
+) ->
+ %% TODO: fix emqx_plugin_libs_rule:proc_tmpl/2
+ %% it won't add single quotes for string
+ apply_template([Query], Templates);
+%% batch inserts
+apply_template(
+ [{?ACTION_SEND_MESSAGE = Key, _Msg} | _T] = BatchReqs,
+ #{?BATCH_INSERT_TEMP := BatchInsertsTks} = _Templates
+) ->
+ case maps:get(Key, BatchInsertsTks, undefined) of
+ undefined ->
+ BatchReqs;
+ #{?BATCH_INSERT_PART := BatchInserts, ?BATCH_PARAMS_TOKENS := BatchParamsTks} ->
+ SQL = emqx_plugin_libs_rule:proc_batch_sql(BatchReqs, BatchInserts, BatchParamsTks),
+ {Key, SQL}
+ end;
+apply_template(Query, Templates) ->
+ %% TODO: more detail infomatoin
+ ?SLOG(error, #{msg => "apply sql template failed", query => Query, templates => Templates}),
+ {error, failed_to_apply_sql_template}.
+
+do_async_reply(Result, {ReplyFun, Args}) ->
+ erlang:apply(ReplyFun, Args ++ [Result]).
diff --git a/lib-ee/emqx_ee_connector/test/emqx_ee_connector_cassa_SUITE.erl b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_cassa_SUITE.erl
index 95b4407cf..52ed03a62 100644
--- a/lib-ee/emqx_ee_connector/test/emqx_ee_connector_cassa_SUITE.erl
+++ b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_cassa_SUITE.erl
@@ -1,16 +1,5 @@
%%--------------------------------------------------------------------
-%% Copyright (c) 2020-2023 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.
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_ee_connector_cassa_SUITE).
diff --git a/lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_clickhouse_SUITE.erl
similarity index 89%
rename from lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl
rename to lib-ee/emqx_ee_connector/test/emqx_ee_connector_clickhouse_SUITE.erl
index 111deba06..73018e14f 100644
--- a/lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl
+++ b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_clickhouse_SUITE.erl
@@ -1,19 +1,8 @@
-% %%--------------------------------------------------------------------
-% %% Copyright (c) 2020-2023 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.
-% %%--------------------------------------------------------------------
+%%--------------------------------------------------------------------
+%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
--module(ee_connector_clickhouse_SUITE).
+-module(emqx_ee_connector_clickhouse_SUITE).
-compile(nowarn_export_all).
-compile(export_all).
diff --git a/lib-ee/emqx_ee_connector/test/ee_connector_hstreamdb_SUITE.erl b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_hstreamdb_SUITE.erl
similarity index 90%
rename from lib-ee/emqx_ee_connector/test/ee_connector_hstreamdb_SUITE.erl
rename to lib-ee/emqx_ee_connector/test/emqx_ee_connector_hstreamdb_SUITE.erl
index 8acabbef4..ad49d9f62 100644
--- a/lib-ee/emqx_ee_connector/test/ee_connector_hstreamdb_SUITE.erl
+++ b/lib-ee/emqx_ee_connector/test/emqx_ee_connector_hstreamdb_SUITE.erl
@@ -2,7 +2,7 @@
%% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
--module(ee_connector_hstreamdb_SUITE).
+-module(emqx_ee_connector_hstreamdb_SUITE).
-compile(nowarn_export_all).
-compile(export_all).
diff --git a/rel/i18n/emqx_ee_bridge_dynamo.hocon b/rel/i18n/emqx_ee_bridge_dynamo.hocon
index 664b13174..b93b12166 100644
--- a/rel/i18n/emqx_ee_bridge_dynamo.hocon
+++ b/rel/i18n/emqx_ee_bridge_dynamo.hocon
@@ -39,7 +39,7 @@ will be forwarded."""
desc_config {
desc {
- en: """Configuration for an DynamoDB bridge."""
+ en: """Configuration for a DynamoDB bridge."""
zh: """DynamoDB 桥接配置"""
}
label: {
diff --git a/rel/i18n/emqx_ee_bridge_pgsql.hocon b/rel/i18n/emqx_ee_bridge_pgsql.hocon
index 81d2330c5..1ce2818ca 100644
--- a/rel/i18n/emqx_ee_bridge_pgsql.hocon
+++ b/rel/i18n/emqx_ee_bridge_pgsql.hocon
@@ -39,7 +39,7 @@ will be forwarded."""
desc_config {
desc {
- en: """Configuration for an PostgreSQL bridge."""
+ en: """Configuration for a PostgreSQL bridge."""
zh: """PostgreSQL 桥接配置"""
}
label: {
diff --git a/rel/i18n/emqx_ee_bridge_sqlserver.hocon b/rel/i18n/emqx_ee_bridge_sqlserver.hocon
new file mode 100644
index 000000000..a6f96f86d
--- /dev/null
+++ b/rel/i18n/emqx_ee_bridge_sqlserver.hocon
@@ -0,0 +1,85 @@
+emqx_ee_bridge_sqlserver {
+
+ local_topic {
+ desc {
+ en: """The MQTT topic filter to be forwarded to Microsoft SQL Server. All MQTT 'PUBLISH' messages with the topic
+matching the local_topic will be forwarded.
+NOTE: if this bridge is used as the action of a rule (EMQX rule engine), and also local_topic is
+configured, then both the data got from the rule and the MQTT messages that match local_topic
+will be forwarded."""
+ zh: """发送到 'local_topic' 的消息都会转发到 Microsoft SQL Server。
+注意：如果这个 Bridge 被用作规则（EMQX 规则引擎）的输出，同时也配置了 'local_topic' ，那么这两部分的消息都会被转发。"""
+ }
+ label {
+ en: """Local Topic"""
+ zh: """本地 Topic"""
+ }
+ }
+
+ sql_template {
+ desc {
+ en: """SQL Template"""
+ zh: """SQL 模板"""
+ }
+ label {
+ en: """SQL Template"""
+ zh: """SQL 模板"""
+ }
+ }
+
+ driver {
+ desc {
+ en: """SQL Server Driver Name"""
+ zh: """SQL Server Driver 名称"""
+ }
+ label {
+ en: """SQL Server Driver Name"""
+ zh: """SQL Server Driver 名称"""
+ }
+ }
+
+ config_enable {
+ desc {
+ en: """Enable or disable this bridge"""
+ zh: """启用/禁用桥接"""
+ }
+ label {
+ en: """Enable Or Disable Bridge"""
+ zh: """启用/禁用桥接"""
+ }
+ }
+
+ desc_config {
+ desc {
+ en: """Configuration for a Microsoft SQL Server bridge."""
+ zh: """Microsoft SQL Server 桥接配置"""
+ }
+ label: {
+ en: """Microsoft SQL Server Bridge Configuration"""
+ zh: """Microsoft SQL Server 桥接配置"""
+ }
+ }
+
+ desc_type {
+ desc {
+ en: """The Bridge Type"""
+ zh: """Bridge 类型"""
+ }
+ label {
+ en: """Bridge Type"""
+ zh: """桥接类型"""
+ }
+ }
+
+ desc_name {
+ desc {
+ en: """Bridge name."""
+ zh: """桥接名字"""
+ }
+ label {
+ en: """Bridge Name"""
+ zh: """桥接名字"""
+ }
+ }
+
+}
diff --git a/rel/i18n/emqx_ee_bridge_tdengine.hocon b/rel/i18n/emqx_ee_bridge_tdengine.hocon
index 21fc013df..12eb6f062 100644
--- a/rel/i18n/emqx_ee_bridge_tdengine.hocon
+++ b/rel/i18n/emqx_ee_bridge_tdengine.hocon
@@ -39,7 +39,7 @@ will be forwarded."""
desc_config {
desc {
- en: """Configuration for an TDengine bridge."""
+ en: """Configuration for a TDengine bridge."""
zh: """TDengine 桥接配置"""
}
label: {
diff --git a/rel/i18n/emqx_ee_connector_sqlserver.hocon b/rel/i18n/emqx_ee_connector_sqlserver.hocon
new file mode 100644
index 000000000..85280d833
--- /dev/null
+++ b/rel/i18n/emqx_ee_connector_sqlserver.hocon
@@ -0,0 +1,18 @@
+emqx_ee_connector_sqlserver {
+
+ server {
+ desc {
+ en: """The IPv4 or IPv6 address or the hostname to connect to.
+A host entry has the following form: `Host[:Port]`.
+The SQL Server default port 1433 is used if `[:Port]` is not specified."""
+ zh: """将要连接的 IPv4 或 IPv6 地址，或者主机名。
+主机名具有以下形式：`Host[:Port]`。
+如果未指定 `[:Port]`，则使用 SQL Server 默认端口 1433。"""
+ }
+ label: {
+ en: "Server Host"
+ zh: "服务器地址"
+ }
+ }
+
+}
diff --git a/scripts/check-i18n-style.escript b/scripts/check-i18n-style.escript
index cbe79c82e..a76fce90e 100755
--- a/scripts/check-i18n-style.escript
+++ b/scripts/check-i18n-style.escript
@@ -9,7 +9,7 @@
-define(RESET, "\e[39m").
main([Files0]) ->
- io:format(user, "checking i18n file styles", []),
+ io:format(user, "checking i18n file styles~n", []),
_ = put(errors, 0),
Files = string:tokens(Files0, "\n"),
ok = load_hocon(),
@@ -20,7 +20,7 @@ main([Files0]) ->
N when is_integer(N) andalso N > 1 ->
logerr("~p errors found~n", [N]);
_ ->
- io:format(user, "OK~n", [])
+ io:format(user, "~nOK~n", [])
end.
load_hocon() ->
@@ -42,7 +42,7 @@ die(Msg, Args) ->
halt(1).
logerr(Fmt, Args) ->
- io:format(standard_error, ?RED ++ "ERROR: " ++ Fmt ++ ?RESET, Args),
+ io:format(standard_error, "~n" ++ ?RED ++ "ERROR: " ++ Fmt ++ ?RESET, Args),
N = get(errors),
_ = put(errors, N + 1),
ok.
diff --git a/scripts/ct/run.sh b/scripts/ct/run.sh
index 83e69d1ad..4e79476e0 100755
--- a/scripts/ct/run.sh
+++ b/scripts/ct/run.sh
@@ -39,6 +39,7 @@ ONLY_UP='no'
ATTACH='no'
STOP='no'
IS_CI='no'
+ODBC_REQUEST='no'
while [ "$#" -gt 0 ]; do
case $1 in
-h|--help)
@@ -180,6 +181,10 @@ for dep in ${CT_DEPS}; do
cassandra)
FILES+=( '.ci/docker-compose-file/docker-compose-cassandra.yaml' )
;;
+ sqlserver)
+ ODBC_REQUEST='yes'
+ FILES+=( '.ci/docker-compose-file/docker-compose-sqlserver.yaml' )
+ ;;
*)
echo "unknown_ct_dependency $dep"
exit 1
@@ -187,6 +192,12 @@ for dep in ${CT_DEPS}; do
esac
done
+if [ "$ODBC_REQUEST" = 'yes' ]; then
+ INSTALL_ODBC="./scripts/install-odbc-driver.sh"
+else
+ INSTALL_ODBC="echo 'Driver msodbcsql driver not requested'"
+fi
+
F_OPTIONS=""
for file in "${FILES[@]}"; do
@@ -229,6 +240,7 @@ docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "openssl rand -base
# the user must exist inside the container for `whoami` to work
docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "useradd --uid $DOCKER_USER -M -d / emqx" || true
docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "chown -R $DOCKER_USER /var/lib/secret" || true
+docker exec -i $TTY -u root:root "$ERLANG_CONTAINER" bash -c "$INSTALL_ODBC" || true
if [ "$ONLY_UP" = 'yes' ]; then
exit 0
diff --git a/scripts/install-odbc-driver.sh b/scripts/install-odbc-driver.sh
new file mode 100755
index 000000000..7ceab95a0
--- /dev/null
+++ b/scripts/install-odbc-driver.sh
@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+
+## this script install msodbcsql17 and unixodbc-dev on ci environment
+## specific to ubuntu 16.04, 18.04, 20.04, 22.04
+set -euo pipefail
+
+# install msodbcsql17
+VERSION=$(lsb_release -rs)
+if ! [[ "16.04 18.04 20.04 22.04" == *"$VERSION"* ]];
+then
+ echo "Ubuntu $VERSION is not currently supported.";
+ exit 1;
+fi
+
+curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \
+curl https://packages.microsoft.com/config/ubuntu/"$VERSION"/prod.list > /etc/apt/sources.list.d/mssql-release.list && \
+## TODO: upgrade builder image
+apt-get update && \
+ACCEPT_EULA=Y apt-get install -y msodbcsql17 unixodbc-dev mssql-tools && \
+## and not needed to modify /etc/odbcinst.ini
+## docker-compose will mount one in .ci/docker-compose-file/odbc
+sed -i 's/ODBC Driver 17 for SQL Server/ms-sql/g' /etc/odbcinst.ini