Update gen_stage version requirement

.github/workflows/main.yml

@@ -1,16 +1,24 @@
 name: Elixir CI
 
-# This workflow runs on pushes to the main branch
+# This workflow runs on pushes and pull requests targeting master.
 on:
   push:
     branches: [ master ]
+  pull_request:
+    branches: [ master ]
 
 jobs:
   build:
-    runs-on: elixir:1.8.2-otp-22
+    runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
 
+      - name: Set up Elixir
+        uses: erlef/setup-beam@v1
+        with:
+          elixir-version: '1.18'
+          otp-version: '27'
+
       - name: Install dependencies
         run: mix deps.get
 

.gitlab-ci.yml

@@ -1,4 +1,4 @@
-image: elixir:1.6.6-alpine
+image: elixir:1.18-alpine
 cache:
   key: flume
   paths:

README.md

@@ -38,8 +38,7 @@ Flume is a job processing system backed by [GenStage](https://github.com/elixir-
 
 ## Requirements
 
-- Elixir 1.6.6+
-- Erlang/OTP 21.1+
+- Elixir 1.18+
 - Redis 4.0+
 
 ## Installation

lib/flume/bulk_event.ex

@@ -3,6 +3,7 @@ defmodule Flume.BulkEvent do
   This module is responsible for creating the bulk event
   received by the consumer stages.
   """
+
   # Sample Event Schema
   # {
   #   "class": "Elixir.Worker",

lib/flume/pipeline/control/options.ex

@@ -54,9 +54,7 @@ defmodule Flume.Pipeline.Control.Options do
        when (not is_integer(value) or value < 0) and value != :infinity,
        do:
          {:error,
-          "expected #{inspect(key)} to be a non-negative integer or :infinity, got: #{
-            inspect(value)
-          }"}
+          "expected #{inspect(key)} to be a non-negative integer or :infinity, got: #{inspect(value)}"}
 
   defp validate_type(_type, _key, value), do: {:ok, value}
 end

lib/flume/pipeline/event.ex

@@ -39,15 +39,15 @@ defmodule Flume.Pipeline.Event do
   end
 
   def pause(pipeline_name, opts) do
-    unless opts[:temporary] do
+    if !opts[:temporary] do
       RedisClient.set(paused_redis_key(pipeline_name), true)
     end
 
     EventPipeline.Producer.pause(pipeline_name, opts[:async], opts[:timeout])
   end
 
   def resume(pipeline_name, opts) do
-    unless opts[:temporary] do
+    if !opts[:temporary] do
       RedisClient.del(paused_redis_key(pipeline_name))
     end
 

lib/flume/redis/bulk_dequeue.ex

@@ -259,9 +259,7 @@ defmodule Flume.Redis.BulkDequeue do
 
       error_res ->
         Logger.info(
-          "[Dequeue]: #{dequeue_key} error: Expected: #{inspect(expected_response)}, Got: #{
-            inspect(error_res)
-          }, queue_name: #{dequeue_key}"
+          "[Dequeue]: #{dequeue_key} error: Expected: #{inspect(expected_response)}, Got: #{inspect(error_res)}, queue_name: #{dequeue_key}"
         )
 
         {:ok, []}
@@ -289,9 +287,7 @@ defmodule Flume.Redis.BulkDequeue do
 
       error_res ->
         Logger.info(
-          "[Dequeue]: error: Expected: #{inspect(expected_response)}, Got: #{inspect(error_res)}, queue_name: #{
-            dequeue_key
-          }"
+          "[Dequeue]: error: Expected: #{inspect(expected_response)}, Got: #{inspect(error_res)}, queue_name: #{dequeue_key}"
         )
 
         {:ok, []}

lib/flume/support/time.ex

@@ -2,21 +2,21 @@ defmodule Flume.Support.Time do
   import DateTime, only: [utc_now: 0, to_unix: 2, from_unix!: 2]
 
   def offset_from_now(offset_in_ms, current_time \\ utc_now()) do
-    now_in_µs = current_time |> to_unix(:microsecond)
-    offset_in_µs = offset_in_ms * 1_000
+    now_in_μs = current_time |> to_unix(:microsecond)
+    offset_in_μs = offset_in_ms * 1_000
 
-    now_in_µs
-    |> Kernel.+(offset_in_µs)
+    now_in_μs
+    |> Kernel.+(offset_in_μs)
     |> round()
     |> from_unix!(:microsecond)
   end
 
   def offset_before(offset_in_ms, current_time \\ utc_now()) do
-    now_in_µs = current_time |> to_unix(:microsecond)
-    offset_in_µs = offset_in_ms * 1_000
+    now_in_μs = current_time |> to_unix(:microsecond)
+    offset_in_μs = offset_in_ms * 1_000
 
-    now_in_µs
-    |> Kernel.-(offset_in_µs)
+    now_in_μs
+    |> Kernel.-(offset_in_μs)
     |> round()
     |> from_unix!(:microsecond)
   end

mix.exs

@@ -5,7 +5,7 @@ defmodule Flume.Mixfile do
     [
       app: :flume,
       version: "0.2.0",
-      elixir: "~> 1.8",
+      elixir: "~> 1.18",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,
       deps: deps(),
@@ -25,15 +25,6 @@ defmodule Flume.Mixfile do
   # Run "mix help compile.app" to learn about applications.
   def application do
     [
-      applications: [
-        :redix,
-        :logger_file_backend,
-        :gen_stage,
-        :jason,
-        :poolboy,
-        :retry,
-        :telemetry
-      ],
       extra_applications: [:logger],
       mod: {Flume, []}
     ]
@@ -43,12 +34,12 @@ defmodule Flume.Mixfile do
   defp deps do
     [
       {:redix, "~> 1.0"},
-      {:gen_stage, "~> 0.14.0"},
+      {:gen_stage, "~> 1.0"},
       {:jason, "~> 1.1.0"},
       {:poolboy, "~> 1.5.1"},
       {:elixir_uuid, "~> 1.2"},
       {:logger_file_backend, "~> 0.0.10"},
-      {:retry, "0.8.2"},
+      {:retry, "~> 0.16.0"},
       {:benchee, "~> 1.0"},
       {:telemetry, "~> 1.0"},
       {:excoveralls, "~> 0.10.6", only: :test}

mix.lock

@@ -1,23 +1,25 @@
 %{
-  "benchee": {:hex, :benchee, "1.0.1", "66b211f9bfd84bd97e6d1beaddf8fc2312aaabe192f776e8931cb0c16f53a521", [:mix], [{:deep_merge, "~> 1.0", [hex: :deep_merge, repo: "hexpm", optional: false]}], "hexpm", "3ad58ae787e9c7c94dd7ceda3b587ec2c64604563e049b2a0e8baafae832addb"},
-  "certifi": {:hex, :certifi, "2.5.1", "867ce347f7c7d78563450a18a6a28a8090331e77fa02380b4a21962a65d36ee5", [:rebar3], [{:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm", "805abd97539caf89ec6d4732c91e62ba9da0cda51ac462380bbd28ee697a8c42"},
+  "benchee": {:hex, :benchee, "1.5.1", "b95cbc36c4b98969a5c592a246e171041eb683c56bad1cb4f49a3b081ba66087", [:mix], [{:deep_merge, "~> 1.0", [hex: :deep_merge, repo: "hexpm", optional: false]}, {:statistex, "~> 1.1", [hex: :statistex, repo: "hexpm", optional: false]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a539301f8dfd4efc5c5123bfb9d47ebde20092a863a5b5b16c2a60d2243dfce7"},
+  "certifi": {:hex, :certifi, "2.15.0", "0e6e882fcdaaa0a5a9f2b3db55b1394dba07e8d6d9bcad08318fb604c6839712", [:rebar3], [], "hexpm", "b147ed22ce71d72eafdad94f055165c1c182f61a2ff49df28bcc71d1d5b94a60"},
   "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm"},
-  "deep_merge": {:hex, :deep_merge, "1.0.0", "b4aa1a0d1acac393bdf38b2291af38cb1d4a52806cf7a4906f718e1feb5ee961", [:mix], [], "hexpm", "ce708e5f094b9cd4e8f2be4f00d2f4250c4095be93f8cd6d018c753894885430"},
+  "deep_merge": {:hex, :deep_merge, "1.0.2", "476aa7ea61c54de96220051b998d893869069094da65b96101aebf79416f8a1e", [:mix], [], "hexpm", "737a53cdc9758fedbb608bdc213969e65729466c4ef3cd8e8726d0335dff116c"},
   "elixir_uuid": {:hex, :elixir_uuid, "1.2.1", "dce506597acb7e6b0daeaff52ff6a9043f5919a4c3315abb4143f0b00378c097", [:mix], [], "hexpm", "f7eba2ea6c3555cea09706492716b0d87397b88946e6380898c2889d68585752"},
   "excoveralls": {:hex, :excoveralls, "0.10.6", "e2b9718c9d8e3ef90bc22278c3f76c850a9f9116faf4ebe9678063310742edc2", [:mix], [{:hackney, "~> 1.13", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "b06c73492aa9940c4c29cfc1356bcf5540ae318f17b423749a0615a66ee3e049"},
-  "gen_stage": {:hex, :gen_stage, "0.14.0", "65ae78509f85b59d360690ce3378d5096c3130a0694bab95b0c4ae66f3008fad", [:mix], [], "hexpm", "095d38418e538af99ac82043985d26724164b78736a1d0f137c308332ad46250"},
-  "hackney": {:hex, :hackney, "1.15.1", "9f8f471c844b8ce395f7b6d8398139e26ddca9ebc171a8b91342ee15a19963f4", [:rebar3], [{:certifi, "2.5.1", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "6.0.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "1.0.1", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~>1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "1.1.4", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm", "c2790c9f0f7205f4a362512192dee8179097394400e745e4d20bab7226a8eaad"},
-  "idna": {:hex, :idna, "6.0.0", "689c46cbcdf3524c44d5f3dde8001f364cd7608a99556d8fbd8239a5798d4c10", [:rebar3], [{:unicode_util_compat, "0.4.1", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "4bdd305eb64e18b0273864920695cb18d7a2021f31a11b9c5fbcd9a253f936e2"},
-  "jason": {:hex, :jason, "1.1.0", "9634bca30f2f7468dde3e704d5865319b1eb88e4a8cded5c995baf0aa957524f", [:mix], [{:decimal, "~> 1.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "23e9d96104cce9e74a63356f280ad7c7abc8ee68a45fb051d0845236bc94386c"},
-  "logger_file_backend": {:hex, :logger_file_backend, "0.0.11", "3bbc5f31d3669e8d09d7a9443e86056fae7fc18e45c6f748c33b8c79a7e147a1", [:mix], [], "hexpm", "62be826f04644c62b0a2bc98a13e2e7ae52c0a4eda020f4c59d7287356d5e445"},
+  "gen_stage": {:hex, :gen_stage, "1.3.2", "7c77e5d1e97de2c6c2f78f306f463bca64bf2f4c3cdd606affc0100b89743b7b", [:mix], [], "hexpm", "0ffae547fa777b3ed889a6b9e1e64566217413d018cabd825f786e843ffe63e7"},
+  "hackney": {:hex, :hackney, "1.25.0", "390e9b83f31e5b325b9f43b76e1a785cbdb69b5b6cd4e079aa67835ded046867", [:rebar3], [{:certifi, "~> 2.15.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~> 6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~> 1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.4", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.4.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~> 0.7.1", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "7209bfd75fd1f42467211ff8f59ea74d6f2a9e81cbcee95a56711ee79fd6b1d4"},
+  "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"},
+  "jason": {:hex, :jason, "1.1.2", "b03dedea67a99223a2eaf9f1264ce37154564de899fd3d8b9a21b1a6fd64afe7", [:mix], [{:decimal, "~> 1.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "fdf843bca858203ae1de16da2ee206f53416bbda5dc8c9e78f43243de4bc3afe"},
+  "logger_file_backend": {:hex, :logger_file_backend, "0.0.14", "774bb661f1c3fed51b624d2859180c01e386eb1273dc22de4f4a155ef749a602", [:mix], [], "hexpm", "071354a18196468f3904ef09413af20971d55164267427f6257b52cfba03f9e6"},
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"},
-  "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm", "f278585650aa581986264638ebf698f8bb19df297f66ad91b18910dfc6e19323"},
-  "parse_trans": {:hex, :parse_trans, "3.3.0", "09765507a3c7590a784615cfd421d101aec25098d50b89d7aa1d66646bc571c1", [:rebar3], [], "hexpm", "17ef63abde837ad30680ea7f857dd9e7ced9476cdd7b0394432af4bfc241b960"},
+  "mimerl": {:hex, :mimerl, "1.5.0", "f35aca6f23242339b3666e0ac0702379e362b469d0aea167f6cc713547e777ed", [:rebar3], [], "hexpm", "db648ce065bae14ea84ca8b5dd123f42f49417cef693541110bf6f9e9be9ecc4"},
+  "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"},
+  "parse_trans": {:hex, :parse_trans, "3.4.1", "6e6aa8167cb44cc8f39441d05193be6e6f4e7c2946cb2759f015f8c56b76e5ff", [:rebar3], [], "hexpm", "620a406ce75dada827b82e453c19cf06776be266f5a67cff34e1ef2cbb60e49a"},
   "poison": {:hex, :poison, "3.1.0", "d9eb636610e096f86f25d9a46f35a9facac35609a7591b3be3326e99a0484665", [:mix], [], "hexpm"},
-  "poolboy": {:hex, :poolboy, "1.5.1", "6b46163901cfd0a1b43d692657ed9d7e599853b3b21b95ae5ae0a777cf9b6ca8", [:rebar], [], "hexpm", "8f7168911120e13419e086e78d20e4d1a6776f1eee2411ac9f790af10813389f"},
-  "redix": {:hex, :redix, "1.1.5", "6fc460d66a5c2287e83e6d73dddc8d527ff59cb4d4f298b41e03a4db8c3b2bd5", [:mix], [{:castore, "~> 0.1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "679afdd4c14502fe9c11387ff1cdcb33065a1cf511097da1eee407f17c7a418b"},
-  "retry": {:hex, :retry, "0.8.2", "7b57bd5e1e7efeca04dd740cabdc3930c472cfa7e0186949de180c64417e9c35", [:mix], [], "hexpm", "ae8969bfea65bf2adf9e7cc6e59bc2d1b6d9e04bcb7d2c53b996d9b8a023fe78"},
-  "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.4", "f0eafff810d2041e93f915ef59899c923f4568f4585904d010387ed74988e77b", [:make, :mix, :rebar3], [], "hexpm", "603561dc0fd62f4f2ea9b890f4e20e1a0d388746d6e20557cafb1b16950de88c"},
-  "telemetry": {:hex, :telemetry, "1.1.0", "a589817034a27eab11144ad24d5c0f9fab1f58173274b1e9bae7074af9cbee51", [:rebar3], [], "hexpm", "b727b2a1f75614774cff2d7565b64d0dfa5bd52ba517f16543e6fc7efcc0df48"},
-  "unicode_util_compat": {:hex, :unicode_util_compat, "0.4.1", "d869e4c68901dd9531385bb0c8c40444ebf624e60b6962d95952775cac5e90cd", [:rebar3], [], "hexpm", "1d1848c40487cdb0b30e8ed975e34e025860c02e419cb615d255849f3427439d"}
+  "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm", "dad79704ce5440f3d5a3681c8590b9dc25d1a561e8f5a9c995281012860901e3"},
+  "redix": {:hex, :redix, "1.6.0", "694179c7a3c71bffac8848fcbfe16f6f6e2a1e020f00a2ad01bc6484815470d1", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:nimble_options, "~> 0.5.0 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "b2eccb05e02f21c0c3ca57513e6bacb4dd48e6406dadbd7ff9fbe07bd6745999"},
+  "retry": {:hex, :retry, "0.16.0", "bda95a342de242088b2a68e4e826fd2a61f080b5fb2b596bec898e26520d262c", [:mix], [], "hexpm", "bebaa63ebf7c9ebc2741fcdbbda62cf49734a8740ce0a80708b669a595e6be8e"},
+  "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"},
+  "statistex": {:hex, :statistex, "1.1.1", "73612aa7f79e53c30569be065fd121e380f1cf57bc4c2da5b41be9246da18df9", [:mix], [], "hexpm", "310c4b49b34adf683de3103639006bed233ab54c08a4add65a531448e653857c"},
+  "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"},
+  "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.1", "a48703a25c170eedadca83b11e88985af08d35f37c6f664d6dcfb106a97782fc", [:rebar3], [], "hexpm", "b3a917854ce3ae233619744ad1e0102e05673136776fb2fa76234f3e03b23642"},
 }

test/flume_test.exs

@@ -57,8 +57,8 @@ defmodule FlumeTest do
     end
   end
 
-  @tag :slow_test
   describe "regular pipelines" do
+    @tag :slow_test
     test "pulls max_demand events from redis" do
       max_demand = 2
       sleep_time = 500
@@ -103,8 +103,8 @@ defmodule FlumeTest do
     end
   end
 
-  @tag :slow_test
   describe "rate-limited pipelines" do
+    @tag :slow_test
     test "pulls max_demand events from redis within the rate-limit-scale" do
       max_demand = 10
       sleep_time = 500