Class: TypedBus::Delivery

Inherits:
Object
  • Object
show all
Defined in:
lib/typed_bus/delivery.rb

Overview

Envelope wrapping a message delivered to a single subscriber. Each subscriber receives its own Delivery instance for the same message.

The subscriber must call ack! or nack! to resolve the delivery. If neither is called before the timeout expires, the delivery auto-nacks.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(message, channel_name:, subscriber_id:, timeout: nil, on_ack: nil, on_nack: nil) ⇒ Delivery

Returns a new instance of Delivery.

Parameters:

  • message (Object) —

    the published payload

  • channel_name (Symbol) —

    originating channel

  • subscriber_id (Integer) —

    identity of the target subscriber

  • timeout (Numeric) (defaults to: nil) —

    seconds before auto-nack (nil = no timeout)

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

    called with subscriber_id on ack

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

    called with subscriber_id on nack



18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/typed_bus/delivery.rb', line 18

def initialize(message, channel_name:, subscriber_id:, timeout: nil, on_ack: nil, on_nack: nil)
  @message       = message
  @channel_name  = channel_name
  @subscriber_id = subscriber_id
  @on_ack        = on_ack
  @on_nack       = on_nack
  @state         = :pending
  @timed_out     = false
  @timeout_task  = nil

  start_timeout(timeout) if timeout
end

Instance Attribute Details

#channel_name ⇒ Object (readonly)

Returns the value of attribute channel_name.



10
11
12
 
# File 'lib/typed_bus/delivery.rb', line 10

def channel_name
  @channel_name
end

#message ⇒ Object (readonly)

Returns the value of attribute message.



10
11
12
 
# File 'lib/typed_bus/delivery.rb', line 10

def message
  @message
end

#subscriber_id ⇒ Object (readonly)

Returns the value of attribute subscriber_id.



10
11
12
 
# File 'lib/typed_bus/delivery.rb', line 10

def subscriber_id
  @subscriber_id
end

Instance Method Details

#ack! ⇒ Object 



31
32
33
34
35
36
 
# File 'lib/typed_bus/delivery.rb', line 31

def ack!
  transition!(:acked)
  cancel_timeout
  log(:info, "acked")
  @on_ack&.call(@subscriber_id)
end

#acked? ⇒ Boolean

Returns:

  • (Boolean) 


45
 
# File 'lib/typed_bus/delivery.rb', line 45

def acked?    = @state == :acked

#cancel_timeout ⇒ Object

Cancel the timeout timer without resolving.



51
52
53
54
 
# File 'lib/typed_bus/delivery.rb', line 51

def cancel_timeout
  @timeout_task&.stop
  @timeout_task = nil
end

#nack! ⇒ Object 



38
39
40
41
42
43
 
# File 'lib/typed_bus/delivery.rb', line 38

def nack!
  transition!(:nacked)
  cancel_timeout
  log(:warn, "nacked")
  @on_nack&.call(@subscriber_id)
end

#nacked? ⇒ Boolean

Returns:

  • (Boolean) 


46
 
# File 'lib/typed_bus/delivery.rb', line 46

def nacked?   = @state == :nacked

#pending? ⇒ Boolean

Returns:

  • (Boolean) 


47
 
# File 'lib/typed_bus/delivery.rb', line 47

def pending?  = @state == :pending

#timed_out? ⇒ Boolean

Returns:

  • (Boolean) 


48
 
# File 'lib/typed_bus/delivery.rb', line 48

def timed_out? = @timed_out