Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
张重言
rails
提交
7059ab35
R
rails
项目概览
张重言
/
rails
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
R
rails
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
7059ab35
编写于
3月 18, 2015
作者:
W
wvengen
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Add job priorities to ActiveJob
上级
61f9e47f
变更
11
隐藏空白更改
内联
并排
Showing
11 changed file
with
132 addition
and
6 deletion
+132
-6
activejob/CHANGELOG.md
activejob/CHANGELOG.md
+4
-0
activejob/lib/active_job/base.rb
activejob/lib/active_job/base.rb
+2
-0
activejob/lib/active_job/core.rb
activejob/lib/active_job/core.rb
+8
-0
activejob/lib/active_job/enqueuing.rb
activejob/lib/active_job/enqueuing.rb
+4
-0
activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb
...ejob/lib/active_job/queue_adapters/delayed_job_adapter.rb
+2
-2
activejob/lib/active_job/queue_adapters/que_adapter.rb
activejob/lib/active_job/queue_adapters/que_adapter.rb
+2
-2
activejob/lib/active_job/queue_priority.rb
activejob/lib/active_job/queue_priority.rb
+44
-0
activejob/test/cases/queue_priority_test.rb
activejob/test/cases/queue_priority_test.rb
+47
-0
activejob/test/integration/queuing_test.rb
activejob/test/integration/queuing_test.rb
+12
-0
activejob/test/support/integration/test_case_helpers.rb
activejob/test/support/integration/test_case_helpers.rb
+6
-2
activejob/test/support/que/inline.rb
activejob/test/support/que/inline.rb
+1
-0
未找到文件。
activejob/CHANGELOG.md
浏览文件 @
7059ab35
*
Add job priorities to Active Job.
*wvengen*
*
Implement a simple
`AsyncJob`
processor and associated
`AsyncAdapter`
that
queue jobs to a
`concurrent-ruby`
thread pool.
...
...
activejob/lib/active_job/base.rb
浏览文件 @
7059ab35
require
'active_job/core'
require
'active_job/queue_adapter'
require
'active_job/queue_name'
require
'active_job/queue_priority'
require
'active_job/enqueuing'
require
'active_job/execution'
require
'active_job/callbacks'
...
...
@@ -57,6 +58,7 @@ class Base
include
Core
include
QueueAdapter
include
QueueName
include
QueuePriority
include
Enqueuing
include
Execution
include
Callbacks
...
...
activejob/lib/active_job/core.rb
浏览文件 @
7059ab35
...
...
@@ -18,6 +18,9 @@ module Core
# Queue in which the job will reside.
attr_writer
:queue_name
# Priority that the job will have (lower is more priority).
attr_writer
:priority
# ID optionally provided by adapter
attr_accessor
:provider_job_id
...
...
@@ -43,6 +46,7 @@ def deserialize(job_data)
# * <tt>:wait</tt> - Enqueues the job with the specified delay
# * <tt>:wait_until</tt> - Enqueues the job at the time specified
# * <tt>:queue</tt> - Enqueues the job on the specified queue
# * <tt>:priority</tt> - Enqueues the job with the specified priority
#
# ==== Examples
#
...
...
@@ -51,6 +55,7 @@ def deserialize(job_data)
# VideoJob.set(wait_until: Time.now.tomorrow).perform_later(Video.last)
# VideoJob.set(queue: :some_queue, wait: 5.minutes).perform_later(Video.last)
# VideoJob.set(queue: :some_queue, wait_until: Time.now.tomorrow).perform_later(Video.last)
# VideoJob.set(queue: :some_queue, wait: 5.minutes, priority: 10).perform_later(Video.last)
def
set
(
options
=
{})
ConfiguredJob
.
new
(
self
,
options
)
end
...
...
@@ -62,6 +67,7 @@ def initialize(*arguments)
@arguments
=
arguments
@job_id
=
SecureRandom
.
uuid
@queue_name
=
self
.
class
.
queue_name
@priority
=
self
.
class
.
priority
end
# Returns a hash with the job data that can safely be passed to the
...
...
@@ -71,6 +77,7 @@ def serialize
'job_class'
=>
self
.
class
.
name
,
'job_id'
=>
job_id
,
'queue_name'
=>
queue_name
,
'priority'
=>
priority
,
'arguments'
=>
serialize_arguments
(
arguments
),
'locale'
=>
I18n
.
locale
}
...
...
@@ -99,6 +106,7 @@ def serialize
def
deserialize
(
job_data
)
self
.
job_id
=
job_data
[
'job_id'
]
self
.
queue_name
=
job_data
[
'queue_name'
]
self
.
priority
=
job_data
[
'priority'
]
self
.
serialized_arguments
=
job_data
[
'arguments'
]
self
.
locale
=
job_data
[
'locale'
]
||
I18n
.
locale
end
...
...
activejob/lib/active_job/enqueuing.rb
浏览文件 @
7059ab35
...
...
@@ -32,6 +32,7 @@ def job_or_instantiate(*args)
# * <tt>:wait</tt> - Enqueues the job with the specified delay
# * <tt>:wait_until</tt> - Enqueues the job at the time specified
# * <tt>:queue</tt> - Enqueues the job on the specified queue
# * <tt>:priority</tt> - Enqueues the job with the specified priority
#
# ==== Examples
#
...
...
@@ -54,6 +55,7 @@ def retry_job(options={})
# * <tt>:wait</tt> - Enqueues the job with the specified delay
# * <tt>:wait_until</tt> - Enqueues the job at the time specified
# * <tt>:queue</tt> - Enqueues the job on the specified queue
# * <tt>:priority</tt> - Enqueues the job with the specified priority
#
# ==== Examples
#
...
...
@@ -61,10 +63,12 @@ def retry_job(options={})
# my_job_instance.enqueue wait: 5.minutes
# my_job_instance.enqueue queue: :important
# my_job_instance.enqueue wait_until: Date.tomorrow.midnight
# my_job_instance.enqueue priority: 10
def
enqueue
(
options
=
{})
self
.
scheduled_at
=
options
[
:wait
].
seconds
.
from_now
.
to_f
if
options
[
:wait
]
self
.
scheduled_at
=
options
[
:wait_until
].
to_f
if
options
[
:wait_until
]
self
.
queue_name
=
self
.
class
.
queue_name_from_part
(
options
[
:queue
])
if
options
[
:queue
]
self
.
priority
=
options
[
:priority
].
to_i
if
options
[
:priority
]
run_callbacks
:enqueue
do
if
self
.
scheduled_at
self
.
class
.
queue_adapter
.
enqueue_at
self
,
self
.
scheduled_at
...
...
activejob/lib/active_job/queue_adapters/delayed_job_adapter.rb
浏览文件 @
7059ab35
...
...
@@ -14,13 +14,13 @@ module QueueAdapters
# Rails.application.config.active_job.queue_adapter = :delayed_job
class
DelayedJobAdapter
def
enqueue
(
job
)
#:nodoc:
delayed_job
=
Delayed
::
Job
.
enqueue
(
JobWrapper
.
new
(
job
.
serialize
),
queue:
job
.
queue_name
)
delayed_job
=
Delayed
::
Job
.
enqueue
(
JobWrapper
.
new
(
job
.
serialize
),
queue:
job
.
queue_name
,
priority:
job
.
priority
)
job
.
provider_job_id
=
delayed_job
.
id
delayed_job
end
def
enqueue_at
(
job
,
timestamp
)
#:nodoc:
delayed_job
=
Delayed
::
Job
.
enqueue
(
JobWrapper
.
new
(
job
.
serialize
),
queue:
job
.
queue_name
,
run_at:
Time
.
at
(
timestamp
))
delayed_job
=
Delayed
::
Job
.
enqueue
(
JobWrapper
.
new
(
job
.
serialize
),
queue:
job
.
queue_name
,
priority:
job
.
priority
,
run_at:
Time
.
at
(
timestamp
))
job
.
provider_job_id
=
delayed_job
.
id
delayed_job
end
...
...
activejob/lib/active_job/queue_adapters/que_adapter.rb
浏览文件 @
7059ab35
...
...
@@ -16,13 +16,13 @@ module QueueAdapters
# Rails.application.config.active_job.queue_adapter = :que
class
QueAdapter
def
enqueue
(
job
)
#:nodoc:
que_job
=
JobWrapper
.
enqueue
job
.
serialize
que_job
=
JobWrapper
.
enqueue
job
.
serialize
,
priority:
job
.
priority
job
.
provider_job_id
=
que_job
.
attrs
[
"job_id"
]
que_job
end
def
enqueue_at
(
job
,
timestamp
)
#:nodoc:
que_job
=
JobWrapper
.
enqueue
job
.
serialize
,
run_at:
Time
.
at
(
timestamp
)
que_job
=
JobWrapper
.
enqueue
job
.
serialize
,
priority:
job
.
priority
,
run_at:
Time
.
at
(
timestamp
)
job
.
provider_job_id
=
que_job
.
attrs
[
"job_id"
]
que_job
end
...
...
activejob/lib/active_job/queue_priority.rb
0 → 100644
浏览文件 @
7059ab35
module
ActiveJob
module
QueuePriority
extend
ActiveSupport
::
Concern
# Includes the ability to override the default queue priority.
module
ClassMethods
mattr_accessor
(
:default_priority
)
# Specifies the priority of the queue to create the job with.
#
# class PublishToFeedJob < ActiveJob::Base
# queue_with_priority 50
#
# def perform(post)
# post.to_feed!
# end
# end
#
# Specify either an argument or a block.
def
queue_with_priority
(
priority
=
nil
,
&
block
)
if
block_given?
self
.
priority
=
block
else
self
.
priority
=
priority
end
end
end
included
do
class_attribute
:priority
,
instance_accessor:
false
self
.
priority
=
default_priority
end
# Returns the priority that the job will be created with
def
priority
if
@priority
.
is_a?
(
Proc
)
@priority
=
instance_exec
(
&
@priority
)
end
@priority
end
end
end
activejob/test/cases/queue_priority_test.rb
0 → 100644
浏览文件 @
7059ab35
require
'helper'
require
'jobs/hello_job'
class
QueuePriorityTest
<
ActiveSupport
::
TestCase
test
'priority unset by default'
do
assert_equal
nil
,
HelloJob
.
priority
end
test
'uses given priority'
do
original_priority
=
HelloJob
.
priority
begin
HelloJob
.
queue_with_priority
90
assert_equal
90
,
HelloJob
.
new
.
priority
ensure
HelloJob
.
priority
=
original_priority
end
end
test
'evals block given to priority to determine priority'
do
original_priority
=
HelloJob
.
priority
begin
HelloJob
.
queue_with_priority
{
25
}
assert_equal
25
,
HelloJob
.
new
.
priority
ensure
HelloJob
.
priority
=
original_priority
end
end
test
'can use arguments to determine priority in priority block'
do
original_priority
=
HelloJob
.
priority
begin
HelloJob
.
queue_with_priority
{
self
.
arguments
.
first
==
'1'
?
99
:
11
}
assert_equal
99
,
HelloJob
.
new
(
'1'
).
priority
assert_equal
11
,
HelloJob
.
new
(
'3'
).
priority
ensure
HelloJob
.
priority
=
original_priority
end
end
test
'uses priority passed to #set'
do
job
=
HelloJob
.
set
(
priority:
123
).
perform_later
assert_equal
123
,
job
.
priority
end
end
activejob/test/integration/queuing_test.rb
浏览文件 @
7059ab35
...
...
@@ -84,4 +84,16 @@ class QueuingTest < ActiveSupport::TestCase
I18n
.
locale
=
:en
end
end
test
'should run job with higher priority first'
do
skip
unless
adapter_is?
(
:delayed_job
,
:que
)
wait_until
=
Time
.
now
+
3
.
seconds
TestJob
.
set
(
wait_until:
wait_until
,
priority:
20
).
perform_later
"
#{
@id
}
.1"
TestJob
.
set
(
wait_until:
wait_until
,
priority:
10
).
perform_later
"
#{
@id
}
.2"
wait_for_jobs_to_finish_for
(
10
.
seconds
)
assert
job_executed
"
#{
@id
}
.1"
assert
job_executed
"
#{
@id
}
.2"
assert
job_executed_at
(
"
#{
@id
}
.2"
)
<
job_executed_at
(
"
#{
@id
}
.1"
)
end
end
activejob/test/support/integration/test_case_helpers.rb
浏览文件 @
7059ab35
...
...
@@ -42,8 +42,12 @@ def wait_for_jobs_to_finish_for(seconds=60)
end
end
def
job_executed
Dummy
::
Application
.
root
.
join
(
"tmp/
#{
@id
}
"
).
exist?
def
job_executed
(
id
=
@id
)
Dummy
::
Application
.
root
.
join
(
"tmp/
#{
id
}
"
).
exist?
end
def
job_executed_at
(
id
=
@id
)
File
.
new
(
Dummy
::
Application
.
root
.
join
(
"tmp/
#{
id
}
"
)).
ctime
end
def
job_output
...
...
activejob/test/support/que/inline.rb
浏览文件 @
7059ab35
...
...
@@ -6,6 +6,7 @@ def self.enqueue(*args)
if
args
.
last
.
is_a?
(
Hash
)
options
=
args
.
pop
options
.
delete
(
:run_at
)
options
.
delete
(
:priority
)
args
<<
options
unless
options
.
empty?
end
self
.
run
(
*
args
)
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录