import franz/producer_config import gleam/erlang/process.{type Pid} pub type FranzError { UnknownError ClientDown UnknownTopicOrPartition ProducerDown TopicAlreadyExists ConsumerNotFound(String) ProducerNotFound(String, Int) } pub type FranzClient pub type Partition = Int pub type Offset = Int pub type Topic = String pub type KafkaMessage { KafkaMessage( offset: Int, key: BitArray, value: BitArray, timestamp_type: TimeStampType, timestamp: Int, headers: List(#(String, String)), ) KafkaMessageSet( topic: Topic, partition: Int, high_wm_offset: Int, messages: List(KafkaMessage), ) } pub type TimeStampType { Undefined Create Append } pub type ClientConfig { RestartDelaySeconds(Int) GetMetadataTimeoutSeconds(Int) ReconnectCoolDownSeconds(Int) AllowTopicAutoCreation(Bool) AutoStartProducers(Bool) DefaultProducerConfig(List(producer_config.ProducerConfig)) UnknownTopicCacheTtl(Int) } pub opaque type ClientBuilder { ClientBuilder( bootstrap_endpoints: List(#(String, Int)), config: List(ClientConfig), ) } @external(erlang, "franz_ffi", "start_client") fn start_client( bootstrap_endpoints: List(#(String, Int)), client_config: List(ClientConfig), ) -> Result(FranzClient, FranzError) pub fn new_client(bootstrap_endpoints: List(#(String, Int))) -> ClientBuilder { ClientBuilder(bootstrap_endpoints, []) } pub fn with_config( client_builder: ClientBuilder, client_config: ClientConfig, ) -> ClientBuilder { ClientBuilder(..client_builder, config: [ client_config, ..client_builder.config ]) } pub fn start(client_builder: ClientBuilder) -> Result(FranzClient, FranzError) { start_client(client_builder.bootstrap_endpoints, client_builder.config) } @external(erlang, "franz_ffi", "stop_client") pub fn stop_client(client: FranzClient) -> Nil @external(erlang, "franz_ffi", "create_topic") pub fn create_topic( bootstrap_endpoints: List(#(String, Int)), topic: Topic, partitions: Partition, replication_factor: Int, ) -> Result(Nil, FranzError) @external(erlang, "franz_ffi", "fetch") pub fn fetch( client: FranzClient, topic: Topic, partition: Partition, offset: Offset, ) -> Result(#(Offset, KafkaMessage), FranzError)