anlicheng/iot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

simple influxdb

Browse Source
This commit is contained in:
anlicheng 2025-04-08 11:22:40 +08:00
parent 80c3508072
commit ba6b7d9f43
1 changed files with 4 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
apps/iot/src/influxdb/influx_client_pool.erl
View File

@ -149,18 +149,10 @@ handle_cast({write_data, Bucket, Points}, State = #state{pool_pid = PoolPid, buf
{stop, Reason :: term(), NewState :: #state{}}). {stop, Reason :: term(), NewState :: #state{}}).
handle_info({timeout, _, flush_ticker}, State = #state{buffers = Buffers, pool_pid = PoolPid}) -> handle_info({timeout, _, flush_ticker}, State = #state{buffers = Buffers, pool_pid = PoolPid}) ->
erlang:start_timer(5000, self(), flush_ticker), erlang:start_timer(5000, self(), flush_ticker),
[begin
Acc = lists:foldl(fun({Bucket, Q}, Acc0) -> Len = queue:len(Q),
Len = queue:len(Q), Len > 0 andalso poolboy:transaction(PoolPid, fun(Pid) -> influx_client:write(Pid, Bucket, ?DEFAULT_ORG, <<"ms">>, queue:to_list(Q)) end)
case Len > 0 of end || {Bucket, Q} <- maps:to_list(Buffers)],
true ->
poolboy:transaction(PoolPid, fun(Pid) -> influx_client:write(Pid, Bucket, ?DEFAULT_ORG, <<"ms">>, queue:to_list(Q)) end),
Acc0 + Len;
false ->
Acc0
end
end, 0, maps:to_list(Buffers)),
lager:debug("[influx_client_pool] flush_ticker acc write num: ~p", [Acc]),
{noreply, State#state{buffers = #{}}}; {noreply, State#state{buffers = #{}}};