drop worker demand to 1

apps/core/lib/core.ex

@@ -6,6 +6,24 @@ defmodule Core do
 
   def conf(key), do: Application.get_env(:core, key)
 
+  def env(var, :int, default) do
+    case System.get_env(var) do
+      "" -> default
+      v when is_binary(v) -> String.to_integer(v)
+      _ -> default
+    end
+  end
+
+  def env(var, :bool, default) do
+    case System.get_env(var) do
+      "1" -> true
+      "0" -> false
+      "true" -> true
+      "false" -> false
+      _ -> default
+    end
+  end
+
   def random_phrase(len, sep \\ "-") do
     Enum.map(0..len, fn _ -> Dictionary.random_word() end)
     |> Enum.join(sep)

apps/worker/lib/worker/docker/pipeline.ex

@@ -3,12 +3,14 @@ defmodule Worker.Docker.Pipeline do
   require Logger
 
   def start_link(producer) do
-    Flow.from_stages([producer], stages: 1, max_demand: 5)
+    Flow.from_stages([producer], stages: 1, max_demand: demand())
     |> Flow.map(fn img ->
       Logger.info "Scheduling docker scan for #{img.id}"
       img
     end)
     |> Flow.map(&Worker.Conduit.Subscribers.Docker.scan_image/1)
     |> Flow.start_link()
   end
+
+  defp demand(), do: Core.env("DOCKER_SCAN_PARALLELISM", :int, 1)
 end

apps/worker/lib/worker/docker/producer.ex

@@ -41,10 +41,7 @@ defmodule Worker.Docker.Producer do
   end
 
   defp poll_interval() do
-    case System.get_env("DOCKER_SCAN_POLL_INTERVAL") do
-      v when is_binary(v) -> String.to_integer(v)
-      _ -> 60
-    end
+    Core.env("DOCKER_SCAN_POLL_INTERVAL", :int, 60)
     |> :timer.seconds()
   end
 end