fix

config/sys-dev.config

@@ -28,6 +28,7 @@
     {api_url, "http://127.0.0.1:18090/simulator"},
     %% 支持的协议
     {endpoints, [
+      {root_dir, "/usr/local/code/database/"},
       {support_protocols, [
         http
       ]}
@@ -49,18 +50,6 @@
       {<<"test">>, <<"iot2023">>}
     ]}
 
-    %{pools, [
-    %  %% redis连接池
-    %  {redis_pool,
-    %    [{size, 10}, {max_overflow, 20}, {worker_module, eredis}],
-    %    [
-    %      {host, "127.0.0.1"},
-    %      {port, 6379},
-    %      {database, 1}
-    %    ]
-    %  }
-    %]}
-
   ]},
 
   %% 系统日志配置，使用 OTP logger

config/sys-prod.config

@@ -21,6 +21,13 @@
       {backlog, 10240}
     ]},
 
+    {endpoints, [
+      {root_dir, "/usr/local/code/database/"},
+      {support_protocols, [
+        http
+      ]}
+    ]},
+
     {udp_server, [
       {port, 18080}
     ]},
@@ -37,23 +44,8 @@
       {port, 8086},
       {token, <<"A-ZRjqMK_7NR45lXXEiR7AEtYCd1ETzq9Z61FTMQLb5O4-1hSf8sCrjdPB84e__xsrItKHL3qjJALgbYN-H_VQ==">>}
     ]}
-
-    %{pools, [
-    %  %% redis连接池
-    %  {redis_pool,
-    %    [{size, 10}, {max_overflow, 20}, {worker_module, eredis}],
-    %    [
-    %      {host, "172.30.6.175"},
-    %      {port, 26379},
-    %      {database, 1}
-    %    ]
-    %  }
-
-    %]}
-
   ]},
 
-
   %% 系统日志配置，使用 OTP logger
   {kernel, [
     %% 设置 Logger 的 primary log level

src/endpoint/endpoint_buffer.erl

@@ -19,11 +19,7 @@
 
 -record(buffer, {
     endpoint :: #endpoint{},
-    next_id = 1 :: integer(),
+    outbox :: endpoint_outbox:outbox(),
-    %% 最近一次派发的数据游标
-    cursor = 0 :: integer(),
-    %% ets存储的引用
-    tid :: ets:tid(),
     %% 定时器
     timer_pid :: pid(),
     %% 窗口大小，允许最大的未确认消息数
@@ -34,34 +30,38 @@
     acc_num = 0
 }).
 
--record(north_data, {
-    id :: integer(),
-    tuple :: any(),
-    inflight = false :: boolean()
-}).
-
 -type buffer() :: #buffer{}.
 
 -spec new(Endpoint :: #endpoint{}, WindowSize :: integer()) -> Buffer :: #buffer{}.
 new(Endpoint = #endpoint{id = Id}, WindowSize) when is_integer(WindowSize), WindowSize > 0 ->
-    %% 初始化存储
+    %% 读取配置文件
-    EtsName = list_to_atom("endpoint_buffer_ets:" ++ integer_to_list(Id)),
+    {ok, Endpoints} = application:get_env(iot, endpoints),
-    Tid = ets:new(EtsName, [ordered_set, private, {keypos, 2}]),
+    RootDir = proplists:get_value(root_dir, Endpoints),
+    OutboxDir = filename:join(RootDir, integer_to_list(Id)),
+
+    {ok, Outbox} = endpoint_outbox:open(OutboxDir, #{}),
     %% 定义重发器
     {ok, TimerPid} = endpoint_timer:start_link(?RETRY_INTERVAL),
+    #buffer{outbox = Outbox, timer_pid = TimerPid, endpoint = Endpoint, window_size = WindowSize}.
 
-    #buffer{cursor = 0, tid = Tid, timer_pid = TimerPid, endpoint = Endpoint, window_size = WindowSize}.
+-spec append(Payload :: binary(), Buffer :: #buffer{}) -> NBuffer :: #buffer{}.
-
+append(Payload, Buffer = #buffer{outbox = Outbox, window_size = WindowSize, flight_num = FlightNum}) when is_binary(Payload) ->
--spec append(Tuple :: any(), Buffer :: #buffer{}) -> NBuffer :: #buffer{}.
+    case endpoint_outbox:append(Payload, Outbox) of
-append(Tuple, Buffer = #buffer{tid = Tid, next_id = NextId, window_size = WindowSize, flight_num = FlightNum}) ->
+        {ok, _Seq, NOutbox} ->
-    NorthData = #north_data{id = NextId, tuple = Tuple},
+            NBuffer = Buffer#buffer{outbox = NOutbox},
-    true = ets:insert(Tid, NorthData),
-    NBuffer = Buffer#buffer{next_id = NextId + 1},
             case FlightNum < WindowSize of
                 true ->
                     trigger_next(NBuffer);
                 false ->
                     NBuffer
+            end;
+        {dropped, capacity_reached, NOutbox} ->
+            logger:warning("[endpoint_buffer] outbox capacity reached, endpoint_id: ~p", [buffer_endpoint_id(Buffer)]),
+            Buffer#buffer{outbox = NOutbox};
+        {error, Reason} ->
+            logger:warning("[endpoint_buffer] append outbox failed, endpoint_id: ~p, reason: ~p",
+                [buffer_endpoint_id(Buffer), Reason]),
+            Buffer
     end.
 
 -spec trigger_n(Buffer :: #buffer{}) -> NBuffer :: #buffer{}.
@@ -71,40 +71,55 @@ trigger_n(Buffer = #buffer{window_size = WindowSize}) ->
 
 %% 触发读取下一条数据
 -spec trigger_next(Buffer :: #buffer{}) -> NBuffer :: #buffer{}.
-trigger_next(Buffer = #buffer{tid = Tid, cursor = Cursor, timer_pid = TimerPid, flight_num = FlightNum, window_size = WindowSize}) ->
+trigger_next(Buffer = #buffer{outbox = Outbox, timer_pid = TimerPid, flight_num = FlightNum, window_size = WindowSize}) ->
     case FlightNum < WindowSize of
         false ->
             Buffer;
         true ->
-            case next_pending_data(Tid, Cursor) of
+            case endpoint_outbox:next(Outbox) of
-                none ->
+                eof ->
                     Buffer;
-                #north_data{id = Id, tuple = Tuple} = NorthData ->
+                {ok, Id, Payload, NOutbox} ->
-                    true = ets:insert(Tid, NorthData#north_data{inflight = true}),
                     ReceiverPid = self(),
-                    ReceiverPid ! {next_data, Id, Tuple},
+                    ReceiverPid ! {next_data, Id, Payload},
-                    endpoint_timer:task(TimerPid, Id, fun() -> ReceiverPid ! {next_data, Id, Tuple} end),
+                    endpoint_timer:task(TimerPid, Id, fun() -> ReceiverPid ! {next_data, Id, Payload} end),
-                    Buffer#buffer{flight_num = FlightNum + 1, cursor = Id}
+                    Buffer#buffer{outbox = NOutbox, flight_num = FlightNum + 1};
+                {error, Reason} ->
+                    logger:warning("[endpoint_buffer] read next outbox failed, endpoint_id: ~p, reason: ~p",
+                        [buffer_endpoint_id(Buffer), Reason]),
+                    Buffer
             end
     end.
 
 -spec ack(Id :: integer(), Buffer :: #buffer{}) -> NBuffer :: #buffer{}.
-ack(Id, Buffer = #buffer{timer_pid = TimerPid, tid = Tid, acc_num = AccNum, flight_num = FlightNum}) when is_integer(Id) ->
+ack(Id, Buffer = #buffer{timer_pid = TimerPid, outbox = Outbox, acc_num = AccNum, flight_num = FlightNum}) when is_integer(Id) ->
-    endpoint_timer:ack(TimerPid, Id),
+    case endpoint_timer:ack(TimerPid, Id) of
-    case ets:take(Tid, Id) of
+        true ->
-        [#north_data{inflight = true}] ->
+            case endpoint_outbox:ack(Id, Outbox) of
-            trigger_next(Buffer#buffer{acc_num = AccNum + 1, flight_num = max(FlightNum - 1, 0)});
+                {ok, NOutbox} ->
-        [#north_data{}] ->
+                    NBuffer = Buffer#buffer{
-            trigger_next(Buffer#buffer{acc_num = AccNum + 1});
+                        outbox = NOutbox,
-        [] ->
+                        acc_num = AccNum + 1,
+                        flight_num = max(FlightNum - 1, 0)
+                    },
+                    trigger_next(NBuffer);
+                {error, Reason} ->
+                    logger:warning("[endpoint_buffer] ack outbox failed, endpoint_id: ~p, id: ~p, reason: ~p",
+                        [buffer_endpoint_id(Buffer), Id, Reason]),
+                    Buffer
+            end;
+        false ->
             Buffer
     end.
 
 %% 获取当前统计信息
 -spec stat(Buffer :: #buffer{}) -> map().
-stat(#buffer{acc_num = AccNum, tid = Tid, flight_num = FlightNum}) ->
+stat(#buffer{acc_num = AccNum, outbox = Outbox, flight_num = FlightNum}) ->
-    QueueNum = max(ets:info(Tid, size) - FlightNum, 0),
+    OutboxStat = endpoint_outbox:stat(Outbox),
-    #{
+    WriteSeq = maps:get(write_seq, OutboxStat, 0),
+    AckedSeq = maps:get(acked_seq, OutboxStat, 0),
+    QueueNum = max(WriteSeq - AckedSeq - FlightNum, 0),
+    OutboxStat#{
         <<"acc_num">> => AccNum,
         <<"queue_num">> => QueueNum,
         <<"inflight_num">> => FlightNum
@@ -116,10 +131,16 @@ cleanup(Buffer = #buffer{timer_pid = TimerPid}) ->
     Buffer.
 
 -spec recover_inflight(Buffer :: #buffer{}) -> #buffer{}.
-recover_inflight(Buffer = #buffer{tid = Tid, timer_pid = TimerPid}) ->
+recover_inflight(Buffer = #buffer{outbox = Outbox, timer_pid = TimerPid}) ->
     endpoint_timer:cleanup(TimerPid),
-    reset_inflight(Tid, ets:first(Tid)),
+    case endpoint_outbox:reset_reader(Outbox) of
-    Buffer#buffer{cursor = 0, flight_num = 0}.
+        {ok, NOutbox} ->
+            Buffer#buffer{outbox = NOutbox, flight_num = 0};
+        {error, Reason} ->
+            logger:warning("[endpoint_buffer] recover inflight failed, endpoint_id: ~p, reason: ~p",
+                [buffer_endpoint_id(Buffer), Reason]),
+            Buffer#buffer{flight_num = 0}
+    end.
 
 -spec resize(Buffer :: #buffer{}, WindowSize :: integer()) -> #buffer{}.
 resize(Buffer = #buffer{}, WindowSize) when is_integer(WindowSize), WindowSize > 0 ->
@@ -129,40 +150,6 @@ resize(Buffer = #buffer{}, WindowSize) when is_integer(WindowSize), WindowSize >
 %%% Internal functions
 %%%===================================================================
 
--spec next_pending_data(ets:tid(), integer()) -> none | #north_data{}.
+-spec buffer_endpoint_id(buffer()) -> integer().
-next_pending_data(Tid, Cursor) ->
+buffer_endpoint_id(#buffer{endpoint = #endpoint{id = Id}}) ->
-    Key = case Cursor of
+    Id.
-        0 ->
-            ets:first(Tid);
-        _ ->
-            ets:next(Tid, Cursor)
-    end,
-    next_pending_data_by_key(Tid, Key).
-
--spec next_pending_data_by_key(ets:tid(), '$end_of_table' | integer()) -> none | #north_data{}.
-next_pending_data_by_key(_Tid, '$end_of_table') ->
-    none;
-next_pending_data_by_key(Tid, Key) ->
-    case ets:lookup(Tid, Key) of
-        [#north_data{inflight = false} = NorthData] ->
-            NorthData;
-        [_] ->
-            next_pending_data_by_key(Tid, ets:next(Tid, Key));
-        [] ->
-            next_pending_data_by_key(Tid, ets:next(Tid, Key))
-    end.
-
--spec reset_inflight(ets:tid(), '$end_of_table' | integer()) -> ok.
-reset_inflight(_Tid, '$end_of_table') ->
-    ok;
-reset_inflight(Tid, Key) ->
-    NextKey = ets:next(Tid, Key),
-    case ets:lookup(Tid, Key) of
-        [#north_data{inflight = true} = NorthData] ->
-            true = ets:insert(Tid, NorthData#north_data{inflight = false});
-        [_] ->
-            ok;
-        [] ->
-            ok
-    end,
-    reset_inflight(Tid, NextKey).

src/endpoint/endpoint_http.erl

@@ -71,14 +71,7 @@ handle_call(get_stat, _From, State = #state{buffer = Buffer}) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_cast({forward, Metric}, State = #state{buffer = Buffer, endpoint = #endpoint{config = #http_endpoint{token = Token}}}) ->
-    Tuple = case is_binary(Token) andalso Token /= <<>> of
+    NBuffer = endpoint_buffer:append(Metric, Buffer),
-        true ->
-            Sign = iot_util:sha256(erlang:iolist_to_binary([Token, Metric, Token])),
-            {Metric, Sign};
-        false ->
-            {Metric, <<>>}
-    end,
-    NBuffer = endpoint_buffer:append(Tuple, Buffer),
     {noreply, State#state{buffer = NBuffer}};
 handle_cast({reload, NEndpoint = #endpoint{matcher = NMatcher, config = #http_endpoint{pool_size = PoolSize}}}, State = #state{endpoint = #endpoint{matcher = Matcher}, buffer = Buffer}) ->
     ensure_subscription(Matcher, NMatcher),
@@ -95,14 +88,8 @@ handle_cast(cleanup, State = #state{buffer = Buffer}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
-handle_info({next_data, Id, {Metric, Sign}}, State = #state{buffer = Buffer, endpoint = #endpoint{config = #http_endpoint{url = Url}}}) ->
+handle_info({next_data, Id, Metric}, State = #state{buffer = Buffer, endpoint = #endpoint{config = #http_endpoint{url = Url, token = Token}}}) ->
-    BaseHeaders = [{<<"Content-Type">>, <<"application/json">>}],
+    Headers = [{<<"Content-Type">>, <<"application/json">>}] ++ patch_headers(Metric, Token),
-    ExtraHeaders = if
-                       Sign =:= <<>> -> [];
-                       true -> [{<<"X-Signature">>, Sign}]
-                   end,
-    Headers = BaseHeaders ++ ExtraHeaders,
-
     case hackney:request(post, Url, Headers, Metric) of
         {ok, HttpCode, _, ClientRef} when HttpCode >= 200, HttpCode < 300 ->
             RespBody = read_response_body(ClientRef),
@@ -158,3 +145,10 @@ read_response_body(ClientRef) ->
             hackney:close(ClientRef),
             Reason
     end.
+
+-spec patch_headers(Metric :: binary(), any()) -> list().
+patch_headers(Metric, Token) when is_binary(Token), Token /= <<>> ->
+    Sign = iot_util:sha256(erlang:iolist_to_binary([Token, Metric, Token])),
+    [{<<"X-Signature">>, Sign}];
+patch_headers(_, _) ->
+    [].

src/endpoint/endpoint_timer.erl

@@ -32,8 +32,9 @@
 task(Pid, Id, Task) when is_pid(Pid), is_integer(Id), is_function(Task, 0) ->
     gen_server:cast(Pid, {task, Id, Task}).
 
+-spec ack(pid(), integer()) -> boolean().
 ack(Pid, Id) when is_pid(Pid), is_integer(Id) ->
-    gen_server:cast(Pid, {ack, Id}).
+    gen_server:call(Pid, {ack, Id}).
 
 cleanup(Pid) when is_pid(Pid) ->
     gen_server:cast(Pid, cleanup).
@@ -67,6 +68,14 @@ init([RetryInterval]) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
     {stop, Reason :: term(), NewState :: #state{}}).
+handle_call({ack, Id}, _From, State = #state{timer_map = TimerMap}) ->
+    case maps:take(Id, TimerMap) of
+        error ->
+            {reply, false, State};
+        {TimerRef, NTimerMap} ->
+            is_reference(TimerRef) andalso erlang:cancel_timer(TimerRef),
+            {reply, true, State#state{timer_map = NTimerMap}}
+    end;
 handle_call(_Request, _From, State = #state{}) ->
     {reply, ok, State}.
 
@@ -81,15 +90,6 @@ handle_cast({task, Id, Task}, State = #state{retry_interval = RetryInterval, tim
     {noreply, State#state{timer_map = maps:put(Id, TimerRef, TimerMap)}};
 
 %% 取消
-handle_cast({ack, Id}, State = #state{timer_map = TimerMap}) ->
-    case maps:take(Id, TimerMap) of
-        error ->
-            {noreply, State};
-        {TimerRef, NTimerMap} ->
-            is_reference(TimerRef) andalso erlang:cancel_timer(TimerRef),
-            {noreply, State#state{timer_map = NTimerMap}}
-    end;
-
 handle_cast(cleanup, State = #state{timer_map = TimerMap}) ->
     lists:foreach(fun({_, TimerRef}) -> catch erlang:cancel_timer(TimerRef) end, maps:to_list(TimerMap)),
     {noreply, State#state{timer_map = #{}}}.