Current section

Files

Jump to
franz src franz@topic_subscriber.erl
Raw

src/franz@topic_subscriber.erl

-module(franz@topic_subscriber).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-export([ack/1, new/6, with_config/2, with_commited_offset/3, start/1]).
-export_type([builder/1, ack/0]).
-opaque builder(FTG) :: {builder,
franz:franz_client(),
binary(),
franz@partitions:partitions(),
list({integer(), integer()}),
franz@message_type:message_type(),
fun((integer(), franz:kafka_message(), FTG) -> ack()),
FTG,
list(franz@consumer_config:consumer_config())}.
-type ack() :: any().
-file("src/franz/topic_subscriber.gleam", 23).
-spec ack(any()) -> ack().
ack(Cb_state) ->
franz_ffi:ack(Cb_state).
-file("src/franz/topic_subscriber.gleam", 37).
-spec new(
franz:franz_client(),
binary(),
franz@partitions:partitions(),
franz@message_type:message_type(),
fun((integer(), franz:kafka_message(), FTN) -> ack()),
FTN
) -> builder(FTN).
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", 57).
-spec with_config(builder(FTP), franz@consumer_config:consumer_config()) -> builder(FTP).
with_config(Builder, Consumer_config) ->
_record = Builder,
{builder,
erlang:element(2, _record),
erlang:element(3, _record),
erlang:element(4, _record),
erlang:element(5, _record),
erlang:element(6, _record),
erlang:element(7, _record),
erlang:element(8, _record),
[Consumer_config | erlang:element(9, Builder)]}.
-file("src/franz/topic_subscriber.gleam", 67).
-spec with_commited_offset(builder(FTS), integer(), integer()) -> builder(FTS).
with_commited_offset(Builder, Partition, Offset) ->
_record = Builder,
{builder,
erlang:element(2, _record),
erlang:element(3, _record),
erlang:element(4, _record),
[{Partition, Offset} | erlang:element(5, Builder)],
erlang:element(6, _record),
erlang:element(7, _record),
erlang:element(8, _record),
erlang:element(9, _record)}.
-file("src/franz/topic_subscriber.gleam", 78).
-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)
).