Class: Flu::PendingPublications

Inherits:
Object
  • Object
show all
Defined in:
lib/flu-rails/pending_publications.rb

Overview

The events whose publication failed, kept for another attempt.

A broker that closes a connection takes a few seconds to be usable again, since Bunny reopens it in the background, and an event published in that window has nowhere to go. Rather than being given up on there, it waits here for the connection to be back: the next transaction to commit on the thread comes back for it, and so does the end of the request or the job it belongs to.

In memory, and per thread: what is waiting here dies with the process. An application that cannot afford to lose an event keeps it itself, from 'on_publication_failure', which is called for every event this gives up on.

Constant Summary collapse

MAX_ATTEMPTS = 
3

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ PendingPublications

Returns a new instance of PendingPublications.



21
22
23
 
# File 'lib/flu-rails/pending_publications.rb', line 21

def initialize
  @entries = []
end

Class Method Details

.current ⇒ Object 



17
18
19
 
# File 'lib/flu-rails/pending_publications.rb', line 17

def self.current
  Thread.current[:flu_pending_publications] ||= new
end

Instance Method Details

#drain ⇒ Object

Publishes again what its publisher can reach again, keeps what it cannot, and gives up on what has been refused MAX_ATTEMPTS times, an event the broker itself rejects being no more publishable on the tenth attempt than on the first.



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/flu-rails/pending_publications.rb', line 37

def drain
  return if @entries.empty?

  kept = []
  @entries.each do |entry|
    next kept.push(entry) unless reachable?(entry[:publisher])

    begin
      entry[:publisher].publish(entry[:event])
    rescue StandardError => error
      entry[:error]     = error
      entry[:attempts] += 1
      entry[:attempts] < MAX_ATTEMPTS ? kept.push(entry) : give_up(entry)
    end
  end
  @entries = kept
end

#push(event, publisher, error) ⇒ Object 



29
30
31
32
 
# File 'lib/flu-rails/pending_publications.rb', line 29

def push(event, publisher, error)
  give_up(@entries.shift) while @entries.size >= Flu.config.max_pending_events
  @entries.push({ event: event, publisher: publisher, error: error, attempts: 1 })
end

#size ⇒ Object 



25
26
27
 
# File 'lib/flu-rails/pending_publications.rb', line 25

def size
  @entries.size
end