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.



348
349
350
351
352
353
354
355
356
357
358
359
 
# File 'lib/wurk/client/buffered.rb', line 348

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.



346
347
348
 
# File 'lib/wurk/client/buffered.rb', line 346

def interval
  @interval
end

Instance Method Details

#running? ⇒ Boolean

Returns:

  • (Boolean) 


382
383
384
 
# File 'lib/wurk/client/buffered.rb', line 382

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

#start ⇒ Object 



361
362
363
364
365
366
367
368
369
370
371
 
# File 'lib/wurk/client/buffered.rb', line 361

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 



373
374
375
376
377
378
379
380
 
# File 'lib/wurk/client/buffered.rb', line 373

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