<commit_before>%%%
%%%   Copyright (c) 2014, 2015, Klarna AB
%%%
%%%   Licensed under the Apache License, Version 2.0 (the "License");
%%%   you may not use this file except in compliance with the License.
%%%   You may obtain a copy of the License at
%%%
%%%       http://www.apache.org/licenses/LICENSE-2.0
%%%
%%%   Unless required by applicable law or agreed to in writing, software
%%%   distributed under the License is distributed on an "AS IS" BASIS,
%%%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%%   See the License for the specific language governing permissions and
%%%   limitations under the License.
%%%

%%%=============================================================================
%%% @doc
%%% @copyright 2014, 2015 Klarna AB
%%% @end
%%%=============================================================================

-module(brod).

%% Producer API
-export([ start_producer/1
        , start_producer/3
        , stop_producer/1
        , produce/3
        , produce/4
        , produce/5
        , produce_sync/3
        , produce_sync/4
        , produce_sync/5
        ]).

%% Consumer API
-export([ start_consumer/3
        , stop_consumer/1
        , consume/2
        , consume/6
        ]).

%% Management and testing API
-export([ get_metadata/1
        , get_metadata/2
        , get_offsets/3
        , get_offsets/5
        , fetch/4
        , fetch/7
        , console_consumer/4
        , file_consumer/5
        , simple_consumer/5
        ]).

-export_type([host/0]).

%%%_* Includes -----------------------------------------------------------------
-include("brod_int.hrl").

%%%_* Types --------------------------------------------------------------------
-type host() :: {string(), integer()}.

%%%_* Macros -------------------------------------------------------------------
-define(DEFAULT_ACKS,            1). % default required acks
-define(DEFAULT_ACK_TIMEOUT,  1000). % default broker ack timeout

%%%_* API ----------------------------------------------------------------------
%% @equiv start_producer(Hosts, 1, 1000)
-spec start_producer([host()]) -> {ok, pid()} | {error, any()}.
start_producer(Hosts) ->
  start_producer(Hosts, ?DEFAULT_ACKS, ?DEFAULT_ACK_TIMEOUT).

%% @doc Start a process to publish messages to kafka.
%%      Hosts:
%%        list of "bootstrap" kafka nodes, {"hostname", 1234}
%%      RequiredAcks:
%%        How many acknowledgements the kafka broker should receive
%%        from the clustered replicas before responding to the
%%        producer.
%%        If it is 0 the broker will not send any response (this is
%%        the only case where the broker will not reply to a
%%        request). If it is 1, the broker will wait the data is
%%        written to the local log before sending a response. If it is
%%        -1 the broker will block until the message is committed by
%%        all in sync replicas before sending a response. For any
%%        number > 1 the broker will block waiting for this number of
%%        acknowledgements to occur (but the broker will never wait
%%        for more acknowledgements than there are in-sync replicas).
%%      AckTimeout:
%%        Maximum time in milliseconds the broker can await the
%%        receipt of the number of acknowledgements in
%%        RequiredAcks. The timeout is not an exact limit on
%%        the request time for a few reasons: (1) it does not
%%        include network latency, (2) the timer begins at the
%%        beginning of the processing of this request so if many
%%        requests are queued due to broker overload that wait
%%        time will not be included, (3) we will not terminate a
%%        local write so if the local write time exceeds this
%%        timeout it will not be respected.
-spec start_producer([host()], integer(), integer()) ->
                        {ok, pid()} | {error, any()}.
start_producer(Hosts, RequiredAcks, AckTimeout) ->
  brod_producer:start_link(Hosts, RequiredAcks, AckTimeout).

%% @doc Stop producer process
-spec stop_producer(pid()) -> ok.
stop_producer(Pid) ->
  brod_producer:stop(Pid).

%% @equiv produce(Pid, Topic, 0, <<>>, Value)
-spec produce(pid(), binary(), binary()) -> {ok, reference()}.
produce(Pid, Topic, Value) ->
  produce(Pid, Topic, 0, <<>>, Value).

%% @equiv produce(Pid, Topic, Partition, [{Key, Value}])
-spec produce(pid(), binary(), integer(), binary(), binary()) ->
                 {ok, reference()}.
produce(Pid, Topic, Partition, Key, Value) ->
  produce(Pid, Topic, Partition, [{Key, Value}]).

%% @doc Send one or more {key, value} messages to a broker.
%%      Returns a reference which can later be used to match on an
%%      'ack' message from producer acknowledging that the payload
%%      has been handled by kafka cluster.
%%      'ack' message has format {{Reference, ProducerPid}, ack}.
-spec produce(pid(), binary(), integer(), [{binary(), binary()}]) ->
                 {ok, reference()}.
produce(Pid, Topic, Partition, KVList) when is_list(KVList) ->
  brod_producer:produce(Pid, Topic, Partition, KVList).

%% @equiv produce_sync(Pid, Topic, 0, <<>>, Value)
-spec produce_sync(pid(), binary(), binary()) -> ok.
produce_sync(Pid, Topic, Value) ->
  produce_sync(Pid, Topic, 0, <<>>, Value).

%% @equiv produce_sync(Pid, Topic, Partition, [{Key, Value}])
-spec produce_sync(pid(), binary(), integer(), binary(), binary()) -> ok.
produce_sync(Pid, Topic, Partition, Key, Value) ->
  produce_sync(Pid, Topic, Partition, [{Key, Value}]).

%% @doc Send one or more {key, value} messages to a broker and block
%%      until producer acknowledges the payload.
-spec produce_sync(pid(), binary(), integer(), [{binary(), binary()}]) ->
        ok | no_return().
produce_sync(Pid, Topic, Partition, KVList) when is_list(KVList) ->
  MonitorRef = erlang:monitor(process, Pid),
  {ok, Ref} = produce(Pid, Topic, Partition, KVList),
  receive
    {'DOWN', MonitorRef, _, _, Info} ->
      erlang:throw({producer_down, Info});
    {{Ref, Pid}, ack} ->
      erlang:demonitor(MonitorRef, [flush]),
      ok
  end.

%% @doc Start consumer process
-spec start_consumer([host()], binary(), integer()) ->
                        {ok, pid()} | {error, any()}.
start_consumer(Hosts, Topic, Partition) ->
  brod_consumer:start_link(Hosts, Topic, Partition).

%% @doc Stop consumer process
-spec stop_consumer(pid()) -> ok.
stop_consumer(Pid) ->
  brod_consumer:stop(Pid).
<commit_msg>return {error, Info} instead of throw<commit_after>%%%
%%%   Copyright (c) 2014, 2015, Klarna AB
%%%
%%%   Licensed under the Apache License, Version 2.0 (the "License");
%%%   you may not use this file except in compliance with the License.
%%%   You may obtain a copy of the License at
%%%
%%%       http://www.apache.org/licenses/LICENSE-2.0
%%%
%%%   Unless required by applicable law or agreed to in writing, software
%%%   distributed under the License is distributed on an "AS IS" BASIS,
%%%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%%   See the License for the specific language governing permissions and
%%%   limitations under the License.
%%%

%%%=============================================================================
%%% @doc
%%% @copyright 2014, 2015 Klarna AB
%%% @end
%%%=============================================================================

-module(brod).

%% Producer API
-export([ start_producer/1
        , start_producer/3
        , stop_producer/1
        , produce/3
        , produce/4
        , produce/5
        , produce_sync/3
        , produce_sync/4
        , produce_sync/5
        ]).

%% Consumer API
-export([ start_consumer/3
        , stop_consumer/1
        , consume/2
        , consume/6
        ]).

%% Management and testing API
-export([ get_metadata/1
        , get_metadata/2
        , get_offsets/3
        , get_offsets/5
        , fetch/4
        , fetch/7
        , console_consumer/4
        , file_consumer/5
        , simple_consumer/5
        ]).

-export_type([host/0]).

%%%_* Includes -----------------------------------------------------------------
-include("brod_int.hrl").

%%%_* Types --------------------------------------------------------------------
-type host() :: {string(), integer()}.

%%%_* Macros -------------------------------------------------------------------
-define(DEFAULT_ACKS,            1). % default required acks
-define(DEFAULT_ACK_TIMEOUT,  1000). % default broker ack timeout

%%%_* API ----------------------------------------------------------------------
%% @equiv start_producer(Hosts, 1, 1000)
-spec start_producer([host()]) -> {ok, pid()} | {error, any()}.
start_producer(Hosts) ->
  start_producer(Hosts, ?DEFAULT_ACKS, ?DEFAULT_ACK_TIMEOUT).

%% @doc Start a process to publish messages to kafka.
%%      Hosts:
%%        list of "bootstrap" kafka nodes, {"hostname", 1234}
%%      RequiredAcks:
%%        How many acknowledgements the kafka broker should receive
%%        from the clustered replicas before responding to the
%%        producer.
%%        If it is 0 the broker will not send any response (this is
%%        the only case where the broker will not reply to a
%%        request). If it is 1, the broker will wait the data is
%%        written to the local log before sending a response. If it is
%%        -1 the broker will block until the message is committed by
%%        all in sync replicas before sending a response. For any
%%        number > 1 the broker will block waiting for this number of
%%        acknowledgements to occur (but the broker will never wait
%%        for more acknowledgements than there are in-sync replicas).
%%      AckTimeout:
%%        Maximum time in milliseconds the broker can await the
%%        receipt of the number of acknowledgements in
%%        RequiredAcks. The timeout is not an exact limit on
%%        the request time for a few reasons: (1) it does not
%%        include network latency, (2) the timer begins at the
%%        beginning of the processing of this request so if many
%%        requests are queued due to broker overload that wait
%%        time will not be included, (3) we will not terminate a
%%        local write so if the local write time exceeds this
%%        timeout it will not be respected.
-spec start_producer([host()], integer(), integer()) ->
                        {ok, pid()} | {error, any()}.
start_producer(Hosts, RequiredAcks, AckTimeout) ->
  brod_producer:start_link(Hosts, RequiredAcks, AckTimeout).

%% @doc Stop producer process
-spec stop_producer(pid()) -> ok.
stop_producer(Pid) ->
  brod_producer:stop(Pid).

%% @equiv produce(Pid, Topic, 0, <<>>, Value)
-spec produce(pid(), binary(), binary()) -> {ok, reference()}.
produce(Pid, Topic, Value) ->
  produce(Pid, Topic, 0, <<>>, Value).

%% @equiv produce(Pid, Topic, Partition, [{Key, Value}])
-spec produce(pid(), binary(), integer(), binary(), binary()) ->
                 {ok, reference()}.
produce(Pid, Topic, Partition, Key, Value) ->
  produce(Pid, Topic, Partition, [{Key, Value}]).

%% @doc Send one or more {key, value} messages to a broker.
%%      Returns a reference which can later be used to match on an
%%      'ack' message from producer acknowledging that the payload
%%      has been handled by kafka cluster.
%%      'ack' message has format {{Reference, ProducerPid}, ack}.
-spec produce(pid(), binary(), integer(), [{binary(), binary()}]) ->
                 {ok, reference()}.
produce(Pid, Topic, Partition, KVList) when is_list(KVList) ->
  brod_producer:produce(Pid, Topic, Partition, KVList).

%% @equiv produce_sync(Pid, Topic, 0, <<>>, Value)
-spec produce_sync(pid(), binary(), binary()) -> ok.
produce_sync(Pid, Topic, Value) ->
  produce_sync(Pid, Topic, 0, <<>>, Value).

%% @equiv produce_sync(Pid, Topic, Partition, [{Key, Value}])
-spec produce_sync(pid(), binary(), integer(), binary(), binary()) -> ok.
produce_sync(Pid, Topic, Partition, Key, Value) ->
  produce_sync(Pid, Topic, Partition, [{Key, Value}]).

%% @doc Send one or more {key, value} messages to a broker and block
%%      until producer acknowledges the payload.
-spec produce_sync(pid(), binary(), integer(), [{binary(), binary()}]) ->
        ok | {error, any()}.
produce_sync(Pid, Topic, Partition, KVList) when is_list(KVList) ->
  MonitorRef = erlang:monitor(process, Pid),
  {ok, Ref} = produce(Pid, Topic, Partition, KVList),
  receive
    {'DOWN', MonitorRef, _, _, Info} ->
      {error, Info};
    {{Ref, Pid}, ack} ->
      erlang:demonitor(MonitorRef, [flush]),
      ok
  end.

%% @doc Start consumer process
-spec start_consumer([host()], binary(), integer()) ->
                        {ok, pid()} | {error, any()}.
start_consumer(Hosts, Topic, Partition) ->
  brod_consumer:start_link(Hosts, Topic, Partition).

%% @doc Stop consumer process
-spec stop_consumer(pid()) -> ok.
stop_consumer(Pid) ->
  brod_consumer:stop(Pid).

{'fraction_non_alphanumeric': 0.19583947838236435, 'fraction_numerical': 0.010556547388030739, 'mean_word_length': 2.696987087517934, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1489706', 'n_tokens_mistral': 3816, 'n_tokens_neox': 3695, 'n_words': 1272}