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

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

Overview

Background drain thread. Wakes every interval seconds and tries Buffered.drain! against a fresh Wurk::Client. drain! already short-circuits on the first transient failure, so a still-down Redis just leaves the buffer 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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(interval: DEFAULT_INTERVAL, client_factory: -> { Wurk::Client.new }) ⇒ Drainer

Returns a new instance of Drainer.



355
356
357
358
359
360
361
362
363
364
365
366
 
# File 'lib/wurk/client/buffered.rb', line 355

def initialize(interval: DEFAULT_INTERVAL, client_factory: -> { Wurk::Client.new })
  unless interval.is_a?(Numeric) && interval.positive?
    raise ArgumentError, 'interval must be a positive Numeric'
  end

  @interval = interval
  @client_factory = client_factory
  @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.



353
354
355
 
# File 'lib/wurk/client/buffered.rb', line 353

def interval
  @interval
end

Instance Method Details

#running? ⇒ Boolean

Returns:

  • (Boolean) 


389
390
391
 
# File 'lib/wurk/client/buffered.rb', line 389

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

#start ⇒ Object 



368
369
370
371
372
373
374
375
376
377
378
 
# File 'lib/wurk/client/buffered.rb', line 368

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 



380
381
382
383
384
385
386
387
 
# File 'lib/wurk/client/buffered.rb', line 380

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