I’ve only read articles about Broadway, never used it in production, so I can’t speak to the specific functionality here. Some more general code notes:
def handle_message(_, %Message{data: string_payload} = message, _) do
apply(unquote(module), :handle_message, [string_payload])
message
# I rescue any crash and move the message to the other queue
rescue
e ->
Beware: there are more ways to crash than rescue
deals with: see GenServer timeout and retry for recent discussion
exchange_fct = Keyword.get(opts, :exchange)
exchange = exchange_fct.()
This pattern happens often enough that it feels like it should be named.