diff --git a/apps/iot/src/influxdb/influx_client.erl b/apps/iot/src/influxdb/influx_client.erl index ce67900..46eb613 100644 --- a/apps/iot/src/influxdb/influx_client.erl +++ b/apps/iot/src/influxdb/influx_client.erl @@ -70,13 +70,23 @@ write_data(Measurement, Tags, FieldsList, Timestamp) when is_binary(Measurement) influx_point:new(Measurement, Tags, NFields, Timestamp) end, NFieldsList), Precision = influx_client:get_precision(Timestamp), + Bucket = get_bucket(Measurement), %poolboy:transaction(influx_pool_backup, fun(Pid) -> influx_client:write(Pid, ?DEFAULT_BUCKET, ?DEFAULT_ORG, Precision, Points) end); - poolboy:transaction(influx_pool, fun(Pid) -> influx_client:write(Pid, ?DEFAULT_BUCKET, ?DEFAULT_ORG, Precision, Points) end); + poolboy:transaction(influx_pool, fun(Pid) -> influx_client:write(Pid, Bucket, ?DEFAULT_ORG, Precision, Points) end); false -> ok end. +-spec get_bucket(DeviceUUID :: binary()) -> binary(). +get_bucket(DeviceUUID) when is_binary(DeviceUUID) -> + case iot_env:exists(DeviceUUID) of + true -> + <<"metric_", DeviceUUID/binary>>; + false -> + ?DEFAULT_BUCKET + end. + -spec write(Pid :: pid(), Bucket :: binary(), Org :: binary(), Points :: list()) -> no_return(). write(Pid, Bucket, Org, Points) when is_pid(Pid), is_binary(Bucket), is_binary(Org), is_list(Points) -> write(Pid, Bucket, Org, <<"ms">>, Points). diff --git a/apps/iot/src/iot_app.erl b/apps/iot/src/iot_app.erl index 8402d47..ea7eee3 100644 --- a/apps/iot/src/iot_app.erl +++ b/apps/iot/src/iot_app.erl @@ -15,6 +15,8 @@ start(_StartType, _StartArgs) -> io:setopts([{encoding, unicode}]), %% 加速内存的回收 erlang:system_flag(fullsweep_after, 16), + %% 加载环境变量 + ok = iot_env:new(), %% 启动数据库 % start_mnesia(), diff --git a/apps/iot/src/iot_env.erl b/apps/iot/src/iot_env.erl new file mode 100644 index 0000000..ad52894 --- /dev/null +++ b/apps/iot/src/iot_env.erl @@ -0,0 +1,48 @@ +%%%------------------------------------------------------------------- +%%% @author anlicheng +%%% @copyright (C) 2024, +%%% @doc +%%% +%%% @end +%%% Created : 30. 8月 2024 10:44 +%%%------------------------------------------------------------------- +-module(iot_env). +-author("anlicheng"). + +%% API +-export([new/0, reload/0]). +-export([exists/1]). + +-spec new() -> ok. +new() -> + ets:new(iot_env, [public, set, named_table, {read_concurrency, true}]), + DeviceUUIDs = load_env(), + true = ets:insert(iot_env, {device_uuids, DeviceUUIDs}), + ok. + +-spec reload() -> ok. +reload() -> + DeviceUUIDs = load_env(), + true = ets:insert(iot_env, {device_uuids, DeviceUUIDs}), + ok. + +-spec exists(DeviceUUID :: binary()) -> boolean(). +exists(DeviceUUID) when is_binary(DeviceUUID) -> + case ets:lookup(iot_env, device_uuids) of + [] -> + false; + [{device_uuids, DeviceUUIDs}|_] -> + lists:member(DeviceUUID, DeviceUUIDs) + end. + +-spec load_env() -> [binary()]. +load_env() -> + RootDir = code:root_dir(), + EnvFile = RootDir ++ "/.env", + case file:read_file(EnvFile) of + {ok, Content} -> + binary:split(string:trim(Content), <<"\n">>, [global, trim]); + {error, Reason} -> + lager:warning("[iot] read .env file get error: ~p", [Reason]), + [] + end. \ No newline at end of file