diff --git a/bin/que b/bin/que index 7766431..dab3032 100755 --- a/bin/que +++ b/bin/que @@ -67,10 +67,6 @@ OptionParser.new do |opts| $stdout.puts opts exit 0 end - - opts.on("--using-lock-database [USING_LOCK_DATABSE]", FalseClass, "sets if we want to use seperate database for locking") do |using_lock_database| - options.using_lock_database = using_lock_database - end end.parse!(ARGV) # rubocop:enable Metrics/LineLength @@ -97,7 +93,6 @@ cursor_expiry = options.cursor_expiry || wake_interval worker_count = options.worker_count || 1 timeout = options.timeout secondary_queues = options.secondary_queues || [] -using_lock_database = options.using_lock_database || false Que.logger ||= Logger.new(STDOUT) @@ -123,7 +118,6 @@ worker_group = Que::WorkerGroup.start( lock_window: options.lock_window, lock_budget: options.lock_budget, secondary_queues: secondary_queues, - using_lock_database: using_lock_database, ) if options.metrics_port diff --git a/lib/que/locker.rb b/lib/que/locker.rb index 63800a3..00df098 100644 --- a/lib/que/locker.rb +++ b/lib/que/locker.rb @@ -53,14 +53,14 @@ class Locker ), ].freeze - def initialize(queue:, cursor_expiry:, window: nil, budget: nil, secondary_queues: [], using_lock_database: false) + def initialize(queue:, cursor_expiry:, window: nil, budget: nil, secondary_queues: []) @queue = queue @cursor_expiry = cursor_expiry @queue_cursors = {} @queue_expires_at = {} @secondary_queues = secondary_queues @consolidated_queues = Array.wrap(queue).concat(secondary_queues) - @using_lock_database = using_lock_database + @using_lock_database = ENV.fetch("YUGABYTE_QUE_WORKER_ENABLED", false) if @using_lock_database @lock_database_connection = LockDataBaseRecord.connection diff --git a/lib/que/worker.rb b/lib/que/worker.rb index d90dd0c..8bfddf6 100644 --- a/lib/que/worker.rb +++ b/lib/que/worker.rb @@ -121,8 +121,7 @@ def initialize( lock_cursor_expiry: DEFAULT_WAKE_INTERVAL, lock_window: nil, lock_budget: nil, - secondary_queues: [], - using_lock_database: false + secondary_queues: [] ) @queue = queue @wake_interval = wake_interval @@ -135,8 +134,7 @@ def initialize( cursor_expiry: lock_cursor_expiry, window: lock_window, budget: lock_budget, - secondary_queues: secondary_queues, - using_lock_database: using_lock_database, + secondary_queues: secondary_queues ) end diff --git a/spec/lib/que/worker_spec.rb b/spec/lib/que/worker_spec.rb index 7cb7985..2d1ec0c 100644 --- a/spec/lib/que/worker_spec.rb +++ b/spec/lib/que/worker_spec.rb @@ -206,7 +206,7 @@ class ExceptionalJobWithCustomLogging < ExceptionalJob FakeJob.enqueue(1) expect(Que). - to receive(:execute).with(:lock_job, ["default", 0]).and_raise(PG::Error) + to receive(:execute).with(:lock_job, include("default", 0)).and_raise(PG::Error) expect(subject).to eq(:postgres_error) end end @@ -216,7 +216,7 @@ class ExceptionalJobWithCustomLogging < ExceptionalJob FakeJob.enqueue(1) expect(Que). - to receive(:execute).with(:lock_job, ["default", 0]). + to receive(:execute).with(:lock_job, include("default", 0)). and_raise(ActiveRecord::ConnectionTimeoutError) expect(subject).to eq(:postgres_error) end @@ -227,7 +227,7 @@ class ExceptionalJobWithCustomLogging < ExceptionalJob FakeJob.enqueue(1) expect(Que). - to receive(:execute).with(:lock_job, ["default", 0]). + to receive(:execute).with(:lock_job, include("default", 0)). and_raise(ActiveRecord::ConnectionNotEstablished) expect(subject).to eq(:postgres_error) end