yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(bridge): add Redis bridge

This commit is contained in:
Ilya Averyanov 2022-11-28 18:20:15 +03:00
parent 485c4ddcab
commit 6692b0c895
21 changed files with 1117 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.ci/docker-compose-file/Makefile.local
View File

@ -16,7 +16,7 @@ up:
REDIS_TAG=6 \
MONGO_TAG=5 \
PGSQL_TAG=13 \
docker-compose \
docker compose \
-f .ci/docker-compose-file/docker-compose.yaml \
-f .ci/docker-compose-file/docker-compose-mongo-single-tcp.yaml \
-f .ci/docker-compose-file/docker-compose-mongo-single-tls.yaml \
@ -28,10 +28,13 @@ up:
-f .ci/docker-compose-file/docker-compose-redis-single-tls.yaml \
-f .ci/docker-compose-file/docker-compose-redis-sentinel-tcp.yaml \
-f .ci/docker-compose-file/docker-compose-redis-sentinel-tls.yaml \
up -d --build
-f .ci/docker-compose-file/docker-compose-redis-cluster-tcp.yaml \
-f .ci/docker-compose-file/docker-compose-redis-cluster-tls.yaml \
-f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \
up -d --build --remove-orphans
down:
docker-compose \
docker compose \
-f .ci/docker-compose-file/docker-compose.yaml \
-f .ci/docker-compose-file/docker-compose-mongo-single-tcp.yaml \
-f .ci/docker-compose-file/docker-compose-mongo-single-tls.yaml \
@ -43,7 +46,10 @@ down:
-f .ci/docker-compose-file/docker-compose-redis-single-tls.yaml \
-f .ci/docker-compose-file/docker-compose-redis-sentinel-tcp.yaml \
-f .ci/docker-compose-file/docker-compose-redis-sentinel-tls.yaml \
down
-f .ci/docker-compose-file/docker-compose-redis-cluster-tcp.yaml \
-f .ci/docker-compose-file/docker-compose-redis-cluster-tls.yaml \
-f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \
down --remove-orphans
ct:
docker exec -i "$(CONTAINER)" bash -c "rebar3 ct --name 'test@127.0.0.1' -v --suite $(SUITE)"

4
.ci/docker-compose-file/docker-compose-redis-cluster-tcp.yaml
View File

@ -1,9 +1,9 @@
version: '3.9'
services:
redis_server:
redis_cluster:
image: redis:${REDIS_TAG}
container_name: redis
container_name: redis-cluster
volumes:
- ./redis/:/data/conf
command: bash -c "/bin/bash /data/conf/redis.sh --node cluster && tail -f /var/log/redis-server.log"

4
.ci/docker-compose-file/docker-compose-redis-cluster-tls.yaml
View File

@ -1,8 +1,8 @@
version: '3.9'
services:
redis_server:
container_name: redis
redis_cluster_tls:
container_name: redis-cluster-tls
image: redis:${REDIS_TAG}
volumes:
- ../../apps/emqx/etc/certs/cacert.pem:/etc/certs/ca.crt

6
.ci/docker-compose-file/redis/.gitignore vendored
View File

@ -1,3 +1,3 @@
r7000i.log
r7001i.log
r7002i.log
r700?i.log
nodes.700?.conf
*.rdb

7
.ci/docker-compose-file/redis/redis-tls.conf
View File

@ -1,11 +1,12 @@
daemonize yes
bind 0.0.0.0 ::
logfile /var/log/redis-server.log
protected-mode no
requirepass public
masterauth public
tls-cert-file /etc/certs/redis.crt
tls-key-file /etc/certs/redis.key
tls-ca-cert-file /etc/certs/ca.crt
tls-replication yes
tls-cluster yes
protected-mode no
requirepass public
masterauth public

1
.ci/docker-compose-file/redis/redis.conf
View File

@ -1,5 +1,6 @@
daemonize yes
bind 0.0.0.0 ::
logfile /var/log/redis-server.log
protected-mode no
requirepass public
masterauth public

91
.ci/docker-compose-file/redis/redis.sh
View File

@ -16,13 +16,8 @@ case $key in
shift # past argument
shift # past value
;;
-t)
tls="$2"
shift # past argument
shift # past value
;;
--tls-enabled)
tls=1
tls=true
shift # past argument
;;
*)
@ -37,69 +32,71 @@ rm -f \
/data/conf/r7002i.log \
/data/conf/nodes.7000.conf \
/data/conf/nodes.7001.conf \
/data/conf/nodes.7002.conf ;
/data/conf/nodes.7002.conf
if [ "${node}" = "cluster" ] ; then
if $tls ; then
if [ "$node" = "cluster" ]; then
if $tls; then
redis-server /data/conf/redis-tls.conf --port 7000 --cluster-config-file /data/conf/nodes.7000.conf \
--tls-port 8000 --cluster-enabled yes ;
--tls-port 8000 --cluster-enabled yes
redis-server /data/conf/redis-tls.conf --port 7001 --cluster-config-file /data/conf/nodes.7001.conf \
--tls-port 8001 --cluster-enabled yes;
--tls-port 8001 --cluster-enabled yes
redis-server /data/conf/redis-tls.conf --port 7002 --cluster-config-file /data/conf/nodes.7002.conf \
--tls-port 8002 --cluster-enabled yes;
--tls-port 8002 --cluster-enabled yes
else
redis-server /data/conf/redis.conf --port 7000 --cluster-config-file /data/conf/nodes.7000.conf --cluster-enabled yes;
redis-server /data/conf/redis.conf --port 7001 --cluster-config-file /data/conf/nodes.7001.conf --cluster-enabled yes;
redis-server /data/conf/redis.conf --port 7002 --cluster-config-file /data/conf/nodes.7002.conf --cluster-enabled yes;
redis-server /data/conf/redis.conf --port 7000 --cluster-config-file /data/conf/nodes.7000.conf \
--cluster-enabled yes
redis-server /data/conf/redis.conf --port 7001 --cluster-config-file /data/conf/nodes.7001.conf \
--cluster-enabled yes
redis-server /data/conf/redis.conf --port 7002 --cluster-config-file /data/conf/nodes.7002.conf \
--cluster-enabled yes
fi
elif [ "${node}" = "sentinel" ] ; then
if $tls ; then
elif [ "$node" = "sentinel" ]; then
if $tls; then
redis-server /data/conf/redis-tls.conf --port 7000 --cluster-config-file /data/conf/nodes.7000.conf \
--tls-port 8000 --cluster-enabled no;
--tls-port 8000 --cluster-enabled no
redis-server /data/conf/redis-tls.conf --port 7001 --cluster-config-file /data/conf/nodes.7001.conf \
--tls-port 8001 --cluster-enabled no --slaveof "$LOCAL_IP" 8000;
--tls-port 8001 --cluster-enabled no --slaveof "$LOCAL_IP" 8000
redis-server /data/conf/redis-tls.conf --port 7002 --cluster-config-file /data/conf/nodes.7002.conf \
--tls-port 8002 --cluster-enabled no --slaveof "$LOCAL_IP" 8000;
--tls-port 8002 --cluster-enabled no --slaveof "$LOCAL_IP" 8000
else
redis-server /data/conf/redis.conf --port 7000 --cluster-config-file /data/conf/nodes.7000.conf \
--cluster-enabled no;
--cluster-enabled no
redis-server /data/conf/redis.conf --port 7001 --cluster-config-file /data/conf/nodes.7001.conf \
--cluster-enabled no --slaveof "$LOCAL_IP" 7000;
--cluster-enabled no --slaveof "$LOCAL_IP" 7000
redis-server /data/conf/redis.conf --port 7002 --cluster-config-file /data/conf/nodes.7002.conf \
--cluster-enabled no --slaveof "$LOCAL_IP" 7000;
--cluster-enabled no --slaveof "$LOCAL_IP" 7000
fi
fi
REDIS_LOAD_FLG=true;
REDIS_LOAD_FLG=true
while $REDIS_LOAD_FLG;
do
sleep 1;
redis-cli --pass public --no-auth-warning -p 7000 info 1> /data/conf/r7000i.log 2> /dev/null;
if [ -s /data/conf/r7000i.log ]; then
:
else
continue;
sleep 1
redis-cli --pass public --no-auth-warning -p 7000 info 1> /data/conf/r7000i.log 2> /dev/null
if ! [ -s /data/conf/r7000i.log ]; then
continue
fi
redis-cli --pass public --no-auth-warning -p 7001 info 1> /data/conf/r7001i.log 2> /dev/null;
if [ -s /data/conf/r7001i.log ]; then
:
else
continue;
redis-cli --pass public --no-auth-warning -p 7001 info 1> /data/conf/r7001i.log 2> /dev/null
if ! [ -s /data/conf/r7001i.log ]; then
continue
fi
redis-cli --pass public --no-auth-warning -p 7002 info 1> /data/conf/r7002i.log 2> /dev/null;
if [ -s /data/conf/r7002i.log ]; then
:
else
continue;
if ! [ -s /data/conf/r7002i.log ]; then
continue
fi
if [ "${node}" = "cluster" ] ; then
if $tls ; then
yes "yes" | redis-cli --cluster create "$LOCAL_IP:8000" "$LOCAL_IP:8001" "$LOCAL_IP:8002" --pass public --no-auth-warning --tls true --cacert /etc/certs/ca.crt --cert /etc/certs/redis.crt --key /etc/certs/redis.key;
if [ "$node" = "cluster" ] ; then
if $tls; then
yes "yes" | redis-cli --cluster create "$LOCAL_IP:8000" "$LOCAL_IP:8001" "$LOCAL_IP:8002" \
--pass public --no-auth-warning \
--tls true --cacert /etc/certs/ca.crt \
--cert /etc/certs/redis.crt --key /etc/certs/redis.key
else
yes "yes" | redis-cli --cluster create "$LOCAL_IP:7000" "$LOCAL_IP:7001" "$LOCAL_IP:7002" --pass public --no-auth-warning;
yes "yes" | redis-cli --cluster create "$LOCAL_IP:7000" "$LOCAL_IP:7001" "$LOCAL_IP:7002" \
--pass public --no-auth-warning
fi
elif [ "${node}" = "sentinel" ] ; then
elif [ "$node" = "sentinel" ]; then
tee /_sentinel.conf>/dev/null << EOF
port 26379
bind 0.0.0.0 ::
@ -107,7 +104,7 @@ daemonize yes
logfile /var/log/redis-server.log
dir /tmp
EOF
if $tls ; then
if $tls; then
cat >>/_sentinel.conf<<EOF
tls-port 26380
tls-replication yes
@ -121,9 +118,9 @@ EOF
sentinel monitor mymaster $LOCAL_IP 7000 1
EOF
fi
redis-server /_sentinel.conf --sentinel;
redis-server /_sentinel.conf --sentinel
fi
REDIS_LOAD_FLG=false;
REDIS_LOAD_FLG=false
done
exit 0;

7
.ci/docker-compose-file/toxiproxy.json
View File

@ -22,5 +22,12 @@
"listen": "0.0.0.0:3307",
"upstream": "mysql-tls:3306",
"enabled": true
},
{
"name": "redis_single_tcp",
"listen": "0.0.0.0:6379",
"upstream": "redis:6379",
"enabled": true
}
]

5
apps/emqx/test/emqx_common_test_helpers.erl
View File

@ -467,7 +467,10 @@ is_all_tcp_servers_available(Servers) ->
fun({Host, Port}) ->
is_tcp_server_available(Host, Port)
end,
lists:all(Fun, Servers).
case lists:partition(Fun, Servers) of
{_, []} -> true;
{_, Unavail} -> ct:print("Unavailable servers: ~p", [Unavail])
end.
-spec is_tcp_server_available(
Host :: inet:socket_address() | inet:hostname(),

1
apps/emqx_authz/test/emqx_authz_api_sources_SUITE.erl
View File

@ -80,7 +80,6 @@
<<"servers">> => <<?REDIS_SINGLE_HOST, ",127.0.0.1:6380">>,
<<"redis_type">> => <<"cluster">>,
<<"pool_size">> => 1,
<<"database">> => 0,
<<"password">> => <<"ee">>,
<<"auto_reconnect">> => true,
<<"ssl">> => #{<<"enable">> => false},

16
apps/emqx_bridge/include/emqx_bridge.hrl
View File

@ -1,3 +1,19 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2020-2022 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.
%%--------------------------------------------------------------------
-define(EMPTY_METRICS,
?METRICS(
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0

19
apps/emqx_bridge/src/emqx_bridge.erl
View File

@ -52,7 +52,10 @@
T == webhook;
T == mysql;
T == influxdb_api_v1;
T == influxdb_api_v2
T == influxdb_api_v2;
T == redis_single;
T == redis_sentinel;
T == redis_cluster
%% T == influxdb_udp
).
@ -135,6 +138,7 @@ on_message_publish(Message = #message{topic = Topic, flags = Flags}) ->
{ok, Message}.
send_to_matched_egress_bridges(Topic, Msg) ->
MatchedBridgeIds = get_matched_bridges(Topic),
lists:foreach(
fun(Id) ->
try send_message(Id, Msg) of
@ -157,7 +161,7 @@ send_to_matched_egress_bridges(Topic, Msg) ->
})
end
end,
get_matched_bridges(Topic)
MatchedBridgeIds
).
send_message(BridgeId, Message) ->
@ -242,6 +246,12 @@ disable_enable(Action, BridgeType, BridgeName) when
).
create(BridgeType, BridgeName, RawConf) ->
?SLOG(debug, #{
brige_action => create,
bridge_type => BridgeType,
bridge_name => BridgeName,
bridge_raw_config => RawConf
}),
emqx_conf:update(
emqx_bridge:config_key_path() ++ [BridgeType, BridgeName],
RawConf,
@ -249,6 +259,11 @@ create(BridgeType, BridgeName, RawConf) ->
).
remove(BridgeType, BridgeName) ->
?SLOG(debug, #{
brige_action => remove,
bridge_type => BridgeType,
bridge_name => BridgeName
}),
emqx_conf:remove(
emqx_bridge:config_key_path() ++ [BridgeType, BridgeName],
#{override_to => cluster}

77
apps/emqx_connector/src/emqx_connector_redis.erl
View File

@ -37,7 +37,7 @@
-export([connect/1]).
-export([cmd/3]).
-export([do_cmd/3]).
%% redis host don't need parse
-define(REDIS_HOST_OPTIONS, #{
@ -63,7 +63,8 @@ fields(single) ->
[
{server, fun server/1},
{redis_type, #{
type => hoconsc:enum([single]),
type => single,
default => single,
required => true,
desc => ?DESC("single")
}}
@ -74,18 +75,20 @@ fields(cluster) ->
[
{servers, fun servers/1},
{redis_type, #{
type => hoconsc:enum([cluster]),
type => cluster,
default => cluster,
required => true,
desc => ?DESC("cluster")
}}
] ++
redis_fields() ++
lists:keydelete(database, 1, redis_fields()) ++
emqx_connector_schema_lib:ssl_fields();
fields(sentinel) ->
[
{servers, fun servers/1},
{redis_type, #{
type => hoconsc:enum([sentinel]),
type => sentinel,
default => sentinel,
required => true,
desc => ?DESC("sentinel")
}},
@ -119,7 +122,6 @@ on_start(
InstId,
#{
redis_type := Type,
database := Database,
pool_size := PoolSize,
auto_reconnect := AutoReconn,
ssl := SSL
@ -135,13 +137,17 @@ on_start(
single -> [{servers, [maps:get(server, Config)]}];
_ -> [{servers, maps:get(servers, Config)}]
end,
Database =
case Type of
cluster -> [];
_ -> [{database, maps:get(database, Config)}]
end,
Opts =
[
{pool_size, PoolSize},
{database, Database},
{password, maps:get(password, Config, "")},
{auto_reconnect, reconn_interval(AutoReconn)}
] ++ Servers,
] ++ Database ++ Servers,
Options =
case maps:get(enable, SSL) of
true ->
@ -157,9 +163,12 @@ on_start(
case Type of
cluster ->
case eredis_cluster:start_pool(PoolName, Opts ++ [{options, Options}]) of
{ok, _} -> {ok, State};
{ok, _, _} -> {ok, State};
{error, Reason} -> {error, Reason}
{ok, _} ->
{ok, State};
{ok, _, _} ->
{ok, State};
{error, Reason} ->
{error, Reason}
end;
_ ->
case
@ -180,23 +189,28 @@ on_stop(InstId, #{poolname := PoolName, type := Type}) ->
_ -> emqx_plugin_libs_pool:stop_pool(PoolName)
end.
on_query(InstId, {cmd, Command}, #{poolname := PoolName, type := Type} = State) ->
on_query(InstId, {cmd, _} = Query, State) ->
do_query(InstId, Query, State);
on_query(InstId, {cmds, _} = Query, State) ->
do_query(InstId, Query, State).
do_query(InstId, Query, #{poolname := PoolName, type := Type} = State) ->
?TRACE(
"QUERY",
"redis_connector_received",
#{connector => InstId, sql => Command, state => State}
#{connector => InstId, query => Query, state => State}
),
Result =
case Type of
cluster -> eredis_cluster:q(PoolName, Command);
_ -> ecpool:pick_and_do(PoolName, {?MODULE, cmd, [Type, Command]}, no_handover)
cluster -> do_cmd(PoolName, cluster, Query);
_ -> ecpool:pick_and_do(PoolName, {?MODULE, do_cmd, [Type, Query]}, no_handover)
end,
case Result of
{error, Reason} ->
?SLOG(error, #{
msg => "redis_connector_do_cmd_query_failed",
msg => "redis_connector_do_query_failed",
connector => InstId,
sql => Command,
query => Query,
reason => Reason
});
_ ->
@ -226,7 +240,7 @@ on_get_status(_InstId, #{type := cluster, poolname := PoolName, auto_reconnect :
Health = eredis_cluster_workers_exist_and_are_connected(Workers),
status_result(Health, AutoReconn);
false ->
disconnect
disconnected
end;
on_get_status(_InstId, #{poolname := Pool, auto_reconnect := AutoReconn}) ->
Health = emqx_plugin_libs_pool:health_check_ecpool_workers(Pool, fun ?MODULE:do_get_status/1),
@ -245,10 +259,29 @@ status_result(_Status = false, _AutoReconn = false) -> disconnected.
reconn_interval(true) -> 15;
reconn_interval(false) -> false.
cmd(Conn, cluster, Command) ->
eredis_cluster:q(Conn, Command);
cmd(Conn, _Type, Command) ->
eredis:q(Conn, Command).
do_cmd(PoolName, cluster, {cmd, Command}) ->
eredis_cluster:q(PoolName, Command);
do_cmd(Conn, _Type, {cmd, Command}) ->
eredis:q(Conn, Command);
do_cmd(PoolName, cluster, {cmds, Commands}) ->
wrap_qp_result(eredis_cluster:qp(PoolName, Commands));
do_cmd(Conn, _Type, {cmds, Commands}) ->
wrap_qp_result(eredis:qp(Conn, Commands)).
wrap_qp_result({error, _} = Error) ->
Error;
wrap_qp_result(Results) when is_list(Results) ->
AreAllOK = lists:all(
fun
({ok, _}) -> true;
({error, _}) -> false
end,
Results
),
case AreAllOK of
true -> {ok, Results};
false -> {error, Results}
end.
%% ===================================================================
connect(Opts) ->

27
apps/emqx_connector/test/emqx_connector_redis_SUITE.erl
View File

@ -111,6 +111,14 @@ perform_lifecycle_check(PoolName, InitialConfig, RedisCommand) ->
?assertEqual({ok, connected}, emqx_resource:health_check(PoolName)),
% Perform query as further check that the resource is working as expected
?assertEqual({ok, <<"PONG">>}, emqx_resource:query(PoolName, {cmd, RedisCommand})),
?assertEqual(
{ok, [{ok, <<"PONG">>}, {ok, <<"PONG">>}]},
emqx_resource:query(PoolName, {cmds, [RedisCommand, RedisCommand]})
),
?assertMatch(
{error, [{ok, <<"PONG">>}, {error, _}]},
emqx_resource:query(PoolName, {cmds, [RedisCommand, [<<"INVALID_COMMAND">>]]})
),
?assertEqual(ok, emqx_resource:stop(PoolName)),
% Resource will be listed still, but state will be changed and healthcheck will fail
% as the worker no longer exists.
@ -152,14 +160,14 @@ redis_config_cluster() ->
redis_config_sentinel() ->
redis_config_base("sentinel", "servers").
-define(REDIS_CONFIG_BASE(MaybeSentinel),
-define(REDIS_CONFIG_BASE(MaybeSentinel, MaybeDatabase),
"" ++
"\n" ++
" auto_reconnect = true\n" ++
" database = 1\n" ++
" pool_size = 8\n" ++
" redis_type = ~s\n" ++
MaybeSentinel ++
MaybeDatabase ++
" password = public\n" ++
" ~s = \"~s:~b\"\n" ++
" " ++
@ -171,15 +179,22 @@ redis_config_base(Type, ServerKey) ->
"sentinel" ->
Host = ?REDIS_SENTINEL_HOST,
Port = ?REDIS_SENTINEL_PORT,
MaybeSentinel = " sentinel = mymaster\n";
_ ->
MaybeSentinel = " sentinel = mymaster\n",
MaybeDatabase = " database = 1\n";
"single" ->
Host = ?REDIS_SINGLE_HOST,
Port = ?REDIS_SINGLE_PORT,
MaybeSentinel = ""
MaybeSentinel = "",
MaybeDatabase = " database = 1\n";
"cluster" ->
Host = ?REDIS_SINGLE_HOST,
Port = ?REDIS_SINGLE_PORT,
MaybeSentinel = "",
MaybeDatabase = ""
end,
RawConfig = list_to_binary(
io_lib:format(
?REDIS_CONFIG_BASE(MaybeSentinel),
?REDIS_CONFIG_BASE(MaybeSentinel, MaybeDatabase),
[Type, ServerKey, Host, Port]
)
),

2
lib-ee/emqx_ee_bridge/docker-ct
View File

@ -4,3 +4,5 @@ kafka
mongo
mongo_rs_sharded
mysql
redis
redis_cluster

73
lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_redis.conf Normal file
View File

@ -0,0 +1,73 @@
emqx_ee_bridge_redis {
local_topic {
desc {
en: """The MQTT topic filter to be forwarded to Redis. 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' 的消息都会转发到 Redis。 </br>
注意：如果这个 Bridge 被用作规则（EMQX 规则引擎）的输出，同时也配置了 'local_topic' ，那么这两部分的消息都会被转发到 Redis。
"""
}
label {
en: "Local Topic"
zh: "本地 Topic"
}
}
command_template {
desc {
en: """Redis Command Template"""
zh: """Redis Command 模板"""
}
label {
en: "Redis Command Template"
zh: "Redis Command 模板"
}
}
config_enable {
desc {
en: """Enable or disable this bridge"""
zh: """启用/禁用桥接"""
}
label {
en: "Enable Or Disable Bridge"
zh: "启用/禁用桥接"
}
}
desc_config {
desc {
en: """Configuration for a Redis bridge."""
zh: """Resis 桥接配置"""
}
label: {
en: "Redis Bridge Configuration"
zh: "Redis 桥接配置"
}
}
desc_type {
desc {
en: """The Bridge Type"""
zh: """Bridge 类型"""
}
label {
en: "Bridge Type"
zh: "桥接类型"
}
}
desc_name {
desc {
en: """Bridge name, used as a human-readable description of the bridge."""
zh: """桥接名字，可读描述"""
}
label {
en: "Bridge Name"
zh: "桥接名字"
}
}
}

32
lib-ee/emqx_ee_bridge/src/emqx_ee_bridge.erl
View File

@ -22,7 +22,10 @@ api_schemas(Method) ->
ref(emqx_ee_bridge_hstreamdb, Method),
%% ref(emqx_ee_bridge_influxdb, Method ++ "_udp"),
ref(emqx_ee_bridge_influxdb, Method ++ "_api_v1"),
ref(emqx_ee_bridge_influxdb, Method ++ "_api_v2")
ref(emqx_ee_bridge_influxdb, Method ++ "_api_v2"),
ref(emqx_ee_bridge_redis, Method ++ "_single"),
ref(emqx_ee_bridge_redis, Method ++ "_sentinel"),
ref(emqx_ee_bridge_redis, Method ++ "_cluster")
].
schema_modules() ->
@ -31,7 +34,8 @@ schema_modules() ->
emqx_ee_bridge_hstreamdb,
emqx_ee_bridge_influxdb,
emqx_ee_bridge_mongodb,
emqx_ee_bridge_mysql
emqx_ee_bridge_mysql,
emqx_ee_bridge_redis
].
examples(Method) ->
@ -55,7 +59,10 @@ resource_type(mongodb_single) -> emqx_connector_mongo;
resource_type(mysql) -> emqx_connector_mysql;
resource_type(influxdb_udp) -> emqx_ee_connector_influxdb;
resource_type(influxdb_api_v1) -> emqx_ee_connector_influxdb;
resource_type(influxdb_api_v2) -> emqx_ee_connector_influxdb.
resource_type(influxdb_api_v2) -> emqx_ee_connector_influxdb;
resource_type(redis_single) -> emqx_ee_connector_redis;
resource_type(redis_sentinel) -> emqx_ee_connector_redis;
resource_type(redis_cluster) -> emqx_ee_connector_redis.
fields(bridges) ->
[
@ -83,7 +90,7 @@ fields(bridges) ->
required => false
}
)}
] ++ mongodb_structs() ++ influxdb_structs().
] ++ mongodb_structs() ++ influxdb_structs() ++ redis_structs().
mongodb_structs() ->
[
@ -114,3 +121,20 @@ influxdb_structs() ->
influxdb_api_v2
]
].
redis_structs() ->
[
{Type,
mk(
hoconsc:map(name, ref(emqx_ee_bridge_redis, Type)),
#{
desc => <<"Redis Bridge Config">>,
required => false
}
)}
|| Type <- [
redis_single,
redis_sentinel,
redis_cluster
]
].

193
lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_redis.erl Normal file
View File

@ -0,0 +1,193 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_ee_bridge_redis).
-include_lib("emqx_bridge/include/emqx_bridge.hrl").
-include_lib("typerefl/include/types.hrl").
-include_lib("hocon/include/hoconsc.hrl").
-import(hoconsc, [mk/2, enum/1, ref/2]).
-export([
conn_bridge_examples/1
]).
-export([
namespace/0,
roots/0,
fields/1,
desc/1
]).
%% -------------------------------------------------------------------------------------------------
%% api
conn_bridge_examples(Method) ->
[
#{
<<"redis_single">> => #{
summary => <<"Redis Single Node Bridge">>,
value => values("single", Method)
}
},
#{
<<"redis_sentinel">> => #{
summary => <<"Redis Sentinel Bridge">>,
value => values("sentinel", Method)
}
},
#{
<<"redis_cluster">> => #{
summary => <<"Redis Cluster Bridge">>,
value => values("cluster", Method)
}
}
].
values(Protocol, get) ->
maps:merge(values(Protocol, post), ?METRICS_EXAMPLE);
values("single", post) ->
SpecificOpts = #{
server => <<"127.0.0.1:6379">>,
database => 1
},
values(common, "single", SpecificOpts);
values("sentinel", post) ->
SpecificOpts = #{
servers => [<<"127.0.0.1:26379">>],
sentinel => <<"mymaster">>,
database => 1
},
values(common, "sentinel", SpecificOpts);
values("cluster", post) ->
SpecificOpts = #{
servers => [<<"127.0.0.1:6379">>]
},
values(common, "cluster", SpecificOpts);
values(Protocol, put) ->
maps:without([type, name], values(Protocol, post)).
values(common, RedisType, SpecificOpts) ->
Config = #{
type => list_to_atom("redis_" ++ RedisType),
name => <<"redis_bridge">>,
enable => true,
local_topic => <<"local/topic/#">>,
pool_size => 8,
password => <<"secret">>,
auto_reconnect => true,
command_template => [<<"LPUSH">>, <<"MSGS">>, <<"${payload}">>],
resource_opts => #{
enable_batch => false,
batch_size => 100,
batch_time => <<"20ms">>
},
ssl => #{enable => false}
},
maps:merge(Config, SpecificOpts).
%% -------------------------------------------------------------------------------------------------
%% Hocon Schema Definitions
namespace() -> "bridge_redis".
roots() -> [].
fields("post_single") ->
method_fileds(post, redis_single);
fields("post_sentinel") ->
method_fileds(post, redis_sentinel);
fields("post_cluster") ->
method_fileds(post, redis_cluster);
fields("put_single") ->
method_fileds(put, redis_single);
fields("put_sentinel") ->
method_fileds(put, redis_sentinel);
fields("put_cluster") ->
method_fileds(put, redis_cluster);
fields("get_single") ->
method_fileds(get, redis_single);
fields("get_sentinel") ->
method_fileds(get, redis_sentinel);
fields("get_cluster") ->
method_fileds(get, redis_cluster);
fields(Type) when
Type == redis_single orelse Type == redis_sentinel orelse Type == redis_cluster
->
redis_bridge_common_fields() ++
connector_fields(Type).
method_fileds(post, ConnectorType) ->
redis_bridge_common_fields() ++
connector_fields(ConnectorType) ++
type_name_fields(ConnectorType);
method_fileds(get, ConnectorType) ->
redis_bridge_common_fields() ++
connector_fields(ConnectorType) ++
type_name_fields(ConnectorType) ++
emqx_bridge_schema:metrics_status_fields();
method_fileds(put, ConnectorType) ->
redis_bridge_common_fields() ++
connector_fields(ConnectorType).
redis_bridge_common_fields() ->
emqx_bridge_schema:common_bridge_fields() ++
[
{local_topic, mk(binary(), #{desc => ?DESC("local_topic")})},
{command_template, fun command_template/1}
] ++
emqx_resource_schema:fields("resource_opts").
connector_fields(Type) ->
RedisType = bridge_type_to_redis_conn_type(Type),
emqx_connector_redis:fields(RedisType).
bridge_type_to_redis_conn_type(redis_single) ->
single;
bridge_type_to_redis_conn_type(redis_sentinel) ->
sentinel;
bridge_type_to_redis_conn_type(redis_cluster) ->
cluster.
type_name_fields(Type) ->
[
{type, mk(Type, #{required => true, desc => ?DESC("desc_type")})},
{name, mk(binary(), #{required => true, desc => ?DESC("desc_name")})}
].
desc("config") ->
?DESC("desc_config");
desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
["Configuration for Redis using `", string:to_upper(Method), "` method."];
desc(redis_single) ->
?DESC(emqx_connector_redis, "single");
desc(redis_sentinel) ->
?DESC(emqx_connector_redis, "sentinel");
desc(redis_cluster) ->
?DESC(emqx_connector_redis, "cluster");
desc(_) ->
undefined.
command_template(type) ->
list(binary());
command_template(required) ->
true;
command_template(validator) ->
fun is_command_template_valid/1;
command_template(desc) ->
?DESC("command_template");
command_template(_) ->
undefined.
is_command_template_valid(CommandSegments) ->
case
is_list(CommandSegments) andalso length(CommandSegments) > 0 andalso
lists:all(fun is_binary/1, CommandSegments)
of
true ->
ok;
false ->
{error,
"the value of the field 'command_template' should be a nonempty "
"list of strings (templates for Redis command and arguments)"}
end.

493
lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_redis_SUITE.erl Normal file
View File

@ -0,0 +1,493 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_ee_bridge_redis_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").
-include_lib("emqx_bridge/include/emqx_bridge.hrl").
%%------------------------------------------------------------------------------
%% CT boilerplate
%%------------------------------------------------------------------------------
-define(REDIS_TOXYPROXY_CONNECT_CONFIG, #{
<<"server">> => <<"toxiproxy:6379">>
}).
-define(COMMON_REDIS_OPTS, #{
<<"password">> => <<"public">>,
<<"command_template">> => [<<"RPUSH">>, <<"MSGS">>, <<"${payload}">>],
<<"local_topic">> => <<"local_topic/#">>
}).
-define(BATCH_SIZE, 5).
-define(PROXY_HOST, "toxiproxy").
-define(PROXY_PORT, "8474").
all() -> [{group, redis_types}, {group, rest}].
groups() ->
ResourceSpecificTCs = [t_create_delete_bridge],
TCs = emqx_common_test_helpers:all(?MODULE) -- ResourceSpecificTCs,
TypeGroups = [
{group, redis_single},
{group, redis_sentinel},
{group, redis_cluster}
],
BatchGroups = [
{group, batch_on},
{group, batch_off}
],
[
{rest, TCs},
{redis_types, [
{group, tcp},
{group, tls}
]},
{tcp, TypeGroups},
{tls, TypeGroups},
{redis_single, BatchGroups},
{redis_sentinel, BatchGroups},
{redis_cluster, BatchGroups},
{batch_on, ResourceSpecificTCs},
{batch_off, ResourceSpecificTCs}
].
init_per_group(Group, Config) when
Group =:= redis_single; Group =:= redis_sentinel; Group =:= redis_cluster
->
[{redis_type, Group} | Config];
init_per_group(Group, Config) when
Group =:= tcp; Group =:= tls
->
[{transport, Group} | Config];
init_per_group(Group, Config) when
Group =:= batch_on; Group =:= batch_off
->
[{batch_mode, Group} | Config];
init_per_group(_Group, Config) ->
Config.
end_per_group(_Group, _Config) ->
ok.
init_per_suite(Config) ->
TestHosts = all_test_hosts(),
case emqx_common_test_helpers:is_all_tcp_servers_available(TestHosts) of
true ->
ProxyHost = os:getenv("PROXY_HOST", ?PROXY_HOST),
ProxyPort = list_to_integer(os:getenv("PROXY_PORT", ?PROXY_PORT)),
emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
ok = emqx_common_test_helpers:start_apps([emqx_conf]),
ok = emqx_connector_test_helpers:start_apps([
emqx_resource, emqx_bridge, emqx_rule_engine
]),
{ok, _} = application:ensure_all_started(emqx_connector),
[
{proxy_host, ProxyHost},
{proxy_port, ProxyPort}
| Config
];
false ->
{skip, no_redis}
end.
end_per_suite(_Config) ->
ok = delete_all_bridges(),
ok = emqx_common_test_helpers:stop_apps([emqx_conf]),
ok = emqx_connector_test_helpers:stop_apps([emqx_rule_engine, emqx_bridge, emqx_resource]),
_ = application:stop(emqx_connector),
ok.
init_per_testcase(_Testcase, Config) ->
ok = delete_all_bridges(),
case ?config(redis_type, Config) of
undefined ->
Config;
RedisType ->
Transport = ?config(transport, Config),
BatchMode = ?config(batch_mode, Config),
#{RedisType := #{Transport := RedisConnConfig}} = redis_connect_configs(),
#{BatchMode := ResourceConfig} = resource_configs(),
IsBatch = (BatchMode =:= batch_on),
BridgeConfig0 = maps:merge(RedisConnConfig, ?COMMON_REDIS_OPTS),
BridgeConfig1 = BridgeConfig0#{<<"resource_opts">> => ResourceConfig},
[{bridge_config, BridgeConfig1}, {is_batch, IsBatch} | Config]
end.
end_per_testcase(_Testcase, Config) ->
ProxyHost = ?config(proxy_host, Config),
ProxyPort = ?config(proxy_port, Config),
ok = snabbkaffe:stop(),
emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
ok = delete_all_bridges().
t_create_delete_bridge(Config) ->
Name = <<"mybridge">>,
Type = ?config(redis_type, Config),
BridgeConfig = ?config(bridge_config, Config),
IsBatch = ?config(is_batch, Config),
?assertMatch(
{ok, _},
emqx_bridge:create(Type, Name, BridgeConfig)
),
ResourceId = emqx_bridge_resource:resource_id(Type, Name),
?assertEqual(
{ok, connected},
emqx_resource:health_check(ResourceId)
),
RedisType = atom_to_binary(Type),
Action = <<RedisType/binary, ":", Name/binary>>,
RuleId = <<"my_rule_id">>,
RuleConf = #{
actions => [Action],
description => <<>>,
enable => true,
id => RuleId,
name => <<>>,
sql => <<"SELECT * FROM \"t/#\"">>
},
%% check export by rule
{ok, _} = emqx_rule_engine:create_rule(RuleConf),
_ = check_resource_queries(ResourceId, <<"t/test">>, IsBatch),
ok = emqx_rule_engine:delete_rule(RuleId),
%% check export through local topic
_ = check_resource_queries(ResourceId, <<"local_topic/test">>, IsBatch),
{ok, _} = emqx_bridge:remove(Type, Name).
% check that we provide correct examples
t_check_values(_Config) ->
lists:foreach(
fun(Method) ->
lists:foreach(
fun({RedisType, #{value := Value0}}) ->
Value = maps:without(maps:keys(?METRICS_EXAMPLE), Value0),
MethodBin = atom_to_binary(Method),
Type = string:slice(RedisType, length("redis_")),
RefName = binary_to_list(<<MethodBin/binary, "_", Type/binary>>),
Schema = conf_schema(RefName),
?assertMatch(
#{},
hocon_tconf:check_plain(Schema, #{<<"root">> => Value}, #{
atom_key => true,
required => false
})
)
end,
lists:flatmap(
fun maps:to_list/1,
emqx_ee_bridge_redis:conn_bridge_examples(Method)
)
)
end,
[put, post, get]
).
t_check_replay(Config) ->
Name = <<"toxic_bridge">>,
Type = <<"redis_single">>,
Topic = <<"local_topic/test">>,
ProxyName = "redis_single_tcp",
?assertMatch(
{ok, _},
emqx_bridge:create(Type, Name, toxiproxy_redis_bridge_config())
),
ResourceId = emqx_bridge_resource:resource_id(Type, Name),
Health = emqx_resource:health_check(ResourceId),
?assertEqual(
{ok, connected},
Health
),
?check_trace(
begin
?wait_async_action(
with_down_failure(Config, ProxyName, fun() ->
ct:sleep(100),
lists:foreach(
fun(_) ->
_ = publish_message(Topic, <<"test_payload">>)
end,
lists:seq(1, ?BATCH_SIZE)
)
end),
#{?snk_kind := redis_ee_connector_send_done, batch := true, result := {ok, _}},
10000
)
end,
fun(Trace) ->
?assert(
?strict_causality(
#{?snk_kind := redis_ee_connector_send_done, result := {error, _}},
#{?snk_kind := redis_ee_connector_send_done, result := {ok, _}},
Trace
)
)
end
),
{ok, _} = emqx_bridge:remove(Type, Name).
t_permanent_error(_Config) ->
Name = <<"invalid_command_bridge">>,
Type = <<"redis_single">>,
Topic = <<"local_topic/test">>,
Payload = <<"payload for invalid redis command">>,
?assertMatch(
{ok, _},
emqx_bridge:create(Type, Name, invalid_command_bridge_config())
),
?check_trace(
begin
?wait_async_action(
publish_message(Topic, Payload),
#{?snk_kind := redis_ee_connector_send_done},
10000
)
end,
fun(Trace) ->
?assertMatch(
[#{result := {error, _}} | _],
?of_kind(redis_ee_connector_send_done, Trace)
)
end
),
{ok, _} = emqx_bridge:remove(Type, Name).
t_create_disconnected(Config) ->
Name = <<"toxic_bridge">>,
Type = <<"redis_single">>,
?check_trace(
with_down_failure(Config, "redis_single_tcp", fun() ->
{ok, _} = emqx_bridge:create(
Type, Name, toxiproxy_redis_bridge_config()
)
end),
fun(Trace) ->
?assertMatch(
[#{error := _} | _],
?of_kind(redis_ee_connector_start_error, Trace)
),
ok
end
),
{ok, _} = emqx_bridge:remove(Type, Name).
%%------------------------------------------------------------------------------
%% Helper functions
%%------------------------------------------------------------------------------
with_down_failure(Config, Name, F) ->
ProxyPort = ?config(proxy_port, Config),
ProxyHost = ?config(proxy_host, Config),
emqx_common_test_helpers:with_failure(down, Name, ProxyHost, ProxyPort, F).
check_resource_queries(ResourceId, Topic, IsBatch) ->
RandomPayload = rand:bytes(20),
N =
case IsBatch of
true -> ?BATCH_SIZE;
false -> 1
end,
?check_trace(
begin
?wait_async_action(
lists:foreach(
fun(_) ->
_ = publish_message(Topic, RandomPayload)
end,
lists:seq(1, N)
),
#{?snk_kind := redis_ee_connector_send_done, batch := IsBatch},
1000
)
end,
fun(Trace) ->
AddedMsgCount = length(added_msgs(ResourceId, RandomPayload)),
case IsBatch of
true ->
?assertMatch(
[#{result := {ok, _}, batch := true, batch_size := ?BATCH_SIZE} | _],
?of_kind(redis_ee_connector_send_done, Trace)
),
?assertEqual(?BATCH_SIZE, AddedMsgCount);
false ->
?assertMatch(
[#{result := {ok, _}, batch := false} | _],
?of_kind(redis_ee_connector_send_done, Trace)
),
?assertEqual(1, AddedMsgCount)
end
end
).
added_msgs(ResourceId, Payload) ->
{ok, Results} = emqx_resource:simple_sync_query(
ResourceId, {cmd, [<<"LRANGE">>, <<"MSGS">>, <<"0">>, <<"-1">>]}
),
[El || El <- Results, El =:= Payload].
conf_schema(StructName) ->
#{
fields => #{},
translations => #{},
validations => [],
namespace => undefined,
roots => [{root, hoconsc:ref(emqx_ee_bridge_redis, StructName)}]
}.
delete_all_bridges() ->
lists:foreach(
fun(#{name := Name, type := Type}) ->
emqx_bridge:remove(Type, Name)
end,
emqx_bridge:list()
).
all_test_hosts() ->
Confs = [
?REDIS_TOXYPROXY_CONNECT_CONFIG
| lists:concat([
maps:values(TypeConfs)
|| TypeConfs <- maps:values(redis_connect_configs())
])
],
lists:flatmap(
fun
(#{<<"servers">> := ServersRaw}) ->
lists:map(
fun(Server) ->
parse_server(Server)
end,
string:tokens(binary_to_list(ServersRaw), ", ")
);
(#{<<"server">> := ServerRaw}) ->
[parse_server(ServerRaw)]
end,
Confs
).
parse_server(Server) ->
emqx_connector_schema_lib:parse_server(Server, #{
host_type => hostname,
default_port => 6379
}).
redis_connect_ssl_opts(Type) ->
maps:merge(
client_ssl_cert_opts(Type),
#{
<<"enable">> => <<"true">>,
<<"verify">> => <<"verify_none">>
}
).
client_ssl_cert_opts(redis_single) ->
emqx_authn_test_lib:client_ssl_cert_opts();
client_ssl_cert_opts(_) ->
Dir = code:lib_dir(emqx, etc),
#{
<<"keyfile">> => filename:join([Dir, <<"certs">>, <<"client-key.pem">>]),
<<"certfile">> => filename:join([Dir, <<"certs">>, <<"client-cert.pem">>]),
<<"cacertfile">> => filename:join([Dir, <<"certs">>, <<"cacert.pem">>])
}.
redis_connect_configs() ->
#{
redis_single => #{
tcp => #{
<<"server">> => <<"redis:6379">>
},
tls => #{
<<"server">> => <<"redis-tls:6380">>,
<<"ssl">> => redis_connect_ssl_opts(redis_single)
}
},
redis_sentinel => #{
tcp => #{
<<"servers">> => <<"redis-sentinel:26379">>,
<<"sentinel">> => <<"mymaster">>
},
tls => #{
<<"servers">> => <<"redis-sentinel-tls:26380">>,
<<"sentinel">> => <<"mymaster">>,
<<"ssl">> => redis_connect_ssl_opts(redis_sentinel)
}
},
redis_cluster => #{
tcp => #{
<<"servers">> => <<"redis-cluster:7000,redis-cluster:7001,redis-cluster:7002">>
},
tls => #{
<<"servers">> =>
<<"redis-cluster-tls:8000,redis-cluster-tls:8001,redis-cluster-tls:8002">>,
<<"ssl">> => redis_connect_ssl_opts(redis_cluster)
}
}
}.
toxiproxy_redis_bridge_config() ->
Conf0 = ?REDIS_TOXYPROXY_CONNECT_CONFIG#{
<<"resource_opts">> => #{
<<"query_mode">> => <<"async">>,
<<"enable_batch">> => <<"true">>,
<<"enable_queue">> => <<"true">>,
<<"worker_pool_size">> => <<"1">>,
<<"batch_size">> => integer_to_binary(?BATCH_SIZE),
<<"health_check_interval">> => <<"1s">>
}
},
maps:merge(Conf0, ?COMMON_REDIS_OPTS).
invalid_command_bridge_config() ->
#{redis_single := #{tcp := Conf0}} = redis_connect_configs(),
Conf1 = maps:merge(Conf0, ?COMMON_REDIS_OPTS),
Conf1#{
<<"resource_opts">> => #{
<<"enable_batch">> => <<"false">>,
<<"enable_queue">> => <<"false">>,
<<"worker_pool_size">> => <<"1">>
},
<<"command_template">> => [<<"BAD">>, <<"COMMAND">>, <<"${payload}">>]
}.
resource_configs() ->
#{
batch_off => #{
<<"query_mode">> => <<"sync">>,
<<"enable_batch">> => <<"false">>,
<<"enable_queue">> => <<"false">>
},
batch_on => #{
<<"query_mode">> => <<"async">>,
<<"enable_batch">> => <<"true">>,
<<"enable_queue">> => <<"true">>,
<<"worker_pool_size">> => <<"1">>,
<<"batch_size">> => integer_to_binary(?BATCH_SIZE)
}
}.
publish_message(Topic, Payload) ->
{ok, Client} = emqtt:start_link(),
{ok, _} = emqtt:connect(Client),
ok = emqtt:publish(Client, Topic, Payload),
ok = emqtt:stop(Client).

138
lib-ee/emqx_ee_connector/src/emqx_ee_connector_redis.erl Normal file
View File

@ -0,0 +1,138 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_ee_connector_redis).
-include_lib("emqx/include/logger.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
-behaviour(emqx_resource).
%% callbacks of behaviour emqx_resource
-export([
callback_mode/0,
on_start/2,
on_stop/2,
on_query/3,
on_batch_query/3,
on_get_status/2
]).
%% -------------------------------------------------------------------------------------------------
%% resource callbacks
%% -------------------------------------------------------------------------------------------------
callback_mode() -> always_sync.
on_start(InstId, #{command_template := CommandTemplate} = Config) ->
case emqx_connector_redis:on_start(InstId, Config) of
{ok, RedisConnSt} ->
?tp(
redis_ee_connector_start_success,
#{}
),
{ok, #{
conn_st => RedisConnSt,
command_template => preproc_command_template(CommandTemplate)
}};
{error, _} = Error ->
?tp(
redis_ee_connector_start_error,
#{error => Error}
),
Error
end.
on_stop(InstId, #{conn_st := RedisConnSt}) ->
emqx_connector_redis:on_stop(InstId, RedisConnSt).
on_get_status(InstId, #{conn_st := RedisConnSt}) ->
emqx_connector_redis:on_get_status(InstId, RedisConnSt).
on_query(
InstId,
{send_message, Data},
_State = #{
command_template := CommandTemplate, conn_st := RedisConnSt
}
) ->
Cmd = proc_command_template(CommandTemplate, Data),
?tp(
redis_ee_connector_cmd,
#{cmd => Cmd, batch => false, mode => sync}
),
Result = query(InstId, {cmd, Cmd}, RedisConnSt),
?tp(
redis_ee_connector_send_done,
#{cmd => Cmd, batch => false, mode => sync, result => Result}
),
Result;
on_query(
InstId,
Query,
_State = #{conn_st := RedisConnSt}
) ->
?tp(
redis_ee_connector_query,
#{query => Query, batch => false, mode => sync}
),
Result = query(InstId, Query, RedisConnSt),
?tp(
redis_ee_connector_send_done,
#{query => Query, batch => false, mode => sync, result => Result}
),
Result.
on_batch_query(
InstId, BatchData, _State = #{command_template := CommandTemplate, conn_st := RedisConnSt}
) ->
Cmds = process_batch_data(BatchData, CommandTemplate),
?tp(
redis_ee_connector_send,
#{batch_data => BatchData, batch => true, mode => sync}
),
Result = query(InstId, {cmds, Cmds}, RedisConnSt),
?tp(
redis_ee_connector_send_done,
#{
batch_data => BatchData,
batch_size => length(BatchData),
batch => true,
mode => sync,
result => Result
}
),
Result.
%% -------------------------------------------------------------------------------------------------
%% private helpers
%% -------------------------------------------------------------------------------------------------
query(InstId, Query, RedisConnSt) ->
case emqx_connector_redis:on_query(InstId, Query, RedisConnSt) of
{ok, _} = Ok -> Ok;
{error, no_connection} -> {error, {recoverable_error, no_connection}};
{error, _} = Error -> Error
end.
process_batch_data(BatchData, CommandTemplate) ->
lists:map(
fun({send_message, Data}) ->
proc_command_template(CommandTemplate, Data)
end,
BatchData
).
proc_command_template(CommandTemplate, Msg) ->
lists:map(
fun(ArgTks) ->
emqx_plugin_libs_rule:proc_tmpl(ArgTks, Msg, #{return => full_binary})
end,
CommandTemplate
).
preproc_command_template(CommandTemplate) ->
lists:map(
fun emqx_plugin_libs_rule:preproc_tmpl/1,
CommandTemplate
).

4
scripts/ct/run.sh
View File

@ -113,6 +113,10 @@ for dep in ${CT_DEPS}; do
'.ci/docker-compose-file/docker-compose-redis-sentinel-tcp.yaml'
'.ci/docker-compose-file/docker-compose-redis-sentinel-tls.yaml' )
;;
redis_cluster)
FILES+=( '.ci/docker-compose-file/docker-compose-redis-cluster-tcp.yaml'
'.ci/docker-compose-file/docker-compose-redis-cluster-tls.yaml' )
;;
mysql)
FILES+=( '.ci/docker-compose-file/docker-compose-mysql-tcp.yaml'
'.ci/docker-compose-file/docker-compose-mysql-tls.yaml' )