franz

-module(franz@consumers).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).

-export([start_consumer/3, start_topic_subscriber/8, start_group_subscriber/8, ack_return/1, commit_return/1]).
-export_type([consumer_partition/0, offset_reset_policy/0, offset_time/0, isolation_level/0, consumer_config/0, group_config/0, commited_offset/0, message_type/0, ack/0, commit/0, ack_or_commit/1]).

-type consumer_partition() :: {consumer_partitions, list(integer())} | all.

-type offset_reset_policy() :: reset_by_subscriber |
    reset_to_earliest |
    reset_to_latest.

-type offset_time() :: earliest | latest | {message_timestamp, integer()}.

-type isolation_level() :: read_committed | read_uncommitted.

-type consumer_config() :: {begin_offset, offset_time()} |
    {min_bytes, integer()} |
    {max_bytes, integer()} |
    {max_wait_time, integer()} |
    {sleep_timeout, integer()} |
    {prefetch_count, integer()} |
    {prefetch_bytes, integer()} |
    {offset_reset_policy, offset_reset_policy()} |
    {size_stat_window, integer()} |
    {isolation_level, isolation_level()} |
    {share_leader_conn, boolean()}.

-type group_config() :: {session_timeout_seconds, integer()} |
    {rebalance_timeout_seconds, integer()} |
    {heartbeat_rate_seconds, integer()} |
    {max_rejoin_attempts, integer()} |
    {rejoin_delay_seconds, integer()} |
    {offset_commit_interval_seconds, integer()} |
    {offset_retention_seconds, integer()}.

-type commited_offset() :: {commited_offset, integer(), integer()}.

-type message_type() :: message | message_set.

-type ack() :: any().

-type commit() :: any().

-type ack_or_commit(FOT) :: any() | {gleam_phantom, FOT}.

-file("src/franz/consumers.gleam", 66).
-spec start_consumer(franz:franz_client(), binary(), list(consumer_config())) -> {ok,
        nil} |
    {error, franz:franz_error()}.
start_consumer(Client, Topic, Options) ->
    franz_ffi:start_consumer(Client, Topic, Options).

-file("src/franz/consumers.gleam", 73).
-spec start_topic_subscriber(
    franz:franz_client(),
    binary(),
    consumer_partition(),
    list(consumer_config()),
    list(commited_offset()),
    message_type(),
    fun((integer(), franz:kafka_message(), FOZ) -> ack_or_commit(ack())),
    FOZ
) -> {ok, gleam@erlang@process:pid_()} | {error, franz:franz_error()}.
start_topic_subscriber(
    Client,
    Topic,
    Partitions,
    Consumer_config,
    Commited_offsets,
    Message_type,
    Callback,
    Init_callback_state
) ->
    franz_ffi:start_topic_subscriber(
        Client,
        Topic,
        Partitions,
        Consumer_config,
        Commited_offsets,
        Message_type,
        Callback,
        Init_callback_state
    ).

-file("src/franz/consumers.gleam", 85).
-spec start_group_subscriber(
    franz:franz_client(),
    binary(),
    list(binary()),
    list(consumer_config()),
    list(group_config()),
    message_type(),
    fun((franz:kafka_message(), FPG) -> ack_or_commit(any())),
    FPG
) -> {ok, gleam@erlang@process:pid_()} | {error, franz:franz_error()}.
start_group_subscriber(
    Client,
    Group_id,
    Topics,
    Consumer_config,
    Group_config,
    Message_type,
    Callback,
    Init_callback_state
) ->
    franz_ffi:start_group_subscriber(
        Client,
        Group_id,
        Topics,
        Consumer_config,
        Group_config,
        Message_type,
        Callback,
        Init_callback_state
    ).

-file("src/franz/consumers.gleam", 97).
-spec ack_return(any()) -> ack_or_commit(ack()).
ack_return(Cb_state) ->
    franz_ffi:ack_return(Cb_state).

-file("src/franz/consumers.gleam", 100).
-spec commit_return(any()) -> ack_or_commit(commit()).
commit_return(Cb_state) ->
    franz_ffi:commit_return(Cb_state).