From b60582d0fbf6637c6d37b7709bccca9bc2101dd8 Mon Sep 17 00:00:00 2001 From: Luke Bakken Date: Fri, 30 Aug 2019 15:17:52 -0700 Subject: [PATCH] Start working on MQTT/AMQP translation funs that take sparkplug into account --- deps/rabbitmq_mqtt/src/rabbit_mqtt_util.erl | 28 +++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/deps/rabbitmq_mqtt/src/rabbit_mqtt_util.erl b/deps/rabbitmq_mqtt/src/rabbit_mqtt_util.erl index 3dcdf551c6..5da6f876dc 100644 --- a/deps/rabbitmq_mqtt/src/rabbit_mqtt_util.erl +++ b/deps/rabbitmq_mqtt/src/rabbit_mqtt_util.erl @@ -20,6 +20,7 @@ -export([subcription_queue_name/1, mqtt2amqp/1, + mqtt2amqp_fun/0, amqp2mqtt/1, gen_client_id/0, env/1, @@ -42,6 +43,33 @@ mqtt2amqp(Topic) -> re:replace(re:replace(Topic, "/", ".", [global]), "[\+]", "*", [global])). +mqtt2amqp_fun() -> + {ok, Mp} = re:compile("^sb[AB]v\\d\\.\\d/"), + %% > re:run("sbAv1.0/FOO/BAR/BAZ", Mp, [{capture, first}]). + %% {match,[{0,8}]} + case env(sparkplug_b) of + true -> + ok; + _ -> + end. + fun(Topic0) -> + Opts = case SparkplugB of + true -> + Offset = case re:run(Topic0, Mp, [{capture, first}]) of + {match, [{_, Idx1}]} -> + Idx1; % Idx1 is the 1-based offset of the / character + _ -> + 0 + end, + [global, {offset, Offset}]; + _ -> + [global] + end, + Topic1 = re:replace(Topic0, "/", ".", Opts), + Topic2 = re:replace(Topic1, "[\+]", "*", [global]), + erlang:iolist_to_binary(Topic2) + end. + amqp2mqtt(Topic) -> erlang:iolist_to_binary( re:replace(re:replace(Topic, "[\*]", "+", [global]),