From ee0d9fcc083fce02fde4d85759853569596d2ece Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Tue, 12 May 2026 16:07:59 +0800 Subject: [PATCH] fix outbox --- apps/efka/src/iot/efka_iot_client.erl | 15 +- apps/efka/src/iot/efka_iot_outbox.erl | 108 ++++--- apps/efka/src/tests/efka_iot_outbox_test.erl | 315 +++++++++++++++++++ 3 files changed, 398 insertions(+), 40 deletions(-) create mode 100644 apps/efka/src/tests/efka_iot_outbox_test.erl diff --git a/apps/efka/src/iot/efka_iot_client.erl b/apps/efka/src/iot/efka_iot_client.erl index ce0ff8f..eb727f6 100644 --- a/apps/efka/src/iot/efka_iot_client.erl +++ b/apps/efka/src/iot/efka_iot_client.erl @@ -29,6 +29,9 @@ %% 激活状态下 -define(STATE_ACTIVATED, activated). -define(SSL_PING_INTERVAL, 30000). +-define(OUTBOX_SEGMENT_RECORD_LIMIT, 2000). +-define(OUTBOX_MAX_SEGMENTS, 5). +-define(OUTBOX_MAX_RECORD_BYTES, 16 * 1024 * 1024). -record(state, { socket :: undefined | ssl:sslsocket(), @@ -78,7 +81,7 @@ start_link() -> -spec init(list()) -> {ok, atom(), #state{}}. init([]) -> - case efka_iot_outbox:open() of + case efka_iot_outbox:open(outbox_options()) of {ok, Outbox} -> erlang:start_timer(0, self(), create_transport), {ok, ?STATE_DISCONNECTED, #state{socket = undefined, outbox = Outbox}}; @@ -90,6 +93,16 @@ init([]) -> callback_mode() -> handle_event_function. +-spec outbox_options() -> map(). +outbox_options() -> + {ok, DetsDir} = application:get_env(efka, dets_dir), + #{ + dir => filename:join(DetsDir, "iot_outbox"), + segment_record_limit => ?OUTBOX_SEGMENT_RECORD_LIMIT, + max_segments => ?OUTBOX_MAX_SEGMENTS, + max_record_bytes => ?OUTBOX_MAX_RECORD_BYTES + }. + %% 异步发送数据，连接存在时直接发送；否则写入持久化 outbox。 -spec handle_event(term(), term(), atom(), #state{}) -> term(). handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) -> diff --git a/apps/efka/src/iot/efka_iot_outbox.erl b/apps/efka/src/iot/efka_iot_outbox.erl index dc31ee0..8e74f1e 100644 --- a/apps/efka/src/iot/efka_iot_outbox.erl +++ b/apps/efka/src/iot/efka_iot_outbox.erl @@ -12,6 +12,13 @@ %%% writer_segment => N}。进程启动时仍然会扫描 segment 文件，并以 segment %%% 文件里的真实数据作为准确信息；因此即使 metadata 落后，也不会导致跳过或 %%% 删除未消费的数据。 +%%% - outbox 目录通过 open/1 的 dir 参数传入，本模块不直接读取应用环境变量。 +%%% +%%% open/1 参数： +%%% - dir：outbox 数据目录。 +%%% - segment_record_limit：每个 segment 最多保存的 record 数。 +%%% - max_segments：最多保留的 live segment 数。 +%%% - max_record_bytes：允许读取的单条 record 最大 payload 字节数。 %%% %%% record 格式： %%% - 4 字节 unsigned big-endian payload size。 @@ -20,9 +27,11 @@ %%% 写入逻辑： %%% - `append/2' 将一条 record 写入当前 writer segment，随后 fsync 当前 %%% segment，推进 write_seq，并持久化 metadata。 -%%% - 每个 segment 最多保存 200000 条数据。 -%%% - outbox 最多保留 5 个 live segment。当当前 segment 已满，且已经存在 -%%% 5 个 live segment 时，新 packet 会被丢弃，已有磁盘数据保持不变。 +%%% - 每个 segment 最多保存多少条数据由 open/1 的 segment_record_limit +%%% 参数决定。 +%%% - outbox 最多保留多少个 live segment 由 open/1 的 max_segments 参数 +%%% 决定。当当前 segment 已满，且 live segment 数已经达到上限时，新 +%%% packet 会被丢弃，已有磁盘数据保持不变。 %%% - 当当前 segment 已满，但 live segment 数量少于 5 个时，writer 会关闭 %%% 旧文件，并开始写入 `segment-(N + 1).log'。 %%% @@ -44,7 +53,7 @@ %%%------------------------------------------------------------------- -module(efka_iot_outbox). --export([open/0, close/1, append/2, next/1, ack/2]). +-export([open/1, close/1, append/2, next/1, ack/2]). -export_type([outbox/0]). -record(segment, { @@ -60,6 +69,9 @@ metadata_path :: file:filename_all(), writer_segment = 1 :: pos_integer(), fd :: file:fd(), + segment_record_limit :: pos_integer(), + max_segments :: pos_integer(), + max_record_bytes :: pos_integer(), segments = [] :: [#segment{}], next_seq = 1 :: pos_integer(), write_seq = 0 :: non_neg_integer(), @@ -71,19 +83,25 @@ -define(METADATA_FILE, "metadata.term"). -define(SEGMENT_PREFIX, "segment-"). -define(SEGMENT_EXT, ".log"). --define(SEGMENT_RECORD_LIMIT, 200000). --define(MAX_SEGMENTS, 5). --define(MAX_RECORD_BYTES, 16 * 1024 * 1024). --spec open() -> {ok, outbox()} | {error, term()}. -open() -> - {ok, DetsDir} = application:get_env(efka, dets_dir), - Dir = filename:join(DetsDir, "iot_outbox"), +-type open_options() :: #{ + dir := file:filename_all(), + segment_record_limit := pos_integer(), + max_segments := pos_integer(), + max_record_bytes := pos_integer() +}. + +-spec open(open_options()) -> {ok, outbox()} | {error, term()}. +open(Options) when is_map(Options) -> + Dir = maps:get(dir, Options), + SegmentRecordLimit = positive_option(segment_record_limit, Options), + MaxSegments = positive_option(max_segments, Options), + MaxRecordBytes = positive_option(max_record_bytes, Options), MetadataPath = filename:join(Dir, ?METADATA_FILE), maybe ok ?= ensure_dir(Dir), {_MetaWriteSeq, MetaAckedSeq, MetaWriterSegment} = read_metadata(MetadataPath), - {ok, Segments} ?= load_segments(Dir), + {ok, Segments} ?= load_segments(Dir, MaxRecordBytes), ScannedWriteSeq = max_segment_end(Segments), WriteSeq = ScannedWriteSeq, AckedSeq = min(MetaAckedSeq, WriteSeq), @@ -94,6 +112,9 @@ open() -> metadata_path = MetadataPath, writer_segment = WriterSegment, fd = Fd, + segment_record_limit = SegmentRecordLimit, + max_segments = MaxSegments, + max_record_bytes = MaxRecordBytes, segments = Segments, next_seq = WriteSeq + 1, write_seq = WriteSeq, @@ -135,14 +156,14 @@ append(Packet, Outbox0) when is_binary(Packet) -> end. -spec next(outbox()) -> eof | {ok, pos_integer(), binary()} | {error, term()}. -next(#outbox{segments = Segments, acked_seq = AckedSeq}) -> +next(#outbox{segments = Segments, acked_seq = AckedSeq, max_record_bytes = MaxRecordBytes}) -> case next_segment(Segments, AckedSeq) of undefined -> eof; #segment{path = Path} -> case file:open(Path, [read, binary]) of {ok, Fd} -> - Result = read_next_unacked(Fd, AckedSeq), + Result = read_next_unacked(Fd, AckedSeq, MaxRecordBytes), _ = file:close(Fd), Result; {error, enoent} -> @@ -173,6 +194,15 @@ ack(Seq, #outbox{acked_seq = AckedSeq}) when is_integer(Seq) -> ensure_dir(Dir) -> filelib:ensure_dir(filename:join(Dir, "dummy")). +-spec positive_option(atom(), map()) -> pos_integer(). +positive_option(Key, Options) -> + case maps:get(Key, Options) of + Value when is_integer(Value), Value > 0 -> + Value; + Value -> + error({invalid_option, Key, Value}) + end. + -spec read_metadata(file:filename_all()) -> {non_neg_integer(), non_neg_integer(), pos_integer()}. read_metadata(MetadataPath) -> @@ -215,29 +245,29 @@ persist_metadata(#outbox{ {error, Reason} end. --spec load_segments(file:filename_all()) -> {ok, [#segment{}]} | {error, term()}. -load_segments(Dir) -> +-spec load_segments(file:filename_all(), pos_integer()) -> {ok, [#segment{}]} | {error, term()}. +load_segments(Dir, MaxRecordBytes) -> case file:list_dir(Dir) of {ok, Names} -> Ids = lists:sort([Id || Name <- Names, {ok, Id} <- [parse_segment_id(Name)]]), - load_segments(Dir, Ids, []); + load_segments(Dir, Ids, MaxRecordBytes, []); {error, enoent} -> {ok, []}; {error, Reason} -> {error, Reason} end. --spec load_segments(file:filename_all(), [pos_integer()], [#segment{}]) -> +-spec load_segments(file:filename_all(), [pos_integer()], pos_integer(), [#segment{}]) -> {ok, [#segment{}]} | {error, term()}. -load_segments(_Dir, [], Acc) -> +load_segments(_Dir, [], _MaxRecordBytes, Acc) -> {ok, lists:reverse(Acc)}; -load_segments(Dir, [Id | Rest], Acc) -> +load_segments(Dir, [Id | Rest], MaxRecordBytes, Acc) -> Path = segment_path(Dir, Id), - case scan_segment(Path, Id) of + case scan_segment(Path, Id, MaxRecordBytes) of {ok, undefined} -> - load_segments(Dir, Rest, Acc); + load_segments(Dir, Rest, MaxRecordBytes, Acc); {ok, Segment} -> - load_segments(Dir, Rest, [Segment | Acc]); + load_segments(Dir, Rest, MaxRecordBytes, [Segment | Acc]); {error, Reason} -> {error, Reason} end. @@ -290,9 +320,9 @@ ensure_writable_segment(Outbox = #outbox{segments = Segments, writer_segment = W case find_segment(WriterSegment, Segments) of undefined -> {ok, Outbox}; - #segment{records = Records} when Records < ?SEGMENT_RECORD_LIMIT -> + #segment{records = Records} when Records < Outbox#outbox.segment_record_limit -> {ok, Outbox}; - #segment{} when length(Segments) >= ?MAX_SEGMENTS -> + #segment{} when length(Segments) >= Outbox#outbox.max_segments -> {dropped, capacity_reached, Outbox}; #segment{} -> rotate_writer(Outbox, WriterSegment + 1) @@ -420,12 +450,12 @@ delete_files([Path | Rest]) -> {error, {delete_failed, Path, Reason}} end. --spec scan_segment(file:filename_all(), pos_integer()) -> +-spec scan_segment(file:filename_all(), pos_integer(), pos_integer()) -> {ok, #segment{} | undefined} | {error, term()}. -scan_segment(Path, Id) -> +scan_segment(Path, Id, MaxRecordBytes) -> case file:open(Path, [read, binary]) of {ok, Fd} -> - Result = scan_segment(Fd, Id, Path, 0, 0, 0), + Result = scan_segment(Fd, Id, Path, MaxRecordBytes, 0, 0, 0), _ = file:close(Fd), Result; {error, enoent} -> @@ -434,11 +464,11 @@ scan_segment(Path, Id) -> {error, Reason} end. --spec scan_segment(file:fd(), pos_integer(), file:filename_all(), +-spec scan_segment(file:fd(), pos_integer(), file:filename_all(), pos_integer(), non_neg_integer(), non_neg_integer(), non_neg_integer()) -> {ok, #segment{} | undefined} | {error, term()}. -scan_segment(Fd, Id, Path, StartSeq, EndSeq, Records) -> - case read_record(Fd) of +scan_segment(Fd, Id, Path, MaxRecordBytes, StartSeq, EndSeq, Records) -> + case read_record(Fd, MaxRecordBytes) of eof when Records =:= 0 -> {ok, undefined}; eof -> @@ -454,7 +484,7 @@ scan_segment(Fd, Id, Path, StartSeq, EndSeq, Records) -> 0 -> Seq; _ -> StartSeq end, - scan_segment(Fd, Id, Path, NStartSeq, Seq, Records + 1); + scan_segment(Fd, Id, Path, MaxRecordBytes, NStartSeq, Seq, Records + 1); {error, Reason} -> {error, Reason} end. @@ -465,26 +495,26 @@ encode_record(Seq, Packet) -> Size = byte_size(Payload), <>. --spec read_next_unacked(file:fd(), non_neg_integer()) -> +-spec read_next_unacked(file:fd(), non_neg_integer(), pos_integer()) -> eof | {ok, pos_integer(), binary()} | {error, term()}. -read_next_unacked(Fd, AckedSeq) -> - case read_record(Fd) of +read_next_unacked(Fd, AckedSeq, MaxRecordBytes) -> + case read_record(Fd, MaxRecordBytes) of eof -> eof; {ok, Seq, Packet} when Seq > AckedSeq -> {ok, Seq, Packet}; {ok, _Seq, _Packet} -> - read_next_unacked(Fd, AckedSeq); + read_next_unacked(Fd, AckedSeq, MaxRecordBytes); {error, Reason} -> {error, Reason} end. --spec read_record(file:fd()) -> eof | {ok, pos_integer(), binary()} | {error, term()}. -read_record(Fd) -> +-spec read_record(file:fd(), pos_integer()) -> eof | {ok, pos_integer(), binary()} | {error, term()}. +read_record(Fd, MaxRecordBytes) -> case file:read(Fd, 4) of eof -> eof; - {ok, <>} when Size > 0, Size =< ?MAX_RECORD_BYTES -> + {ok, <>} when Size > 0, Size =< MaxRecordBytes -> read_record_payload(Fd, Size); {ok, <>} -> {error, {invalid_record_size, Size}}; diff --git a/apps/efka/src/tests/efka_iot_outbox_test.erl b/apps/efka/src/tests/efka_iot_outbox_test.erl new file mode 100644 index 0000000..d4ef530 --- /dev/null +++ b/apps/efka/src/tests/efka_iot_outbox_test.erl @@ -0,0 +1,315 @@ +%%%------------------------------------------------------------------- +%%% @doc +%%% efka_iot_outbox 的手动模拟测试模块。 +%%% +%%% 该模块基于 gen_server 实现，用两个定时 tick 模拟一边写入、一边读取： +%%% - write_tick 每次向 efka_iot_outbox 写入一条 packet。 +%%% - read_tick 每次从 efka_iot_outbox 读取并 ack 一条 packet。 +%%% - 默认 read_interval = write_interval * 2，因此读取速度是写入速度的 +%%% 50%。 +%%% - 读取到的数据会通过 logger:debug/2 输出。 +%%% +%%% 默认使用独立临时 dir，避免污染真实运行目录。也可以通过 start_link/1 +%%% 传入 {dir, Dir} 指定 outbox 目录。 +%%% +%%% 示例： +%%% 
+%%% efka_iot_outbox_test:start_link([
+%%%     {dir, "/private/tmp/efka_iot_outbox_test"},
+%%%     {write_interval, 100},
+%%%     {read_interval, 200},
+%%%     {max_writes, 1000}
+%%% ]).
+%%% efka_iot_outbox_test:stats().
+%%% efka_iot_outbox_test:stop().
+%%%
+%%% @end +%%%------------------------------------------------------------------- +-module(efka_iot_outbox_test). + +-behaviour(gen_server). + +-export([start/0, start/1, start_link/0, start_link/1, stop/0, stats/0]). +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). +-export([test/0]). + +-define(SERVER, ?MODULE). +-define(DEFAULT_WRITE_INTERVAL, 100). +-define(DEFAULT_SEGMENT_RECORD_LIMIT, 2000). +-define(DEFAULT_MAX_SEGMENTS, 5). +-define(DEFAULT_MAX_RECORD_BYTES, 16 * 1024 * 1024). + +-record(state, { + outbox :: efka_iot_outbox:outbox(), + dir :: file:filename_all(), + write_interval = ?DEFAULT_WRITE_INTERVAL :: pos_integer(), + read_interval = ?DEFAULT_WRITE_INTERVAL * 2 :: pos_integer(), + max_writes = infinity :: pos_integer() | infinity, + write_timer = undefined :: undefined | reference(), + read_timer = undefined :: undefined | reference(), + write_count = 0 :: non_neg_integer(), + appended_count = 0 :: non_neg_integer(), + read_count = 0 :: non_neg_integer(), + dropped_count = 0 :: non_neg_integer(), + error_count = 0 :: non_neg_integer() +}). + +-type option() :: + {dir, file:filename_all()} | + {write_interval, pos_integer()} | + {read_interval, pos_integer()} | + {max_writes, pos_integer() | infinity} | + {segment_record_limit, pos_integer()} | + {max_segments, pos_integer()} | + {max_record_bytes, pos_integer()}. + + +test() -> + efka_iot_outbox_test:start_link([ + {write_interval, 10}, + {read_interval, 20}, + {max_writes, 100_0000} + ]). + +-spec start() -> {ok, pid()} | {error, term()}. +start() -> + start([]). + +-spec start([option()] | map()) -> {ok, pid()} | {error, term()}. +start(Options) -> + gen_server:start({local, ?SERVER}, ?MODULE, Options, []). + +-spec start_link() -> {ok, pid()} | {error, term()}. +start_link() -> + start_link([]). + +-spec start_link([option()] | map()) -> {ok, pid()} | {error, term()}. +start_link(Options) -> + gen_server:start_link({local, ?SERVER}, ?MODULE, Options, []). + +-spec stop() -> ok. +stop() -> + gen_server:stop(?SERVER). + +-spec stats() -> map(). +stats() -> + gen_server:call(?SERVER, stats). + +-spec init([option()] | map()) -> {ok, #state{}} | {stop, term()}. +init(Options0) -> + process_flag(trap_exit, true), + Options = normalize_options(Options0), + WriteInterval = positive_option(write_interval, Options, ?DEFAULT_WRITE_INTERVAL), + ReadInterval = positive_option(read_interval, Options, WriteInterval * 2), + SegmentRecordLimit = positive_option(segment_record_limit, Options, ?DEFAULT_SEGMENT_RECORD_LIMIT), + MaxSegments = positive_option(max_segments, Options, ?DEFAULT_MAX_SEGMENTS), + MaxRecordBytes = positive_option(max_record_bytes, Options, ?DEFAULT_MAX_RECORD_BYTES), + MaxWrites = max_writes_option(Options), + + Dir = "/private/tmp/efka_outbox/", + ok = filelib:ensure_dir(filename:join(Dir, "dummy")), + OutboxOptions = #{ + dir => Dir, + segment_record_limit => SegmentRecordLimit, + max_segments => MaxSegments, + max_record_bytes => MaxRecordBytes + }, + case efka_iot_outbox:open(OutboxOptions) of + {ok, Outbox} -> + State0 = #state{ + outbox = Outbox, + dir = Dir, + write_interval = WriteInterval, + read_interval = ReadInterval, + max_writes = MaxWrites + }, + State1 = schedule_write(State0, 0), + State2 = schedule_read(State1, ReadInterval), + {ok, State2}; + {error, Reason} -> + {stop, Reason} + end. + +-spec handle_call(term(), gen_server:from(), #state{}) -> + {reply, term(), #state{}}. +handle_call(stats, _From, State) -> + {reply, state_stats(State), State}; +handle_call(_Request, _From, State) -> + {reply, {error, unknown_call}, State}. + +-spec handle_cast(term(), #state{}) -> {noreply, #state{}}. +handle_cast(_Request, State) -> + {noreply, State}. + +-spec handle_info(term(), #state{}) -> {noreply, #state{}}. +handle_info(write_tick, State0 = #state{write_timer = WriteTimer}) -> + State1 = State0#state{write_timer = undefined}, + State2 = case can_write(State1) of + true -> + append_one(State1); + false -> + State1 + end, + State3 = case can_write(State2) of + true -> + schedule_write(State2, State2#state.write_interval); + false -> + State2 + end, + _ = WriteTimer, + {noreply, State3}; +handle_info(read_tick, State0 = #state{read_timer = ReadTimer}) -> + State1 = State0#state{read_timer = undefined}, + State2 = read_one(State1), + State3 = schedule_read(State2, State2#state.read_interval), + _ = ReadTimer, + {noreply, State3}; +handle_info(_Info, State) -> + {noreply, State}. + +-spec terminate(term(), #state{}) -> ok. +terminate(_Reason, #state{outbox = Outbox, write_timer = WriteTimer, read_timer = ReadTimer}) -> + cancel_timer(WriteTimer), + cancel_timer(ReadTimer), + efka_iot_outbox:close(Outbox), + ok. + +-spec code_change(term(), #state{}, term()) -> {ok, #state{}}. +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +-spec append_one(#state{}) -> #state{}. +append_one(State = #state{outbox = Outbox, write_count = WriteCount}) -> + PacketId = WriteCount + 1, + Packet = term_to_binary(#{ + source => efka_iot_outbox_test, + packet_id => PacketId, + monotonic_time => erlang:monotonic_time(millisecond) + }), + case efka_iot_outbox:append(Packet, Outbox) of + {ok, NOutbox} -> + State#state{ + outbox = NOutbox, + write_count = PacketId, + appended_count = State#state.appended_count + 1 + }; + {dropped, capacity_reached, NOutbox} -> + logger:warning("[efka_iot_outbox_test] outbox capacity reached, packet_id: ~p", [PacketId]), + State#state{ + outbox = NOutbox, + write_count = PacketId, + dropped_count = State#state.dropped_count + 1 + }; + {error, Reason} -> + logger:warning("[efka_iot_outbox_test] append failed, packet_id: ~p, reason: ~p", [PacketId, Reason]), + State#state{ + write_count = PacketId, + error_count = State#state.error_count + 1 + } + end. + +-spec read_one(#state{}) -> #state{}. +read_one(State = #state{outbox = Outbox}) -> + case efka_iot_outbox:next(Outbox) of + {ok, Seq, Packet} -> + logger:debug("[efka_iot_outbox_test] read seq: ~p, packet: ~p", [Seq, decode_packet(Packet)]), + case efka_iot_outbox:ack(Seq, Outbox) of + {ok, NOutbox} -> + State#state{ + outbox = NOutbox, + read_count = State#state.read_count + 1 + }; + {error, Reason} -> + logger:warning("[efka_iot_outbox_test] ack failed, seq: ~p, reason: ~p", [Seq, Reason]), + State#state{error_count = State#state.error_count + 1} + end; + eof -> + State; + {error, Reason} -> + logger:warning("[efka_iot_outbox_test] read failed, reason: ~p", [Reason]), + State#state{error_count = State#state.error_count + 1} + end. + +-spec schedule_write(#state{}, non_neg_integer()) -> #state{}. +schedule_write(State, Delay) -> + Ref = erlang:send_after(Delay, self(), write_tick), + State#state{write_timer = Ref}. + +-spec schedule_read(#state{}, non_neg_integer()) -> #state{}. +schedule_read(State, Delay) -> + Ref = erlang:send_after(Delay, self(), read_tick), + State#state{read_timer = Ref}. + +-spec cancel_timer(undefined | reference()) -> ok. +cancel_timer(undefined) -> + ok; +cancel_timer(Ref) -> + _ = erlang:cancel_timer(Ref), + ok. + +-spec can_write(#state{}) -> boolean(). +can_write(#state{max_writes = infinity}) -> + true; +can_write(#state{write_count = WriteCount, max_writes = MaxWrites}) -> + WriteCount < MaxWrites. + +-spec state_stats(#state{}) -> map(). +state_stats(#state{ + dir = Dir, + write_interval = WriteInterval, + read_interval = ReadInterval, + max_writes = MaxWrites, + write_count = WriteCount, + appended_count = AppendedCount, + read_count = ReadCount, + dropped_count = DroppedCount, + error_count = ErrorCount +}) -> + #{ + dir => Dir, + write_interval => WriteInterval, + read_interval => ReadInterval, + max_writes => MaxWrites, + write_count => WriteCount, + appended_count => AppendedCount, + read_count => ReadCount, + dropped_count => DroppedCount, + error_count => ErrorCount, + pending_count => AppendedCount - ReadCount + }. + +-spec normalize_options([option()] | map()) -> map(). +normalize_options(Options) when is_map(Options) -> + Options; +normalize_options(Options) when is_list(Options) -> + maps:from_list(Options). + +-spec positive_option(atom(), map(), pos_integer()) -> pos_integer(). +positive_option(Key, Options, Default) -> + case maps:get(Key, Options, Default) of + Value when is_integer(Value), Value > 0 -> + Value; + Value -> + error({invalid_option, Key, Value}) + end. + +-spec max_writes_option(map()) -> pos_integer() | infinity. +max_writes_option(Options) -> + case maps:get(max_writes, Options, infinity) of + infinity -> + infinity; + Value when is_integer(Value), Value > 0 -> + Value; + Value -> + error({invalid_option, max_writes, Value}) + end. + +-spec decode_packet(binary()) -> term(). +decode_packet(Packet) -> + try binary_to_term(Packet, [safe]) of + Term -> + Term + catch + error:_ -> + Packet + end.