Class: AceMQ::AMQP::Ack

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/ack.rb

Overview

What a handler says about a message: it worked, try it again, never again, or put it somewhere a person will look.

Returned rather than performed, so a handler that forgets to decide is caught immediately rather than leaving a message unacknowledged until the connection drops — after which it comes back, usually to the same handler, with the same outcome.

Constant Summary collapse

ACCEPT = 
:accept
RETRY = 
:retry
REJECT = 
:reject
PARK = 
:park
IN_PROGRESS = 
:in_progress
DEFAULT_IN_PROGRESS_DELAY =

Seconds a message found in progress waits before it is looked at again.

5.0

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(action, error = nil, delay: nil) ⇒ Ack

Returns a new instance of Ack.



42
43
44
45
46
47
 
# File 'lib/acemq/amqp/ack.rb', line 42

def initialize(action, error = nil, delay: nil)
  @action = action
  @error = error
  @delay = delay
  freeze
end

Instance Attribute Details

#action ⇒ Object (readonly)

Returns the value of attribute action.



36
37
38
 
# File 'lib/acemq/amqp/ack.rb', line 36

def action
  @action
end

#delay ⇒ Float? (readonly)

Returns seconds to wait before the message is looked at again; read for IN_PROGRESS only, a retry's wait is the policy's.

Returns:

  • (Float, nil) —

    seconds to wait before the message is looked at again; read for IN_PROGRESS only, a retry's wait is the policy's



40
41
42
 
# File 'lib/acemq/amqp/ack.rb', line 40

def delay
  @delay
end

#error ⇒ Object (readonly)

Returns the value of attribute error.



36
37
38
 
# File 'lib/acemq/amqp/ack.rb', line 36

def error
  @error
end

Class Method Details

.accept ⇒ Object

Confirms the message. It will not be delivered again.



50
51
52
 
# File 'lib/acemq/amqp/ack.rb', line 50

def self.accept
  new(ACCEPT)
end

.in_progress(delay = DEFAULT_IN_PROGRESS_DELAY) ⇒ Object

Puts the message back because somebody else is still working on it.

What Patterns.idempotent answers when a redelivery finds a claim that is live but not yet confirmed. Accepting it would lose the message if that other handler then died; running it would do the work twice. So it goes back on its own queue after delay seconds with its attempt unchanged: a claim held elsewhere says nothing about whether this message can be handled, so it neither spends a retry nor can exhaust the policy and be dead-lettered. When the claim is confirmed the next look is a duplicate and is accepted; when its lease runs out, it is taken over.

The wait is spent in the consumer, holding the delivery, as a short retry's is; keep it short beside the store's claim timeout.

Parameters:

  • delay (Numeric) (defaults to: DEFAULT_IN_PROGRESS_DELAY) —

    seconds until the message is looked at again



110
111
112
 
# File 'lib/acemq/amqp/ack.rb', line 110

def self.in_progress(delay = DEFAULT_IN_PROGRESS_DELAY)
  new(IN_PROGRESS, delay: delay.to_f)
end

.park(reason = nil) ⇒ Object

Sends the message to {queue}.parked without trying again.

Parked and not dead-lettered, because they are different problems and the two queues exist to keep them apart: the dead-letter queue holds messages that were tried and failed, and the parking queue holds messages nothing could make sense of. A handler that already knows a message is unreadable — a field that is not a date where a date has to be, a version this service was never taught — used to have to reject it into the dead letters and lose that distinction. Somebody draining the dead-letter queue after an outage should not have to sort out the messages that were never going to work from the ones that failed while the database was down.

The engine parks a message it could not decode by itself; this is the same destination, asked for by a handler that got further.

Parameters:

  • reason (String, Exception, nil) (defaults to: nil) —

    why it cannot be read



90
91
92
 
# File 'lib/acemq/amqp/ack.rb', line 90

def self.park(reason = nil)
  new(PARK, reason)
end

.reject(error = nil) ⇒ Object

Dead-letters the message without trying again.

For when the message itself is the problem — a field that cannot be missing is missing — rather than when the world is temporarily unhelpful.



69
70
71
 
# File 'lib/acemq/amqp/ack.rb', line 69

def self.reject(error = nil)
  new(REJECT, error)
end

.retry(error = nil) ⇒ Object

Returns the message to be tried again.

The retry policy decides whether there is another attempt left; when there is not, the message is dead-lettered with error as the reason. A FatalError skips the remaining attempts, because they would all fail the same way.



60
61
62
 
# File 'lib/acemq/amqp/ack.rb', line 60

def self.retry(error = nil)
  new(RETRY, error)
end

Instance Method Details

#==(other) ⇒ Object 



122
123
124
 
# File 'lib/acemq/amqp/ack.rb', line 122

def ==(other)
  other.is_a?(Ack) && action == other.action
end

#accept? ⇒ Boolean

Returns:

  • (Boolean) 


114
 
# File 'lib/acemq/amqp/ack.rb', line 114

def accept? = action == ACCEPT

#in_progress? ⇒ Boolean

Returns:

  • (Boolean) 


118
 
# File 'lib/acemq/amqp/ack.rb', line 118

def in_progress? = action == IN_PROGRESS

#park? ⇒ Boolean

Returns:

  • (Boolean) 


117
 
# File 'lib/acemq/amqp/ack.rb', line 117

def park? = action == PARK

#reject? ⇒ Boolean

Returns:

  • (Boolean) 


116
 
# File 'lib/acemq/amqp/ack.rb', line 116

def reject? = action == REJECT

#retry? ⇒ Boolean

Returns:

  • (Boolean) 


115
 
# File 'lib/acemq/amqp/ack.rb', line 115

def retry? = action == RETRY

#to_s ⇒ Object 



120
 
# File 'lib/acemq/amqp/ack.rb', line 120

def to_s = action.to_s