franz

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

-export([start_client/2, stop_client/1, create_topic/4, fetch/4]).
-export_type([franz_error/0, franz_client/0, kafka_message/0, time_stamp_type/0, client_config/0, producer_config/0, compression/0]).

-type franz_error() :: unknown_error |
    client_down |
    unknown_topic_or_partition |
    producer_down |
    topic_already_exists |
    {consumer_not_found, binary()} |
    {producer_not_found, binary(), integer()}.

-type franz_client() :: any().

-type kafka_message() :: {kafka_message,
        integer(),
        bitstring(),
        bitstring(),
        time_stamp_type(),
        integer(),
        list({binary(), binary()})} |
    {kafka_message_set, binary(), integer(), integer(), list(kafka_message())}.

-type time_stamp_type() :: undefined | create | append.

-type client_config() :: {restart_delay_seconds, integer()} |
    {get_metadata_timeout_seconds, integer()} |
    {reconnect_cool_down_seconds, integer()} |
    {allow_topic_auto_creation, boolean()} |
    {auto_start_producers, boolean()} |
    {default_producer_config, list(producer_config())} |
    {unknown_topic_cache_ttl, integer()}.

-type producer_config() :: {required_acks, integer()} |
    {ack_timeout, integer()} |
    {partition_buffer_limit, integer()} |
    {partition_onwire_limit, integer()} |
    {max_batch_size, integer()} |
    {max_retries, integer()} |
    {retry_backoff_ms, integer()} |
    {compression, compression()} |
    {max_linger_ms, integer()} |
    {max_linger_count, integer()}.

-type compression() :: no_compression | gzip | snappy.

-file("src/franz.gleam", 77).
-spec start_client(list({binary(), integer()}), list(client_config())) -> {ok,
        franz_client()} |
    {error, franz_error()}.
start_client(Bootstrap_endpoints, Client_config) ->
    franz_ffi:start_client(Bootstrap_endpoints, Client_config).

-file("src/franz.gleam", 83).
-spec stop_client(franz_client()) -> nil.
stop_client(Client) ->
    franz_ffi:stop_client(Client).

-file("src/franz.gleam", 86).
-spec create_topic(list({binary(), integer()}), binary(), integer(), integer()) -> {ok,
        nil} |
    {error, franz_error()}.
create_topic(Bootstrap_endpoints, Topic, Partitions, Replication_factor) ->
    franz_ffi:create_topic(
        Bootstrap_endpoints,
        Topic,
        Partitions,
        Replication_factor
    ).

-file("src/franz.gleam", 94).
-spec fetch(franz_client(), binary(), integer(), integer()) -> {ok,
        {integer(), kafka_message()}} |
    {error, franz_error()}.
fetch(Client, Topic, Partition, Offset) ->
    franz_ffi:fetch(Client, Topic, Partition, Offset).