Initial implementation of ActiveJob AsyncAdapter.

5 files changed, 67 insertions, 1 deletions

diff --git a/activejob/test/adapters/async.rb b/activejob/test/adapters/async.rb
new file mode 100644
index 0000000000..df58027599
--- /dev/null
+++ b/activejob/test/adapters/async.rb
@@ -0,0 +1,5 @@
+require 'concurrent'
+require 'active_job/async_job'
+
+ActiveJob::Base.queue_adapter = :async
+ActiveJob::AsyncJob.perform_immediately!
diff --git a/activejob/test/cases/async_job_test.rb b/activejob/test/cases/async_job_test.rb
new file mode 100644
index 0000000000..2642cfc608
--- /dev/null
+++ b/activejob/test/cases/async_job_test.rb
@@ -0,0 +1,42 @@
+require 'helper'
+require 'jobs/hello_job'
+require 'jobs/queue_as_job'
+
+class AsyncJobTest < ActiveSupport::TestCase
+  def using_async_adapter?
+    ActiveJob::Base.queue_adapter.is_a? ActiveJob::QueueAdapters::AsyncAdapter
+  end
+
+  setup do
+    ActiveJob::AsyncJob.perform_asynchronously!
+  end
+
+  teardown do
+    ActiveJob::AsyncJob::QUEUES.clear
+    ActiveJob::AsyncJob.perform_immediately!
+  end
+
+  test "#create_thread_pool returns a thread_pool" do
+    thread_pool = ActiveJob::AsyncJob.create_thread_pool
+    assert thread_pool.is_a? Concurrent::ExecutorService
+    assert_not thread_pool.is_a? Concurrent::ImmediateExecutor
+  end
+
+  test "#create_thread_pool returns an ImmediateExecutor after #perform_immediately! is called" do
+    ActiveJob::AsyncJob.perform_immediately!
+    thread_pool = ActiveJob::AsyncJob.create_thread_pool
+    assert thread_pool.is_a? Concurrent::ImmediateExecutor
+  end
+
+  test "enqueuing without specifying a queue uses the default queue" do
+    skip unless using_async_adapter?
+    HelloJob.perform_later
+    assert ActiveJob::AsyncJob::QUEUES.key? 'default'
+  end
+
+  test "enqueuing to a queue that does not exist creates the queue" do
+    skip unless using_async_adapter?
+    QueueAsJob.perform_later
+    assert ActiveJob::AsyncJob::QUEUES.key? QueueAsJob::MY_QUEUE.to_s
+  end
+end
diff --git a/activejob/test/integration/queuing_test.rb b/activejob/test/integration/queuing_test.rb
index ca8047ef0b..125ba54302 100644
--- a/activejob/test/integration/queuing_test.rb
+++ b/activejob/test/integration/queuing_test.rb
@@ -11,7 +11,7 @@ class QueuingTest < ActiveSupport::TestCase
   end
 
   test 'should not run jobs queued on a non-listening queue' do
-    skip if adapter_is?(:inline, :sucker_punch, :que)
+    skip if adapter_is?(:inline, :async, :sucker_punch, :que)
     old_queue = TestJob.queue_name
 
     begin
diff --git a/activejob/test/jobs/queue_as_job.rb b/activejob/test/jobs/queue_as_job.rb
new file mode 100644
index 0000000000..897aef52e5
--- /dev/null
+++ b/activejob/test/jobs/queue_as_job.rb
@@ -0,0 +1,10 @@
+require_relative '../support/job_buffer'
+
+class QueueAsJob < ActiveJob::Base
+  MY_QUEUE = :low_priority
+  queue_as MY_QUEUE
+
+  def perform(greeter = "David")
+    JobBuffer.add("#{greeter} says hello")
+  end
+end
diff --git a/activejob/test/support/integration/adapters/async.rb b/activejob/test/support/integration/adapters/async.rb
new file mode 100644
index 0000000000..42beb12b1f
--- /dev/null
+++ b/activejob/test/support/integration/adapters/async.rb
@@ -0,0 +1,9 @@
+module AsyncJobsManager
+  def setup
+    ActiveJob::Base.queue_adapter = :async
+  end
+
+  def clear_jobs
+    ActiveJob::AsyncJob::QUEUES.clear
+  end
+end