diff --git a/.tool-versions b/.tool-versions index 4bddf72..1fe8f91 100644 --- a/.tool-versions +++ b/.tool-versions @@ -1,2 +1,2 @@ -erlang 27.2 +erlang 27.3 elixir 1.18.0-otp-27 \ No newline at end of file diff --git a/lib/data_buffer.ex b/lib/data_buffer.ex index 1d6133d..18e4bcd 100644 --- a/lib/data_buffer.ex +++ b/lib/data_buffer.ex @@ -64,6 +64,7 @@ defmodule DataBuffer do - `:flush_jitter` - Random jitter added to flush_interval (default: 2000) - `:flush_timeout` - Timeout in ms for flush operations (default: 60000) - `:flush_meta` - Metadata passed to handle_flush callback (optional) + - `:shutdown_timeout` - Timeout in ms or `:infinity` for shutdown flush operations (default: 5000) ## Telemetry Events @@ -297,6 +298,7 @@ defmodule DataBuffer do - `:flush_interval` - Time interval for automatic flush - `:flush_jitter` - Random jitter for flush interval - `:flush_timeout` - Timeout for flush operations + - `:shutdown_timeout` - Timeout for shutdown flush operations ## Examples diff --git a/lib/data_buffer/partition.ex b/lib/data_buffer/partition.ex index e75d55e..8b57ce1 100644 --- a/lib/data_buffer/partition.ex +++ b/lib/data_buffer/partition.ex @@ -19,6 +19,7 @@ defmodule DataBuffer.Partition do :flush_interval, :flush_jitter, :flush_timeout, + :shutdown_timeout, :flush_opts, :flush_complete_ref, :flush_schedule_ref, @@ -42,7 +43,8 @@ defmodule DataBuffer.Partition do flush_interval: [is: :integer, default: 10_000, required: true], flush_jitter: [is: :integer, default: 2_000, required: true], flush_meta: [is: :any, required: false], - flush_timeout: [is: :timeout, default: 60_000, required: true] + flush_timeout: [is: :timeout, default: 60_000, required: true], + shutdown_timeout: [is: :timeout, default: 5_000, required: true] ) ################################ @@ -53,7 +55,8 @@ defmodule DataBuffer.Partition do def child_spec(opts) do %{ id: Keyword.fetch!(opts, :name), - start: {__MODULE__, :start_link, [opts]} + start: {__MODULE__, :start_link, [opts]}, + shutdown: Keyword.get(opts, :shutdown_timeout, 5_000) } end @@ -192,6 +195,7 @@ defmodule DataBuffer.Partition do flush_interval: Keyword.get(opts, :flush_interval), flush_jitter: Keyword.get(opts, :flush_jitter), flush_timeout: Keyword.get(opts, :flush_timeout), + shutdown_timeout: Keyword.get(opts, :shutdown_timeout), flush_opts: [ meta: Keyword.get(opts, :flush_meta) ], @@ -369,6 +373,7 @@ defmodule DataBuffer.Partition do flush_interval: state.flush_interval, flush_jitter: state.flush_jitter, flush_timeout: state.flush_timeout, + shutdown_timeout: state.shutdown_timeout, pid: self() } end diff --git a/mix.exs b/mix.exs index 30caeb3..0b474d0 100644 --- a/mix.exs +++ b/mix.exs @@ -1,7 +1,7 @@ defmodule DataBuffer.MixProject do use Mix.Project - @version "1.0.0" + @version "1.1.0" def project do [ diff --git a/test/data_buffer_test.exs b/test/data_buffer_test.exs index 23dd6dd..6af4655 100644 --- a/test/data_buffer_test.exs +++ b/test/data_buffer_test.exs @@ -203,6 +203,43 @@ defmodule DataBufferTest do end) =~ "DataBuffer: flush timeout error" end + test "flushes data on shutdown" do + start_buffer(partitions: 1) + DataBuffer.insert(TestBuffer, "foo") + stop_supervised!(TestBuffer) + assert_receive {:data, ["foo"], _} + end + + test "configures default and custom shutdown_timeout on partition" do + start_buffer(partitions: 1) + [partition] = DataBuffer.info(TestBuffer) + assert partition.shutdown_timeout == 5_000 + + spec = + DataBuffer.Partition.child_spec( + name: :test_part, + buffer: TestBuffer, + shutdown_timeout: 12_345 + ) + + assert spec.shutdown == 12_345 + + default_spec = DataBuffer.Partition.child_spec(name: :test_part, buffer: TestBuffer) + assert default_spec.shutdown == 5_000 + + stop_supervised!(TestBuffer) + + start_buffer(partitions: 1, shutdown_timeout: 10_000) + [partition] = DataBuffer.info(TestBuffer) + assert partition.shutdown_timeout == 10_000 + stop_supervised!(TestBuffer) + + start_buffer(partitions: 1, shutdown_timeout: :infinity) + [partition] = DataBuffer.info(TestBuffer) + assert partition.shutdown_timeout == :infinity + stop_supervised!(TestBuffer) + end + defp receive_all(partitions \\ partitions()) do for _ <- 1..partitions, reduce: [] do data ->