-module(franz@topic_subscriber). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/franz/topic_subscriber.gleam"). -export([ack/1, new/6, with_config/2, with_commited_offset/3, start/1]). -export_type([builder/1, ack/0]). -if(?OTP_RELEASE >= 27). -define(MODULEDOC(Str), -moduledoc(Str)). -define(DOC(Str), -doc(Str)). -else. -define(MODULEDOC(Str), -compile([])). -define(DOC(Str), -compile([])). -endif. -opaque builder(FLH) :: {builder, franz:client(), binary(), franz@partitions:partitions(), list({integer(), integer()}), franz@message_type:message_type(), fun((integer(), franz:kafka_message(), FLH) -> ack()), FLH, list(franz@consumer_config:consumer_config())}. -type ack() :: any(). -file("src/franz/topic_subscriber.gleam", 24). ?DOC(" Acknowledge the processing of the message.\n"). -spec ack(any()) -> ack(). ack(Cb_state) -> franz_ffi:ack(Cb_state). -file("src/franz/topic_subscriber.gleam", 39). ?DOC(" Create a new topic subscriber builder.\n"). -spec new( franz:client(), binary(), franz@partitions:partitions(), franz@message_type:message_type(), fun((integer(), franz:kafka_message(), FLO) -> ack()), FLO ) -> builder(FLO). new(Client, Topic, Partitions, Message_type, Callback, Init_callback_state) -> {builder, Client, Topic, Partitions, [], Message_type, Callback, Init_callback_state, []}. -file("src/franz/topic_subscriber.gleam", 60). ?DOC(" Add a consumer configuration to the topic subscriber builder.\n"). -spec with_config(builder(FLQ), franz@consumer_config:consumer_config()) -> builder(FLQ). with_config(Builder, Consumer_config) -> {builder, erlang:element(2, Builder), erlang:element(3, Builder), erlang:element(4, Builder), erlang:element(5, Builder), erlang:element(6, Builder), erlang:element(7, Builder), erlang:element(8, Builder), [Consumer_config | erlang:element(9, Builder)]}. -file("src/franz/topic_subscriber.gleam", 72). ?DOC( " Add a commited offset to the topic subscriber builder.\n" " CommittedOffsets are the offsets for the messages that have been successfully processed (acknowledged), not the begin-offset to start fetching from\n" ). -spec with_commited_offset(builder(FLT), integer(), integer()) -> builder(FLT). with_commited_offset(Builder, Partition, Offset) -> {builder, erlang:element(2, Builder), erlang:element(3, Builder), erlang:element(4, Builder), [{Partition, Offset} | erlang:element(5, Builder)], erlang:element(6, Builder), erlang:element(7, Builder), erlang:element(8, Builder), erlang:element(9, Builder)}. -file("src/franz/topic_subscriber.gleam", 84). ?DOC(" Start a new topic subscriber.\n"). -spec start(builder(any())) -> {ok, gleam@erlang@process:pid_()} | {error, franz:franz_error()}. start(Builder) -> franz_ffi:start_topic_subscriber( erlang:element(2, Builder), erlang:element(3, Builder), erlang:element(4, Builder), erlang:element(9, Builder), erlang:element(5, Builder), erlang:element(6, Builder), erlang:element(7, Builder), erlang:element(8, Builder) ).