Sha256: b0797bd4980038e19c044bda89beb22d68a742327d26133a8e3c32f85e212677
Contents?: true
Size: 1.4 KB
Versions: 16
Compression:
Stored size: 1.4 KB
Contents
module Rpush
module Daemon
class Feeder
extend Reflectable
def self.start
self.should_stop = false
Rpush.config.push ? enqueue_notifications : feed_forever
end
def self.stop
self.should_stop = true
interruptible_sleeper.stop
@thread.join if @thread
end
def self.wakeup
interruptible_sleeper.wakeup
end
class << self
attr_accessor :should_stop
end
def self.feed_forever
@thread = Thread.new do
loop do
enqueue_notifications
interruptible_sleeper.sleep
break if should_stop
end
Rpush::Daemon.store.release_connection
end
@thread.join
end
def self.enqueue_notifications
batch_size = Rpush.config.batch_size - Rpush::Daemon::AppRunner.total_queued
return if batch_size <= 0
notifications = Rpush::Daemon.store.deliverable_notifications(batch_size)
Rpush::Daemon::AppRunner.enqueue(notifications)
rescue StandardError => e
Rpush.logger.error(e)
reflect(:error, e)
end
def self.interruptible_sleeper
return @interruptible_sleeper if @interruptible_sleeper
@interruptible_sleeper = InterruptibleSleep.new(Rpush.config.push_poll)
@interruptible_sleeper.start
@interruptible_sleeper
end
end
end
end
Version data entries
16 entries across 16 versions & 1 rubygems