Class: CopyTunerClient::Poller

Inherits:
Object
  • Object
show all
Defined in:
lib/copy_tuner_client/poller.rb

Overview

Starts a background thread that continually resynchronizes with the remote server using the given Cache after a set delay.

Instance Method Summary collapse

Constructor Details

#initialize(cache, options) ⇒ Poller

Returns a new instance of Poller.

Parameters:

  • options (Hash)

Options Hash (options):

  • :logger (Logger) —

    where errors should be logged

  • :polling_delay (Fixnum) —

    how long to wait in between requests



11
12
13
14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/copy_tuner_client/poller.rb', line 11

def initialize(cache, options)
  @cache          = cache
  @polling_delay  = options[:polling_delay]
  @logger         = options[:logger]
  @command_queue  = CopyTunerClient::QueueWithTimeout.new
  @mutex          = Mutex.new
  @thread         = nil
  @last_synced_at = nil
  # スレッドの生死とは別に、ライフサイクルの意図を持つ。
  # @running: ポーリングを継続する意図があるか（start で true / stop で false）
  # @aborted: 張り直しても同じ理由で死ぬと分かっている終わり方をしたか
  @running        = false
  @aborted        = false
end

Instance Method Details

#start ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/copy_tuner_client/poller.rb', line 26

def start
  @mutex.synchronize do
    @running = true
    # 回復の見込みがない理由で終了しているなら張り直さない。fork のたびに同じ例外で
    # 死ぬスレッドを作り直してログを埋めるだけになる
    next if @aborted
    # fork 後の子は親から dead な Thread オブジェクトを継承するため、nil かどうかだけでは
    # 「動いていない」を判定できない。死んでいるスレッドは張り直す
    next if @thread&.alive?

    # コマンドキューは世代ごとに作り直し、スレッドに自分のキューを渡す。前の世代宛に
    # 積まれたまま未消費で残った :stop を次の世代が 1 周目で拾って自殺するのを、
    # 「1 つのキューは 1 本のスレッドだけのもの」という不変条件で構造的に防ぐ
    queue = CopyTunerClient::QueueWithTimeout.new
    @command_queue = queue
    @logger.info 'start poller thread'
    @thread = Thread.new { poll(queue) } or logger.error("Couldn't start poller thread")
  end
end

#start_sync ⇒ Object 



69
70
71
 
# File 'lib/copy_tuner_client/poller.rb', line 69

def start_sync
  @command_queue.uniq_push(:sync)
end

#stop ⇒ Boolean

戻り値は ForkHook が「fork 後に張り直すか」を決めるのに使う。スレッドの生死ではなく ポーリングを継続する意図があったかを返す。想定外の例外で死んだだけのスレッドは fork 後に張り直したいが、生死で判定すると張り直せなくなる

Returns:

  • (Boolean) —

    ポーリング継続の意図があった（＝ fork 後に張り直すべき）なら true



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/copy_tuner_client/poller.rb', line 51

def stop
  @mutex.synchronize do
    resumable = @running && !@aborted
    @running = false

    thread = @thread
    @thread = nil
    # 例外で終わったスレッドは非 nil のまま dead で残る。それに :stop を積んでも誰も
    # pop しないので、生きているときだけ積んで待つ
    if thread&.alive?
      @command_queue.uniq_push(:stop)
      thread.join
    end

    resumable
  end
end

#wait_for_download ⇒ Object 



73
74
75
 
# File 'lib/copy_tuner_client/poller.rb', line 73

def wait_for_download
  @cache.wait_for_download
end