herald v0.1.0 Herald.Pipeline

Pipeline is where messages are processed.

All message processing is started by function run/2.

When run/2 receives a message, it runs the following steps:

pre_processing - Will convert the message to a struct, using the schema defined in route for the given queue;
processor_caler - Will call the processor defined in router for the given queue.

Link to this section Summary

t()

Indicates to upper layers what must do doing with message.

Process a given message.

t() :: %Herald.Pipeline{
  message: map(),
  perform: to_perform(),
  processor: (... -> any()),
  result: any(),
  schema: atom()
}

to_perform() :: :ack | :delete | :requeue

Indicates to upper layers what must do doing with message.

Possible values:

:ack - When the message is sucessfully processed;
:delete - When message must be deleted from broker after a processing error;
:requeue - When message must be reprocessed in future after a processing error.

run(String.t(), String.t()) :: :ok | {:error, atom()}

Process a given message.

Receive the following arguments:

queue - The queue where message was received from;
message - The received message in a raw state (i.e. the String as received from broker)

This function must be called by plugins integrating Herald with Brokers.

Warning: When you call this function without configure a router for you application, it will raises the exception Herald.Errors.MissingRouter