Class: Wurk::Client::Buffered::Drainer

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

Overview

Background drain thread and the circuit's only prober. Started on the first failed push if the app did not start one; wakes every interval seconds and runs Buffered.drain_all!, which costs no Redis round trip while the buffer is empty and no circuit is open. drain! already short-circuits on the first transient failure, so a still-down Redis just leaves its entries alone for this tick — no exponential backoff or explicit "reconnect detection" needed; the inner connection retry already lives inside client.raw_push.

Constant Summary collapse

DEFAULT_INTERVAL = 
2.0
STOP_JOIN_TIMEOUT = 
5.0
PROBE_PAUSE =

While a circuit is open, pushes are only in memory, so finding out that Redis is back quickly matters more than the interval an idle drainer keeps.

0.25

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(interval: DEFAULT_INTERVAL) ⇒ Drainer

Returns a new instance of Drainer.



767
768
769
770
771
772
773
774
775
776
777
 
# File 'lib/wurk/client/buffered.rb', line 767

def initialize(interval: DEFAULT_INTERVAL)
  unless interval.is_a?(Numeric) && interval.positive?
    raise ArgumentError, 'interval must be a positive Numeric'
  end

  @interval = interval
  @done = false
  @thread = nil
  @wake = ConditionVariable.new
  @lock = Mutex.new
end

Instance Attribute Details

#interval ⇒ Object (readonly)

Read by Buffered.reset_after_fork! off the inherited drainer, to rebuild an equivalent one in the child without touching its lock.



765
766
767
 
# File 'lib/wurk/client/buffered.rb', line 765

def interval
  @interval
end

Instance Method Details

#running? ⇒ Boolean

Returns:

  • (Boolean) 


800
801
802
 
# File 'lib/wurk/client/buffered.rb', line 800

def running?
  @thread&.alive? == true
end

#start ⇒ Object 



779
780
781
782
783
784
785
786
787
788
789
 
# File 'lib/wurk/client/buffered.rb', line 779

def start
  @lock.synchronize do
    return if @thread&.alive?

    @done = false
    @thread = Thread.new do
      Thread.current.name = 'wurk-reliable_push-drainer'
      run
    end
  end
end

#stop ⇒ Object 



791
792
793
794
795
796
797
798
 
# File 'lib/wurk/client/buffered.rb', line 791

def stop
  @lock.synchronize do
    @done = true
    @wake.broadcast
  end
  @thread&.join(STOP_JOIN_TIMEOUT)
  @thread = nil
end