Galena.Consumer is a customized GenStage producer which is able to receive some messages from some producers or producers-consumers. The consumer will have the possibility to be subscribed to the chosen topics from the chosen producers.

Definition

  defmodule MyConsumer do
    use Galena.Consumer

    def handle_consume(topic, message) do
      IO.puts(topic <> ": " <> message)
    end
  end

Start up

Define the args of your Consumer. It has to be a Keyword list which has to contain a producers_info field which will have a list of tuples of two parameters, where the first one will be a list of topics and the second one the producer or producer-consumer:

  args = [
    producers_info: [
      {["topic_1", "topic_2", "topic_3"], :producer1},
      {["topic_A"], :producer2},
      {["topic_a", "topic_b"], :producer3},
      {[], :producer4}
    ]
  ]

When the list of topics is empty, your consumer will receive all the information published by the producer.

{:ok, consumer} = MyConsumer.start_link(args, [name: :consumer])