Get rid of broadway

lib/faktory_worker/batch.ex

@@ -7,11 +7,6 @@ defmodule FaktoryWorker.Batch do
   all jobs in a batch have completed, Faktory will queue a callback job. This
   allows building complex job workflows with dependencies.
 
-  Jobs pushed as part of a batch _must_ be pushed synchronously. This can be
-  done using the `skip_pipeline: true` option when calling `perform_async/2`. If
-  a job isn't pushed synchronously, you may encounter a race condition where the
-  batch is committed before all jobs have been pushed.
-
   ## Creating a batch
 
   A batch is created using `new!/1` and must provide a description and declare
@@ -25,9 +20,9 @@ defmodule FaktoryWorker.Batch do
   alias FaktoryWorker.Batch
 
   {:ok, bid} = Batch.new!(on_success: {MyApp.EmailReportJob, [], []})
-  MyApp.Job.perform_async([1, 2], custom: %{"bid" => bid}, skip_pipeline: true)
-  MyApp.Job.perform_async([3, 4], custom: %{"bid" => bid}, skip_pipeline: true)
-  MyApp.Job.perform_async([5, 6], custom: %{"bid" => bid}, skip_pipeline: true)
+  MyApp.Job.perform_async([1, 2], custom: %{"bid" => bid})
+  MyApp.Job.perform_async([3, 4], custom: %{"bid" => bid})
+  MyApp.Job.perform_async([5, 6], custom: %{"bid" => bid})
   Batch.commit(bid)
   ```
 
@@ -45,7 +40,7 @@ defmodule FaktoryWorker.Batch do
     def perform(arg1, arg2, bid) do
       Batch.open(bid)
 
-      MyApp.OtherJob.perform_async([1, 2], custom: %{"bid" => bid}, skip_pipeline: true)
+      MyApp.OtherJob.perform_async([1, 2], custom: %{"bid" => bid})
 
       Batch.commit(bid)
     end

lib/faktory_worker/job.ex

@@ -59,13 +59,10 @@ defmodule FaktoryWorker.Job do
 
   ## Synchronous job pushing
 
-  By default, jobs are pushed asynchronously to the Faktory server. To ensure a
-  job has been successfully submitted before continuing, jobs can be pushed
-  synchronously instead. To do this, pass the `:skip_pipeline` option with the
-  value of `true` to `perform_async/2`.
-
-  Synchronous pushing is required in certain situations to guarantee ordering,
-  such as when using the Faktory Enterprise batching feature.
+  Previous version used Broadway to send jobs and `:skip_pipeline` parameter was used to do it synchronously.
+  `:skip_pipeline` is not supported anymore.
+  Since Batch operations is a feature of Faktory Enterprise this library now sends any single job synchronously
+  and makes HTTP call to faktory server (see `FaktoryWorker.Batch`).
 
   ## Worker Configuration
 
@@ -88,7 +85,7 @@ defmodule FaktoryWorker.Job do
   means only values that implement the `Jason.Encoder` protocol are valid when calling the `perform_async/2` function.
   """
 
-  alias FaktoryWorker.{ConnectionManager, Random, Pool, Sandbox, Telemetry}
+  alias FaktoryWorker.{ConnectionManager, Random, Pool, Telemetry}
 
   # Look at supporting the following optional fields when pushing a job
   # priority
@@ -128,37 +125,9 @@ defmodule FaktoryWorker.Job do
 
   @doc false
   def perform_async(payload, opts) do
-    case Keyword.get(opts, :skip_pipeline, false) do
-      false ->
-        opts
-        |> push_pipeline_name()
-        |> perform_async(payload, opts)
-
-      true ->
-        opts
-        |> faktory_name()
-        |> push(payload)
-    end
-  end
-
-  @doc false
-  def perform_async(_, {:error, _} = error, _), do: error
-
-  def perform_async(pipeline_name, payload, opts) do
-    if Sandbox.active?() do
-      Sandbox.enqueue_job(
-        String.to_existing_atom("Elixir." <> payload.jobtype),
-        payload.args,
-        opts
-      )
-    else
-      # TODO:: do we want to split pipelines into multiple threads???
-      # It was implemented with a Broadway & Producers <--> processes model.
-      # But Faktory is a job server and do we want to care and build one more not needed layer???
-      opts
-      |> faktory_name()
-      |> push(payload)
-    end
+    opts
+    |> faktory_name()
+    |> push(payload)
   end
 
   @doc false
@@ -170,6 +139,7 @@ defmodule FaktoryWorker.Job do
   end
 
   @doc false
+  def push(_, invalid_payload = {:error, _}), do: invalid_payload
   def push(faktory_name, job) do
     faktory_name
     |> Pool.format_pool_name()
@@ -215,16 +185,6 @@ defmodule FaktoryWorker.Job do
     "The field '#{Atom.to_string(field)}' has an invalid value '#{inspect(value)}'"
   end
 
-  defp push_pipeline_name(opts) do
-    opts
-    |> faktory_name()
-    |> format_pipeline_name()
-  end
-
-  defp format_pipeline_name(name) when is_atom(name) do
-    :"#{name}_pipeline"
-  end
-
   defp faktory_name(opts) do
     Keyword.get(opts, :faktory_name, FaktoryWorker)
   end

test/faktory_worker/connection_manager_test.exs

@@ -91,7 +91,7 @@ defmodule FaktoryWorker.ConnectionManagerTest do
                {{:ok, result}, _} = ConnectionManager.send_command(state, {:push, payload})
 
                assert result == "halt reason"
-             end) =~ "[warn]  [123456] Halt: halt reason"
+             end) |> String.match?(~r/\[warn.*(?<!ing)\]*\[123456\]*[Halt: halt reason]/)
     end
 
     test "should unset the connection when there is a socket failure" do

test/faktory_worker/job/batch_integration_test.exs

@@ -34,8 +34,7 @@ defmodule FaktoryWorker.BatchIntegrationTest do
 
       job_opts = [
         faktory_name: faktory_name,
-        custom: %{"bid" => bid},
-        skip_pipeline: true
+        custom: %{"bid" => bid}
       ]
 
       DefaultWorker.perform_async(["1"], job_opts)
@@ -72,8 +71,7 @@ defmodule FaktoryWorker.BatchIntegrationTest do
 
       job_opts = [
         faktory_name: faktory_name,
-        custom: %{"bid" => bid},
-        skip_pipeline: true
+        custom: %{"bid" => bid}
       ]
 
       DefaultWorker.perform_async(["1"], job_opts)

test/faktory_worker/job/job_integration_test.exs

@@ -9,7 +9,7 @@ defmodule FaktoryWorker.JobIntegrationTest do
 
   setup :flush_faktory!
 
-  describe "perform_async/3" do
+  describe "perform_async/2" do
     test "should send a new job to faktory" do
       faktory_name = :"Test_#{Random.string()}"
 
@@ -21,7 +21,8 @@ defmodule FaktoryWorker.JobIntegrationTest do
 
       job = Job.build_payload(DefaultWorker, %{hey: "there!"}, opts)
 
-      Job.perform_async(job, opts)
+      {:ok, job_sent} = Job.perform_async(job, opts)
+      assert job_sent == job
 
       assert_queue_size("default", 1)
 

test/faktory_worker/job/job_test.exs

@@ -184,15 +184,4 @@ defmodule FaktoryWorker.Job.JobTest do
     end
   end
 
-  describe "perform_async/3" do
-    test "should not send a bad payload" do
-      data = %{hey: "there!"}
-      opts = [queue: 123]
-      {:error, _} = payload = Job.build_payload(Test.Worker, data, opts)
-
-      {:error, error} = Job.perform_async(TestPipeline, payload, [])
-
-      assert error == "The field 'queue' has an invalid value '123'"
-    end
-  end
 end