misha_cafex

defmodule Cafex do

  @type server :: {host :: String.t, port :: 0..65535}
  @type broker :: server
  @type client_id :: String.t

  def start_topic(name, brokers, opts \\ []) do
    Cafex.Supervisor.start_topic(name, brokers, opts)
  end

  @doc """
  Start a producer.

  Read `Cafex.Producer` for more details.
  """
  @spec start_producer(topic_name :: String.t, opts :: Cafex.Producer.options) :: Supervisor.on_start_child
  def start_producer(topic_name, opts \\ []) do
    Cafex.Supervisor.start_producer(topic_name, opts)
  end
  defdelegate stop_producer(sup), to: Cafex.Supervisor

  @doc """
  Produce message to kafka server in the synchronous way.

  See `Cafex.Producer.produce/3`
  """
  def produce(producer, value, opts \\ []) do
    Cafex.Producer.produce(producer, value, opts)
  end

  @doc """
  Produce message to kafka server in the asynchronous way.

  See `Cafex.Producer.produce/3`
  """
  def async_produce(producer, value, opts \\ []) do
    Cafex.Producer.async_produce(producer, value, opts)
  end

  def fetch(topic_pid, partition, offset) when is_integer(partition)
                                           and is_integer(offset) do
    Cafex.Topic.Server.fetch topic_pid, partition, offset
  end

  @doc """
  Start a consumer.

  Read `Cafex.Consumer.Manager` for more details.
  """
  @spec start_consumer(name :: atom, Cafex.Consumer.Manager.options) :: Supervisor.on_start_child
  def start_consumer(name, opts \\ []) do
    Cafex.Supervisor.start_consumer(name, opts)
  end
  defdelegate stop_consumer(name), to: Cafex.Supervisor
end