diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 84ce844..633f7ff 100644 --- a/.github/workflows/main.yml +++ b/.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 diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 287729b..70fe9e3 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -1,4 +1,4 @@ -image: elixir:1.6.6-alpine +image: elixir:1.18-alpine cache: key: flume paths: diff --git a/README.md b/README.md index e21550f..e10a8ba 100644 --- a/README.md +++ b/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 diff --git a/lib/flume/bulk_event.ex b/lib/flume/bulk_event.ex index 8c04b97..4fb1993 100644 --- a/lib/flume/bulk_event.ex +++ b/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", diff --git a/lib/flume/pipeline/control/options.ex b/lib/flume/pipeline/control/options.ex index 7b29d14..7c03eb2 100644 --- a/lib/flume/pipeline/control/options.ex +++ b/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 diff --git a/lib/flume/pipeline/event.ex b/lib/flume/pipeline/event.ex index ec83d6f..214dfa9 100644 --- a/lib/flume/pipeline/event.ex +++ b/lib/flume/pipeline/event.ex @@ -39,7 +39,7 @@ 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 @@ -47,7 +47,7 @@ defmodule Flume.Pipeline.Event do end def resume(pipeline_name, opts) do - unless opts[:temporary] do + if !opts[:temporary] do RedisClient.del(paused_redis_key(pipeline_name)) end diff --git a/lib/flume/redis/bulk_dequeue.ex b/lib/flume/redis/bulk_dequeue.ex index 84b3085..3f97227 100644 --- a/lib/flume/redis/bulk_dequeue.ex +++ b/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, []} diff --git a/lib/flume/support/time.ex b/lib/flume/support/time.ex index 7b29eb9..36b0d09 100644 --- a/lib/flume/support/time.ex +++ b/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 diff --git a/mix.exs b/mix.exs index f3d8915..1e88298 100644 --- a/mix.exs +++ b/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} diff --git a/mix.lock b/mix.lock index c93abf2..1dbd7d3 100644 --- a/mix.lock +++ b/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"}, } diff --git a/test/flume_test.exs b/test/flume_test.exs index 983bfe6..58cb6b1 100644 --- a/test/flume_test.exs +++ b/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