Packages
kafka_message_bus
2.2.4
4.3.4
4.3.3
4.3.2
4.3.1
4.3.0
4.2.10
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.1
4.1.0
4.0.1
4.0.0
4.0.0-rc.8
4.0.0-rc.7
4.0.0-rc.6
4.0.0-rc.5
4.0.0-rc.4
4.0.0-rc.3
4.0.0-rc.2
3.0.0
2.2.4
2.2.3
2.2.2
2.2.1
2.2.0
2.1.0
2.0.2
2.0.1
2.0.0
1.0.1
1.0.0
0.2.3
0.2.2
0.2.1
0.2.0
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
A general purpose messaging utility library supporting Exq and Kaffe
Current section
Files
Jump to
Current section
Files
lib/kafka_message_bus/consumer.ex
defmodule KafkaMessageBus.Consumer do
require Logger
@processor_config Application.get_env(:kafka_message_bus, :consumers)
@retry_strategy Application.get_env(:kafka_message_bus, :retry_strategy)
def handle_messages(nil), do: :ok
def handle_messages(messages) do
for message <- messages do
Logger.debug fn ->
"Got message: #{message.topic}/#{message.partition} -> #{message.key}: #{message.value}"
end
message.value
|> Poison.decode()
|> process_message(message.topic)
end
:ok
end
defp process_message({:ok, msg_content}, topic) do
Logger.metadata(request_id: msg_content["request_id"])
Enum.each(@processor_config, &execute_message(msg_content, topic, &1))
end
defp process_message({:error, error}, topic),
do: Logger.error("Failed to parse message in topic #{topic}. Error: #{inspect error}")
defp execute_message(msg_content = %{"resource" => resource}, topic, {topic, resource, message_processor}) do
Logger.debug fn -> "[ACCEPTED] #{message_processor}: #{inspect msg_content}" end
try do
message_processor.process(msg_content)
rescue
_ -> enqueue_message_retry(msg_content, message_processor, @retry_strategy)
end
end
defp execute_message(msg_content, _, {_, _, message_processor}), do:
Logger.debug fn -> "[IGNORED] #{message_processor}: #{inspect msg_content}" end
:ok
defp enqueue_message_retry(msg_content, message_processor, retry_strategy)
when retry_strategy == :exq,
do: KafkaMessageBus.ConsumerEnqueuer.enqueue(msg_content, message_processor)
defp enqueue_message_retry(_msg_content, _message_processor, _retry_strategy),
do: Logger.warn("Will not retry message")
end