class Cloudtasker::LocalServer
Process jobs stored in Redis. Only to be used in development.
Constants
- CONCURRENCY
Max number of task requests sent to the processing server
- JOB_POLLING_FREQUENCY
Job Polling. How frequently to poll jobs in redis.
- QUEUE_CONCURRENCY
Default number of threads to allocate to process a specific queue
Public Instance Methods
process_jobs(queue = nil, concurrency = nil)
click to toggle source
Process enqueued workers.
# File lib/cloudtasker/local_server.rb, line 62 def process_jobs(queue = nil, concurrency = nil) @threads ||= {} @threads[queue] ||= [] max_threads = (concurrency || QUEUE_CONCURRENCY).to_i # Remove any done thread @threads[queue].select!(&:alive?) # Process tasks while @threads[queue].count < max_threads && (task = Cloudtasker::Backend::RedisTask.pop(queue)) @threads[queue] << Thread.new { process_task(task) } end end
process_task(task)
click to toggle source
Process a given task
@param [Cloudtasker::CloudTask] task The task to process
# File lib/cloudtasker/local_server.rb, line 81 def process_task(task) Thread.current['task'] = task Thread.current['attempts'] = 0 # Deliver task begin Thread.current['task']&.deliver rescue Errno::EBADF, Errno::ECONNREFUSED => e raise(e) unless Thread.current['attempts'] < 3 # Retry on connection error, in case the web server is not # started yet. Thread.current['attempts'] += 1 sleep(3) retry end end
start(opts = {})
click to toggle source
Start the local server
@param [Hash] opts Server options.
# File lib/cloudtasker/local_server.rb, line 40 def start(opts = {}) # Extract queues to process queues = opts[:queues].to_a.any? ? opts[:queues] : [[nil, CONCURRENCY]] # Display start banner queue_labels = queues.map { |n, c| "#{n || 'all'}=#{c || QUEUE_CONCURRENCY}" }.join(' ') Cloudtasker.logger.info("[Cloudtasker/Server] Processing queues: #{queue_labels}") # Start processing queues @start ||= Thread.new do until @done queues.each { |(n, c)| process_jobs(n, c) } sleep JOB_POLLING_FREQUENCY end Cloudtasker.logger.info('[Cloudtasker/Server] Local server exiting...') end end
stop()
click to toggle source
Stop the local server.
# File lib/cloudtasker/local_server.rb, line 21 def stop @done = true # Terminate threads and repush tasks @threads&.values&.flatten&.each do |t| t.terminate t['task']&.retry_later(0, is_error: false) end # Wait for main server to be done sleep 1 while @start&.alive? end