-module(franz). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -export([start_client/2, produce_sync_offset/5, produce_sync/5, produce/5, create_topic/4, start_topic_subscriber/7, ack_return/1, start_consumer/3, stop_client/1, produce_cb/6, fetch/4]). -export_type([franz_error/0, franz_client/0, ack/1, kafka_message/0, time_stamp_type/0, consumer_config/0, offset_time/0, isolation_level/0, offset_reset_policy/0, client_config/0, producer_config/0, compression/0, producer_partition/0, partitioner/0, value/0, consumer_partition/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 ack(FOA) :: any() | {gleam_phantom, FOA}. -type kafka_message() :: {kafka_message, integer(), bitstring(), bitstring(), time_stamp_type(), integer(), list({binary(), binary()})}. -type time_stamp_type() :: undefined | create | append. -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 offset_time() :: earliest | latest | {message_timestamp, integer()}. -type isolation_level() :: read_committed | read_uncommitted. -type offset_reset_policy() :: reset_by_subscriber | reset_to_earliest | reset_to_latest. -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. -type producer_partition() :: {partition, integer()} | {partitioner, partitioner()}. -type partitioner() :: {partition_fun, fun((integer(), integer(), bitstring(), bitstring()) -> {ok, integer()} | {error, nil})} | random | hash. -type value() :: {value, bitstring(), list({binary(), binary()})} | {value_with_timestamp, bitstring(), integer(), list({binary(), binary()})}. -type consumer_partition() :: {consumer_partitions, list(integer())} | all. -file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 120). -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("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 126). -spec produce_sync_offset( franz_client(), binary(), producer_partition(), bitstring(), value() ) -> {ok, integer()} | {error, franz_error()}. produce_sync_offset(Client, Topic, Partition, Key, Value) -> franz_ffi:produce_sync_offset(Client, Topic, Partition, Key, Value). -file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 135). -spec produce_sync( franz_client(), binary(), producer_partition(), bitstring(), value() ) -> {ok, nil} | {error, franz_error()}. produce_sync(Client, Topic, Partition, Key, Value) -> franz_ffi:produce_sync(Client, Topic, Partition, Key, Value). -file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 144). -spec produce( franz_client(), binary(), producer_partition(), bitstring(), value() ) -> {ok, nil} | {error, franz_error()}. produce(Client, Topic, Partition, Key, Value) -> franz_ffi:produce(Client, Topic, Partition, Key, Value). -file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 153). -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("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 161). -spec start_topic_subscriber( franz_client(), binary(), consumer_partition(), list(consumer_config()), list({integer(), integer()}), fun((integer(), any(), FOR) -> ack(FOR)), FOR ) -> {ok, gleam@erlang@process:pid_()} | {error, franz_error()}. start_topic_subscriber( Client, Topic, Partitions, Consumer_config, Commited_offsets, Callback, Init_callback_state ) -> franz_ffi:start_topic_subscriber( Client, Topic, Partitions, Consumer_config, Commited_offsets, Callback, Init_callback_state ). -file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 172). -spec ack_return(FOV) -> ack(FOV). ack_return(Cb_state) -> franz_ffi:ack_return(Cb_state). -file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 175). -spec start_consumer(franz_client(), binary(), list(consumer_config())) -> {ok, nil} | {error, franz_error()}. start_consumer(Client, Topic, Options) -> franz_ffi:start_consumer(Client, Topic, Options). -file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 182). -spec stop_client(franz_client()) -> nil. stop_client(Client) -> franz_ffi:stop_client(Client). -file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 185). -spec produce_cb( franz_client(), binary(), producer_partition(), bitstring(), bitstring(), fun((integer(), integer()) -> any()) ) -> {ok, integer()} | {error, franz_error()}. produce_cb(Client, Topic, Partition, Key, Value, Callback) -> franz_ffi:produce_cb(Client, Topic, Partition, Key, Value, Callback). -file("/Users/renata-amutio/Projects/renatillas/franz/src/franz.gleam", 195). -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).