Pulsar.Message (Pulsar v3.0.0)

Represents a message received from a Pulsar topic.

This struct encapsulates all information about a message delivered to a consumer callback.

Fields

payload - The message payload as a binary. For a chunked message, the assembled complete payload.
message_id - What to pass to Pulsar.Consumer.ack/2 and Pulsar.Consumer.nack/2. Treat it as opaque rather than pattern matching it: it carries a batch index for a batched message, and for a chunked one it stands for every chunk, so it is a list there.
chunk_metadata - Metadata about chunked messages (nil for non-chunked messages). For complete chunked messages: %{chunked: true, complete: true, uuid: "...", num_chunks: N} For incomplete chunked messages: %{chunked: true, complete: false, error: :reason, uuid: "..."}
validation_error - nil for messages that arrived intact. Otherwise why the frame could not be trusted, in which case payload holds unverified bytes. See valid?/1.
raw - The underlying protocol structs, as a map of :command, :metadata, :single_metadata and :broker_metadata. Unstable: its shape follows the wire protocol and changes with how the broker delivered the message. Use the accessors below instead; reach for raw only for protocol details they do not cover.

Reading a message

Everything a callback normally needs has an accessor that answers the same way whether the message arrived on its own, inside a batch, or split across chunks:

Accessor	Returns
`producer_name/1`	The producer that published it
`publish_time/1`	Broker publish timestamp, in milliseconds
`event_time/1`	Application-set event time, or `nil` when unset
`key/1`	The partition key, or `nil`
`ordering_key/1`	The ordering key, or `nil`
`properties/1`	User properties as a map
`redelivery_count/1`	How many times the broker has redelivered it
`message_id_string/1`	Its id as Pulsar prints it, for logging and correlation

This matters because the same datum lives in different places depending on delivery: a batched message carries its key and properties per message, a non-batched one carries them in the message metadata, and a chunked one has a list of both. The accessors resolve that; reading raw does not.

Usage

def handle_message(%Pulsar.Message{payload: payload}, state) do
  process(payload)
  {:ok, state}
end

def handle_message(%Pulsar.Message{} = message, state) do
  Logger.info("from #{Pulsar.Message.producer_name(message)}, key #{Pulsar.Message.key(message)}")
  {:ok, state}
end

Manual acknowledgement, where the id is captured for use after the callback returns:

def handle_message(%Pulsar.Message{message_id: message_id} = message, state) do
  consumer = self()

  spawn(fn ->
    case process_async(message) do
      :ok -> Pulsar.Consumer.ack(consumer, message_id)
      {:error, _reason} -> Pulsar.Consumer.nack(consumer, message_id)
    end
  end)

  {:noreply, state}
end

Summary

Types

raw()

The protocol structs a message was built from. Unstable; see the :raw field.

t()

Functions

chunked?(message)

Returns true if the message is a chunked message, false otherwise.

complete?(message)

Returns true if the chunked message is complete, false otherwise.

event_time(message)

Returns the event time the publishing application set, or nil when it set none.

key(message)

Returns the message's partition key, or nil when it has none.

message_id_string(message)

Returns the message's id as Pulsar prints it, or nil when it has none.

num_broker_messages(message)

Returns the number of broker messages (permits) consumed.

ordering_key(message)

Returns the message's ordering key, or nil when it has none.

producer_name(message)

Returns the name of the producer that published the message.

properties(message)

Returns the user properties published with the message, as a map.

publish_time(message)

Returns the time the broker published the message, in milliseconds since the epoch.

redelivery_count(message)

Returns the maximum redelivery count across all commands.

valid?(message)

Returns true if the message arrived intact, false if it did not.

Types

raw()

@type raw() :: %{
  command: struct() | [struct()],
  metadata: struct() | [struct()] | nil,
  single_metadata: struct() | [struct()] | nil,
  broker_metadata: term() | [term()]
}

The protocol structs a message was built from. Unstable; see the :raw field.

t()

@type t() :: %Pulsar.Message{
  chunk_metadata: map() | nil,
  message_id: term() | [term()],
  payload: binary(),
  raw: raw() | nil,
  validation_error: atom() | nil
}

Functions

chunked?(message)

@spec chunked?(t()) :: boolean()

Returns true if the message is a chunked message, false otherwise.

This checks for the presence of chunk metadata.

Examples

iex> Pulsar.Message.chunked?(%Pulsar.Message{chunk_metadata: %{chunked: true}})
true

iex> Pulsar.Message.chunked?(%Pulsar.Message{payload: "one"})
false

complete?(message)

@spec complete?(t()) :: boolean()

Returns true if the chunked message is complete, false otherwise.

For non-chunked messages, always returns true since they are inherently complete. For chunked messages, returns true only if all chunks were successfully received.

Examples

iex> Pulsar.Message.complete?(%Pulsar.Message{chunk_metadata: %{chunked: true, complete: true}})
true

iex> Pulsar.Message.complete?(%Pulsar.Message{chunk_metadata: %{chunked: true, complete: false}})
false

iex> Pulsar.Message.complete?(%Pulsar.Message{payload: "one"})
true

event_time(message)

@spec event_time(t()) :: non_neg_integer() | nil

Returns the event time the publishing application set, or nil when it set none.

Pulsar represents an unset event time as 0, which this reports as nil.

Examples

iex> raw = %{metadata: %{event_time: 0}}
iex> Pulsar.Message.event_time(%Pulsar.Message{raw: raw})
nil

key(message)

@spec key(t()) :: String.t() | nil

Returns the message's partition key, or nil when it has none.

A batched message carries its own key, so this reads that one and falls back to the key on the entry it arrived in.

Examples

iex> raw = %{metadata: %{partition_key: "entry"}, single_metadata: %{partition_key: "message"}}
iex> Pulsar.Message.key(%Pulsar.Message{raw: raw})
"message"

message_id_string(message)

@spec message_id_string(t()) :: String.t() | nil

Returns the message's id as Pulsar prints it, or nil when it has none.

The shape is ledgerId:entryId:partition, with a batched message's index within its entry appended, which is what the Java client's MessageId.toString() produces. It is what to log or carry when a message has to be correlated with one seen elsewhere; message_id itself stays opaque.

A chunked message answers for the chunk it began at.

Examples

iex> id = %{ledgerId: 7, entryId: 42, partition: -1, batch_index: -1}
iex> Pulsar.Message.message_id_string(%Pulsar.Message{message_id: id})
"7:42:-1"

A batched message appends its index, so two messages of one batch stay distinguishable:

iex> id = %{ledgerId: 7, entryId: 42, partition: 3, batch_index: 1}
iex> Pulsar.Message.message_id_string(%Pulsar.Message{message_id: id})
"7:42:3:1"

num_broker_messages(message)

@spec num_broker_messages(t()) :: pos_integer()

Returns the number of broker messages (permits) consumed.

For non-chunked messages, this is always 1. For chunked messages, this is the number of chunks actually received.

This is used for flow control permit accounting.

Examples

iex> Pulsar.Message.num_broker_messages(%Pulsar.Message{payload: "one"})
1

iex> three_chunks = %{chunked: true, complete: true, message_ids: [1, 2, 3]}
iex> Pulsar.Message.num_broker_messages(%Pulsar.Message{chunk_metadata: three_chunks})
3

Two chunks of three, given up on, still cost the two permits the broker charged:

iex> expired = %{chunked: true, complete: false, message_ids: [1, 2]}
iex> Pulsar.Message.num_broker_messages(%Pulsar.Message{chunk_metadata: expired})
2

ordering_key(message)

@spec ordering_key(t()) :: binary() | nil

Returns the message's ordering key, or nil when it has none.

producer_name(message)

@spec producer_name(t()) :: String.t() | nil

Returns the name of the producer that published the message.

Examples

iex> raw = %{metadata: %{producer_name: "orders-api"}}
iex> Pulsar.Message.producer_name(%Pulsar.Message{raw: raw})
"orders-api"

properties(message)

@spec properties(t()) :: %{required(String.t()) => String.t()}

Returns the user properties published with the message, as a map.

Examples

iex> raw = %{metadata: %{properties: [%{key: "trace-id", value: "abc"}]}}
iex> Pulsar.Message.properties(%Pulsar.Message{raw: raw})
%{"trace-id" => "abc"}

iex> Pulsar.Message.properties(%Pulsar.Message{raw: %{metadata: nil}})
%{}

publish_time(message)

@spec publish_time(t()) :: non_neg_integer() | nil

Returns the time the broker published the message, in milliseconds since the epoch.

redelivery_count(message)

@spec redelivery_count(t()) :: non_neg_integer()

Returns the maximum redelivery count across all commands.

For chunked messages, returns the maximum redelivery count from all chunks. For non-chunked messages, returns the redelivery count from the single command.

Examples

iex> Pulsar.Message.redelivery_count(%Pulsar.Message{raw: %{command: %{redelivery_count: 3}}})
3

iex> chunks = [%{redelivery_count: 1}, %{redelivery_count: 3}]
iex> Pulsar.Message.redelivery_count(%Pulsar.Message{raw: %{command: chunks}})
3

valid?(message)

@spec valid?(t()) :: boolean()

Returns true if the message arrived intact, false if it did not.

An invalid message failed its CRC32C check or carried metadata that could not be read, so its metadata is nil and its payload is unverified: the bytes the framing points at, or the whole message section when even that does not hold. It is delivered so the callback can record or divert it, but the payload must not be treated as data. validation_error says what went wrong.

Messages that fail validation are routed to Pulsar.Consumer.Callback.handle_invalid_message/2, so handle_message/2 never receives one and rarely needs this check.

Examples

iex> Pulsar.Message.valid?(%Pulsar.Message{payload: "hello"})
true

iex> Pulsar.Message.valid?(%Pulsar.Message{validation_error: :checksum_mismatch})
false

Validity is independent of chunk completeness — an incomplete chunked message is still made of bytes that arrived intact:

iex> expired = %Pulsar.Message{chunk_metadata: %{chunked: true, complete: false}}
iex> {Pulsar.Message.valid?(expired), Pulsar.Message.complete?(expired)}
{true, false}

def handle_invalid_message(%Pulsar.Message{} = message, state) do
  Logger.error("dropping corrupt message: #{message.validation_error}")
  {:ok, state}
end