Class: Rpush::Daemon::DispatcherLoop

Inherits:
Object
  • Object
show all
Includes:
Loggable, Reflectable
Defined in:
lib/rpush/daemon/dispatcher_loop.rb

Constant Summary collapse

STOP = 
:stop

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Loggable

#log_debug, #log_error, #log_info, #log_warn

Methods included from Reflectable

#reflect

Constructor Details

#initialize(queue, dispatcher) ⇒ DispatcherLoop

Returns a new instance of DispatcherLoop.



11
12
13
14
15
 
# File 'lib/rpush/daemon/dispatcher_loop.rb', line 11

def initialize(queue, dispatcher)
  @queue = queue
  @dispatcher = dispatcher
  @dispatch_count = 0
end

Instance Attribute Details

#dispatch_count ⇒ Object (readonly)

Returns the value of attribute dispatch_count.



7
8
9
 
# File 'lib/rpush/daemon/dispatcher_loop.rb', line 7

def dispatch_count
  @dispatch_count
end

#started_at ⇒ Object (readonly)

Returns the value of attribute started_at.



7
8
9
 
# File 'lib/rpush/daemon/dispatcher_loop.rb', line 7

def started_at
  @started_at
end

Instance Method Details

#start ⇒ Object 



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/rpush/daemon/dispatcher_loop.rb', line 21

def start
  @started_at = Time.now

  @thread = Thread.new do
    loop do
      payload = @queue.pop
      if stop_payload?(payload)
        break if should_stop?(payload)

        # Intended for another dispatcher loop.
        @queue.push(payload)
        Thread.pass
        sleep 0.1
      else
        dispatch(payload)
      end
    end

    Rpush::Daemon.store.release_connection
  end
end

#stop ⇒ Object 



43
44
45
46
47
48
49
50
51
52
 
# File 'lib/rpush/daemon/dispatcher_loop.rb', line 43

def stop
  @queue.push([STOP, object_id]) if @thread
  @thread.join if @thread
  @dispatcher.cleanup
rescue StandardError => e
  log_error(e)
  reflect(:error, e)
ensure
  @thread = nil
end

#thread_status ⇒ Object 



17
18
19
 
# File 'lib/rpush/daemon/dispatcher_loop.rb', line 17

def thread_status
  @thread ? @thread.status : 'not started'
end