Class: Narou::Worker

Inherits:
Object
  • Object
show all
Includes:
Singleton
Defined in:
lib/web/worker.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Worker

Returns a new instance of Worker.



13
14
15
16
17
18
19
 
# File 'lib/web/worker.rb', line 13

def initialize
  @queue = Queue.new
  @size = 0
  @mutex = Mutex.new
  @worker_thread = nil
  @push_server = Narou::PushServer.instance
end

Instance Attribute Details

#size ⇒ Object (readonly)

Returns the value of attribute size.



11
12
13
 
# File 'lib/web/worker.rb', line 11

def size
  @size
end

Class Method Details

.push(&block) ⇒ Object 



61
62
63
 
# File 'lib/web/worker.rb', line 61

def self.push(&block)
  instance.push(&block)
end

.push_as_system_worker(&block) ⇒ Object

システム用のワーカー追加。内部カウントは増やさない



57
58
59
 
# File 'lib/web/worker.rb', line 57

def self.push_as_system_worker(&block)
  instance.push(false, &block)
end

Instance Method Details

#countdown ⇒ Object 



81
82
83
84
85
 
# File 'lib/web/worker.rb', line 81

def countdown
  @mutex.synchronize do
    @size -= 1
  end
end

#countup ⇒ Object 



75
76
77
78
79
 
# File 'lib/web/worker.rb', line 75

def countup
  @mutex.synchronize do
    @size += 1
  end
end

#notification_queue ⇒ Object 



71
72
73
 
# File 'lib/web/worker.rb', line 71

def notification_queue
  @push_server.send_all("notification.queue" => @size)
end

#push(counting = true, &block) ⇒ Object 



65
66
67
68
69
 
# File 'lib/web/worker.rb', line 65

def push(counting = true, &block)
  countup if counting
  notification_queue
  @queue.push(block: block, counting: counting)
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


21
22
23
 
# File 'lib/web/worker.rb', line 21

def running?
  !@worker_thread.!
end

#start ⇒ Object 



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/web/worker.rb', line 25

def start
  return if running?
  @worker_thread = Thread.new do
    loop do
      begin
        q = @queue.pop
        q[:block].call
      rescue SystemExit
      rescue Exception => e
        # Workerスレッド内での例外は表示するだけしてスレッドは生かしたままにする
        STDOUT.puts $@.shift + ": #{e.message} (#{e.class})"
        $@.each do |b|
          STDOUT.puts "  from #{b}"
        end
      ensure
        if q[:counting]
          countdown
          notification_queue
        end
      end
    end
  end
end

#stop ⇒ Object 



49
50
51
52
 
# File 'lib/web/worker.rb', line 49

def stop
  @worker_thread.kill if @worker_thread
  @worker_thread = nil
end