Broadway v0.4.0 Broadway.Message View Source

A struct that holds all information about a message.

A message is first created by the producers. Once created, the message is sent downstream and gets updated multiple times, either by the module implementing the Broadway behaviour through the Broadway.handle_message/3 callback or internaly by one of the built-in stages of Broadway.

In order to manipulate a message, you should use one of the imported functions provided by this module.

Link to this section Summary

Types

t()

Functions

failed(message, reason)

Mark a message as failed.

put_batch_key(message, batch_key)

Defines the batch key within a batcher for the message.

put_batch_mode(message, mode)

Sets the batching mode for the message.

put_batcher(message, batcher)

Defines the target batcher which the message should be forwarded to.

update_data(message, fun)

Updates the data from a message.

Link to this section Types

t()

t() :: %Broadway.Message{
  acknowledger: {module(), ack_ref :: term(), data :: term()},
  batch_key: term(),
  batch_mode: :bulk | :flush,
  batcher: atom(),
  data: term(),
  metadata: %{optional(:atom) => term()},
  status: :ok | {:failed, reason :: binary()}
}

Link to this section Functions

failed(message, reason)

failed(message :: Broadway.Message.t(), reason :: term()) ::
  Broadway.Message.t()

Mark a message as failed.

Failed messages are sent directly to the related acknowledger so they're not forwarded to the next step in the pipeline.

put_batch_key(message, batch_key)

put_batch_key(message :: Broadway.Message.t(), batch_key :: term()) ::
  Broadway.Message.t()

Defines the batch key within a batcher for the message.

Inside each batcher, we attempt to build batch_size within batch_timeout for each batch_key.

put_batch_mode(message, mode)

put_batch_mode(message :: Broadway.Message.t(), mode :: :bulk | :flush) ::
  Broadway.Message.t()

Sets the batching mode for the message.

When the mode is :bulk, the batch that the message is in is delivered after the batch size or the batch timeout is reached.

When the mode is :flush, the batch that the message is in is delivered immediately after processing. Note it doesn't mean the batch contains only a single element but rather that all messages receives from the processor are delivered without waiting.

The default mode for messages is :bulk.

put_batcher(message, batcher)

put_batcher(message :: Broadway.Message.t(), batcher :: atom()) ::
  Broadway.Message.t()

Defines the target batcher which the message should be forwarded to.

update_data(message, fun)

update_data(message :: Broadway.Message.t(), fun :: (term() -> term())) ::
  Broadway.Message.t()

Updates the data from a message.

This function is usually used inside the handle_message/3 implementation in order to replace the data with the new processed data.