diff --git a/apps/emqx_bridge_iotdb/include/emqx_bridge_iotdb.hrl b/apps/emqx_bridge_iotdb/include/emqx_bridge_iotdb.hrl index 5e6bf9ac5..5d693547a 100644 --- a/apps/emqx_bridge_iotdb/include/emqx_bridge_iotdb.hrl +++ b/apps/emqx_bridge_iotdb/include/emqx_bridge_iotdb.hrl @@ -5,7 +5,7 @@ -ifndef(EMQX_BRIDGE_IOTDB_HRL). -define(EMQX_BRIDGE_IOTDB_HRL, true). --define(VSN_1_0_X, 'v1.0.x'). +-define(VSN_1_X, 'v1.x'). -define(VSN_0_13_X, 'v0.13.x'). -endif. diff --git a/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.app.src b/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.app.src index 9c5108307..cebf60cb1 100644 --- a/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.app.src +++ b/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.app.src @@ -1,7 +1,7 @@ %% -*- mode: erlang -*- {application, emqx_bridge_iotdb, [ {description, "EMQX Enterprise Apache IoTDB Bridge"}, - {vsn, "0.1.0"}, + {vsn, "0.1.1"}, {modules, [ emqx_bridge_iotdb, emqx_bridge_iotdb_impl diff --git a/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.erl b/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.erl index 90e8d18a4..aa2c32589 100644 --- a/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.erl +++ b/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.erl @@ -109,10 +109,10 @@ basic_config() -> )}, {iotdb_version, mk( - hoconsc:enum([?VSN_1_0_X, ?VSN_0_13_X]), + hoconsc:enum([?VSN_1_X, ?VSN_0_13_X]), #{ desc => ?DESC("config_iotdb_version"), - default => ?VSN_1_0_X + default => ?VSN_1_X } )} ] ++ resource_creation_opts() ++ @@ -217,7 +217,7 @@ conn_bridge_example(_Method, Type) -> is_aligned => false, device_id => <<"my_device">>, base_url => <<"http://iotdb.local:18080/">>, - iotdb_version => ?VSN_1_0_X, + iotdb_version => ?VSN_1_X, connect_timeout => <<"15s">>, pool_type => <<"random">>, pool_size => 8, diff --git a/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_impl.erl b/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_impl.erl index 2f8794560..8331e715f 100644 --- a/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_impl.erl +++ b/apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb_impl.erl @@ -143,24 +143,42 @@ on_query_async(InstanceId, {send_message, Message}, ReplyFunAndArgs0, State) -> %% Internal Functions %%-------------------------------------------------------------------- -preproc_data(DataList) -> +make_parsed_payload(PayloadUnparsed) when is_binary(PayloadUnparsed) -> + emqx_utils_json:decode(PayloadUnparsed, [return_maps]); +make_parsed_payload(PayloadUnparsed) when is_list(PayloadUnparsed) -> + lists:map(fun make_parsed_payload/1, PayloadUnparsed); +make_parsed_payload( + #{ + measurement := Measurement, + data_type := DataType, + value := Value + } = Data +) -> + Data#{ + <<"measurement">> => Measurement, + <<"data_type">> => DataType, + <<"value">> => Value + }. + +preproc_data( + #{ + <<"measurement">> := Measurement, + <<"data_type">> := DataType, + <<"value">> := Value + } = Data +) -> + #{ + timestamp => emqx_plugin_libs_rule:preproc_tmpl( + maps:get(<<"timestamp">>, Data, <<"now">>) + ), + measurement => emqx_plugin_libs_rule:preproc_tmpl(Measurement), + data_type => DataType, + value => emqx_plugin_libs_rule:preproc_tmpl(Value) + }. + +preproc_data_list(DataList) -> lists:map( - fun( - #{ - measurement := Measurement, - data_type := DataType, - value := Value - } = Data - ) -> - #{ - timestamp => emqx_plugin_libs_rule:preproc_tmpl( - maps:get(<<"timestamp">>, Data, <<"now">>) - ), - measurement => emqx_plugin_libs_rule:preproc_tmpl(Measurement), - data_type => DataType, - value => emqx_plugin_libs_rule:preproc_tmpl(Value) - } - end, + fun preproc_data/1, DataList ). @@ -258,12 +276,13 @@ convert_float(Str) when is_binary(Str) -> convert_float(undefined) -> null. -make_iotdb_insert_request(Message, State) -> +make_iotdb_insert_request(MessageUnparsedPayload, State) -> + Message = maps:update_with(payload, fun make_parsed_payload/1, MessageUnparsedPayload), IsAligned = maps:get(is_aligned, State, false), DeviceId = device_id(Message, State), - IotDBVsn = maps:get(iotdb_version, State, ?VSN_1_0_X), + IotDBVsn = maps:get(iotdb_version, State, ?VSN_1_X), Payload = make_list(maps:get(payload, Message)), - PreProcessedData = preproc_data(Payload), + PreProcessedData = preproc_data_list(Payload), DataList = proc_data(PreProcessedData, Message), InitAcc = #{timestamps => [], measurements => [], dtypes => [], values => []}, Rows = replace_dtypes(aggregate_rows(DataList, InitAcc), IotDBVsn), @@ -330,15 +349,15 @@ insert_value(1, Data, [Value | Values]) -> insert_value(Index, Data, [Value | Values]) -> [[null | Value] | insert_value(Index - 1, Data, Values)]. -iotdb_field_key(is_aligned, ?VSN_1_0_X) -> +iotdb_field_key(is_aligned, ?VSN_1_X) -> <<"is_aligned">>; iotdb_field_key(is_aligned, ?VSN_0_13_X) -> <<"isAligned">>; -iotdb_field_key(device_id, ?VSN_1_0_X) -> +iotdb_field_key(device_id, ?VSN_1_X) -> <<"device">>; iotdb_field_key(device_id, ?VSN_0_13_X) -> <<"deviceId">>; -iotdb_field_key(data_types, ?VSN_1_0_X) -> +iotdb_field_key(data_types, ?VSN_1_X) -> <<"data_types">>; iotdb_field_key(data_types, ?VSN_0_13_X) -> <<"dataTypes">>. @@ -350,6 +369,8 @@ device_id(Message, State) -> case maps:get(device_id, State, undefined) of undefined -> case maps:get(payload, Message) of + #{<<"device_id">> := DeviceId} -> + DeviceId; #{device_id := DeviceId} -> DeviceId; _NotFound ->