Class: Wurk::Client::Buffered::Circuit

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/client/buffered.rb

Overview

Which Redis servers are refusing connections, by pool identity. An entry exists only while its circuit is open, so a rebuilt pool leaves nothing behind.

Constant Summary collapse

OPEN_SECONDS =

A circuit closes itself after this long, so a dead drainer costs a caller one attempt per window instead of buffering forever. A failed probe renews it.

30.0

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Circuit

Returns a new instance of Circuit.



101
102
103
104
 
# File 'lib/wurk/client/buffered.rb', line 101

def initialize
  @lock = Mutex.new
  @open_until = {}.compare_by_identity
end

Instance Method Details

#any_open? ⇒ Boolean

Returns:

  • (Boolean) 


127
128
129
 
# File 'lib/wurk/client/buffered.rb', line 127

def any_open?
  !@open_until.empty?
end

#close(pool) ⇒ Object 



123
124
125
 
# File 'lib/wurk/client/buffered.rb', line 123

def close(pool)
  @lock.synchronize { @open_until.delete(pool) }
end

#open?(pool) ⇒ Boolean

Returns:

  • (Boolean) 


106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/wurk/client/buffered.rb', line 106

def open?(pool)
  return false if @open_until.empty?

  @lock.synchronize do
    deadline = @open_until[pool]
    next false unless deadline
    next true if monotonic < deadline

    @open_until.delete(pool)
    false
  end
end

#ping_all(except:) ⇒ Object

For a circuit with nothing buffered behind it (a batched push tripped it, or :raise overflow kept the buffer empty): no replay exists to prove the server is back. Any answer means reachable, and an error such as LOADING belongs to the next real push to report, not to the circuit.



135
136
137
 
# File 'lib/wurk/client/buffered.rb', line 135

def ping_all(except:)
  @lock.synchronize { @open_until.keys }.each { |pool| ping(pool) unless except.any? { |o| o.equal?(pool) } }
end

#trip(pool) ⇒ Object 



119
120
121
 
# File 'lib/wurk/client/buffered.rb', line 119

def trip(pool)
  @lock.synchronize { @open_until[pool] = monotonic + OPEN_SECONDS }
end