Merge pull request #10363 from JimMoen/feat-mssqlserver-birdge
feat: implement Microsoft SQL Server bridge (e5.0)
This commit is contained in:
JimMoen
GitHub
commit
790d841697
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -16,6 +16,7 @@ services:
|
|||
- 8474:8474
|
||||
- 8086:8086
|
||||
- 8087:8087
|
||||
- 11433:1433
|
||||
- 13306:3306
|
||||
- 13307:3307
|
||||
- 15432:5432
|
||||
|
|
|
|||
|
|
@ -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}"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
}
|
||||
]
|
||||
|
|
|
|||
|
|
@ -69,7 +69,8 @@
|
|||
T == tdengine;
|
||||
T == dynamo;
|
||||
T == rocketmq;
|
||||
T == cassandra
|
||||
T == cassandra;
|
||||
T == sqlserver
|
||||
).
|
||||
|
||||
load() ->
|
||||
|
|
|
|||
|
|
@ -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 ->
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
Implement Microsoft SQL Server bridge.
|
||||
|
|
@ -11,3 +11,4 @@ clickhouse
|
|||
dynamo
|
||||
rocketmq
|
||||
cassandra
|
||||
sqlserver
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
)}
|
||||
].
|
||||
|
|
|
|||
|
|
@ -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")})}.
|
||||
|
|
@ -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.
|
||||
|
|
@ -2,3 +2,4 @@ toxiproxy
|
|||
influxdb
|
||||
clickhouse
|
||||
cassandra
|
||||
sqlserver
|
||||
|
|
|
|||
|
|
@ -5,13 +5,15 @@
|
|||
{applications, [
|
||||
kernel,
|
||||
stdlib,
|
||||
ecpool,
|
||||
hstreamdb_erl,
|
||||
influxdb,
|
||||
tdengine,
|
||||
clickhouse,
|
||||
erlcloud,
|
||||
rocketmq,
|
||||
ecql
|
||||
ecql,
|
||||
odbc
|
||||
]},
|
||||
{env, []},
|
||||
{modules, []},
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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]).
|
||||
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
@ -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).
|
||||
|
|
@ -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: {
|
||||
|
|
|
|||
|
|
@ -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: {
|
||||
|
|
|
|||
|
|
@ -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.</br>
|
||||
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。 </br>
|
||||
注意：如果这个 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: """桥接名字"""
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -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: {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,18 @@
|
|||
emqx_ee_connector_sqlserver {
|
||||
|
||||
server {
|
||||
desc {
|
||||
en: """The IPv4 or IPv6 address or the hostname to connect to.<br/>
|
||||
A host entry has the following form: `Host[:Port]`.<br/>
|
||||
The SQL Server default port 1433 is used if `[:Port]` is not specified."""
|
||||
zh: """将要连接的 IPv4 或 IPv6 地址，或者主机名。<br/>
|
||||
主机名具有以下形式：`Host[:Port]`。<br/>
|
||||
如果未指定 `[:Port]`，则使用 SQL Server 默认端口 1433。"""
|
||||
}
|
||||
label: {
|
||||
en: "Server Host"
|
||||
zh: "服务器地址"
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
Loading…
Reference in New Issue