Class: KubeMQ::Queues::QueueMessageReceived

Inherits:
Object
  • Object
show all
Defined in:
lib/kubemq/queues/queue_message_received.rb

Overview

A message received from a queue via KubeMQ::QueuesClient#poll or KubeMQ::QueuesClient#receive_queue_messages.

When received through the stream API (KubeMQ::QueuesClient#poll), transactional methods #ack, #reject, and #requeue are available for per-message acknowledgement within the poll transaction.

Examples:

Acknowledge a single message

response = client.poll(request)
response.messages.each do |msg|
  process(msg.body)
  msg.ack
end

Reject and requeue

response.messages.each do |msg|
  if valid?(msg)
    msg.ack
  else
    msg.requeue(channel: "tasks.retry")
  end
end

See Also:

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(id:, channel:, metadata:, body:, tags:, attributes: nil, action_proc: nil, sequence: 0) ⇒ QueueMessageReceived

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of QueueMessageReceived.

Parameters:

  • id (String) —

    message identifier

  • channel (String) —

    source channel name

  • metadata (String) —

    message metadata

  • body (String) —

    message payload

  • tags (Hash{String => String}, nil) —

    key-value tags

  • attributes (QueueMessageAttributes, nil) (defaults to: nil) —

    server-assigned attributes

  • action_proc (Proc, nil) (defaults to: nil) —

    internal callback for transaction actions

  • sequence (Integer) (defaults to: 0) —

    message sequence for transaction operations



103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 103

def initialize(id:, channel:, metadata:, body:, tags:, attributes: nil,
               action_proc: nil, sequence: 0)
  @id = id
  @channel = channel
  @metadata = metadata
  @body = body
  @tags = tags || {}
  @attributes = attributes
  @action_proc = action_proc
  @sequence = sequence
end

Instance Attribute Details

#attributes ⇒ Object (readonly)

Returns the value of attribute attributes.



92
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 92

attr_reader :id, :channel, :metadata, :body, :tags, :attributes

#body ⇒ String (readonly)

Returns message payload (binary).

Returns:

  • (String) —

    message payload (binary)



92
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 92

attr_reader :id, :channel, :metadata, :body, :tags, :attributes

#channel ⇒ String (readonly)

Returns source queue channel name.

Returns:

  • (String) —

    source queue channel name



92
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 92

attr_reader :id, :channel, :metadata, :body, :tags, :attributes

#id ⇒ String (readonly)

Returns message identifier.

Returns:

  • (String) —

    message identifier



92
93
94
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 92

def id
  @id
end

#metadata ⇒ String (readonly)

Returns message metadata.

Returns:

  • (String) —

    message metadata



92
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 92

attr_reader :id, :channel, :metadata, :body, :tags, :attributes

#tags ⇒ Hash{String => String} (readonly)

Returns user-defined key-value tags.

Returns:

  • (Hash{String => String}) —

    user-defined key-value tags



92
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 92

attr_reader :id, :channel, :metadata, :body, :tags, :attributes

Instance Method Details

#ack ⇒ void

This method returns an undefined value.

Acknowledges this message within the poll transaction.

Raises:



119
120
121
122
123
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 119

def ack
  raise TransactionError, 'No downstream receiver bound' unless @action_proc

  @action_proc.call(:AckRange, sequence_range: [@sequence])
end

#reject ⇒ void

This method returns an undefined value.

Rejects (negative-acknowledges) this message. The broker may redeliver it to another consumer or move it to a dead-letter queue.

Raises:



130
131
132
133
134
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 130

def reject
  raise TransactionError, 'No downstream receiver bound' unless @action_proc

  @action_proc.call(:NAckRange, sequence_range: [@sequence])
end

#requeue(channel:) ⇒ void

This method returns an undefined value.

Requeues this message to a different channel.

Parameters:

  • channel (String) —

    destination channel name

Raises:



141
142
143
144
145
 
# File 'lib/kubemq/queues/queue_message_received.rb', line 141

def requeue(channel:)
  raise TransactionError, 'No downstream receiver bound' unless @action_proc

  @action_proc.call(:ReQueueRange, channel: channel, sequence_range: [@sequence])
end