require_relative './tradesman' module WorkerRoulette class ATradesman < Tradesman def wait_for_work_orders(on_subscribe_callback = nil, &on_message_callback) @redis_pubsub ||= WorkerRoulette.new_redis_pubsub #cannot use connection pool bc redis expects each obj to own its own pubsub connection for the life of the subscription @redis_pubsub.on(:subscribe) {|channel, subscription_count| on_subscribe_callback.call(channel, subscription_count) if on_subscribe_callback} @redis_pubsub.on(:message) {|channel, message| set_timer(on_message_callback); get_messages(message, channel, on_message_callback)} @redis_pubsub.subscribe(@channel) end def unsubscribe(&callback) deferable = @redis_pubsub.unsubscribe(@channel) deferable.callback do @redis_pubsub.close_connection @redis_pubsub = nil callback.call end deferable.errback do @redis_pubsub.close_connection @redis_pubsub = nil callback.call end end private attr_reader :timer def get_messages(message, channel, on_message_callback) return unless on_message_callback work_orders! do |work_orders_1| work_orders! do |work_orders| on_message_callback.call(work_orders_1 + work_orders, message, channel) end end end def set_timer(on_message_callback) return unless on_message_callback @timer && @timer.cancel @timer = EM::PeriodicTimer.new(rand(20..25)) do work_orders! {|work_orders| on_message_callback.call(work_orders, nil, nil)} end end end end