Packages

kafka_ex_tc

0.12.1-21
0.13.0 0.12.1 0.12.1-51-1 0.12.1-50-3 0.12.1-50-2 0.12.1-50-1 0.12.1-49-2 0.12.1-49-1 0.12.1-49-0 0.12.1-48-2 0.12.1-48-1 0.12.1-47-1 0.12.1-46-2 0.12.1-45-2 0.12.1-45-1 0.12.1-44-1 0.12.1-43-1 0.12.1-42-8 0.12.1-42-6 0.12.1-42-5 0.12.1-42-4 0.12.1-42-1 0.12.1-41-2 0.12.1-41-1 0.12.1-40-9 0.12.1-40-8 0.12.1-40-7 0.12.1-40-6 0.12.1-40-5 0.12.1-40-4 0.12.1-40-3 0.12.1-40-2 0.12.1-40-10 0.12.1-40-1 0.12.1-39-2 0.12.1-39-1 0.12.1-36-3 0.12.1-36-2 0.12.1-36-1 0.12.1-34-b 0.12.1-34-a 0.12.1-33-1 0.12.1-32-a1 0.12.1-29-s-3 0.12.1-29-s-2 0.12.1-29-s-1 0.12.1-28-rumble-1 0.12.1-27-2 0.12.1-27-1 0.12.1-26-rumble-2 0.12.1-26-rumble-1 0.12.1-26-dev-1 0.12.1-26-dev.1 0.12.1-26-3 0.12.1-26-2 0.12.1-26-1 0.12.1-25-s-8 0.12.1-25-s-7 0.12.1-25-s-6 0.12.1-25-s-5 0.12.1-25-s-4 0.12.1-25-s-3 0.12.1-25-s-2 0.12.1-25-s-1 0.12.1-25-dev-22 0.12.1-25-dev 0.12.1-24-2 0.12.1-24-1 0.12.1-22-poc-b-7 0.12.1-20-poc-b-6 0.12.1-20-poc-b-5 0.12.1-20-poc-b-4 0.12.1-20-poc-b-3 0.12.1-19-poc-b-3 0.12.1-19-poc-b-2 0.12.1-19-poc-b-1 0.12.1-19-poc-7 0.12.1-19-poc-6 0.12.1-19-poc-4 0.12.1-19-poc-3 0.12.1-19-poc-2 0.12.1-19-poc-1 0.12.1-19-poc 0.12.1-11-debug-1 0.12.1-11-debug 0.12.1-39 0.12.1-38.a 0.12.1-38 0.12.1-37.c 0.12.1-37.b 0.12.1-37.a 0.12.1-37.1 0.12.1-37 0.12.1-35.b 0.12.1-35.a 0.12.1-34 0.12.1-33 0.12.1-32 0.12.1-31 0.12.1-30 0.12.1-29 0.12.1-28 0.12.1-27 0.12.1-26 0.12.1-25 0.12.1-24 0.12.1-23 0.12.1-22.1 0.12.1-22 0.12.1-21 0.12.1-20 0.12.1-19 0.12.1-18 0.12.1-17 0.12.1-16 0.12.1-15 0.12.1-14 0.12.1-13 0.12.1-12 0.12.1-11 0.12.1-10 0.12.1-9 0.12.1-8 0.12.1-7 0.12.1-6 0.12.1-5 0.12.1-4 0.12.1-3 0.12.1-2 0.12.1-1

Kafka client for Elixir/Erlang.

HexDocs

Current section

Files

Jump to
Readme
130 Versions
3 Dependencies
Dependants
Files
Activity
kafka_ex_tc lib kafka_ex worker_pool.ex
Raw

lib/kafka_ex/worker_pool.ex 

defmodule KafkaEx.WorkerPool do

  @moduledoc false



  @overrun_warning 1000



  alias KafkaEx.Config

  alias KafkaEx.Utils.Logger

  import KafkaEx.Utils.Guards



  def start_link() do

    start_link(Application.get_env(:kafka_ex, :worker_pools, []))

  end



  def start_link([]), do: :ok



  def start_link([{wpool_name, workers} | worker_pools])

      when is_pos_integer(workers) do

    {:ok, worker_init} = KafkaEx.build_worker_options([])



    {:ok, _} =

      :wpool.start_pool(wpool_name,

        workers: workers,

        worker: {Config.server_impl(), [worker_init]},

        strategy: {:one_for_one, 1000, 3600},

        pool_sup_intensity: 1000,

        pool_sup_period: 3600,

        overrun_warning: @overrun_warning,

        overrun_handler: {__MODULE__, :producer_overrun_handler}

      )



    start_link(worker_pools)

  end



  def call(wpool_name, request) do

    :wpool.call(wpool_name, request)

  end



  def producer_overrun_handler(report) do

    try do

      {:call, {:produce, request}} = Keyword.get(report, :task)

      %{topic: topic, messages: messages} = request



      msg =

        Enum.join(

          [

            "Worker pool overrun for topic",

            topic,

            "length",

            length(messages)

          ],

          " "

        )



      Logger.warn(msg)

    rescue

      _ ->

        Logger.warn("Worker pool overrun")

    end

  end

end