Fix bug in expiry/3 (#46)

* Fix bug in expiry/3

This fix changes expiry/3 in such a way that the expiry time is
evaluated at runtime of the stream, rather than definition time.
This grants greater flexibility and enables setting up a configured
stream in advance (e.g. at compile time or in a gen_server init)
and passing the stream around for use in various places, at various
times.

See the added tests for an example.

* Refactor expiry/3 to make credo happy

Co-authored-by: Joshua Trees <me@jtrees.io>

Author: jtrees

lib/retry/delay_streams.ex

@@ -159,25 +159,34 @@ defmodule Retry.DelayStreams do
   """
   @spec expiry(Enumerable.t(), pos_integer(), pos_integer()) :: Enumerable.t()
   def expiry(delays, time_budget, min_delay \\ 100) do
-    end_t = :os.system_time(:milli_seconds) + time_budget
-
-    Stream.transform(delays, :normal, fn preferred_delay, status ->
-      now_t = :os.system_time(:milli_seconds)
-      remaining_t = Enum.max([end_t - now_t, min_delay])
+    Stream.resource(
+      fn -> {delays, :os.system_time(:milli_seconds) + time_budget} end,
+      fn
+        :at_end -> {:halt, :at_end}
+        {remaining_delays, end_t} -> reduce_delays(remaining_delays, end_t, min_delay)
+      end,
+      fn _ -> :noop end
+    )
+  end
 
-      cond do
-        # time expired!
-        :at_end == status ->
-          {:halt, status}
+  defp reduce_delays(remaining_delays, end_t, min_delay) do
+    case Enum.take(remaining_delays, 1) do
+      [preferred_delay] ->
+        now_t = :os.system_time(:milli_seconds)
+        remaining_t = Enum.max([end_t - now_t, min_delay])
 
-        # one last try
-        preferred_delay >= remaining_t or remaining_t == min_delay ->
+        if preferred_delay >= remaining_t or remaining_t == min_delay do
+          # one last try before time budget is exceeded
           {[remaining_t], :at_end}
-
-        true ->
-          {[preferred_delay], status}
-      end
-    end)
+        else
+          # default
+          {[preferred_delay], {Stream.drop(remaining_delays, 1), end_t}}
+        end
+
+      _ ->
+        # reached end of stream - no more tries
+        {:halt, :at_end}
+    end
   end
 
   defp random_uniform(n) when n <= 0, do: 0

test/retry/delay_streams_test.exs

@@ -99,6 +99,24 @@ defmodule Retry.DelayStreamsTest do
              end)
              |> Enum.min() >= 10
     end
+
+    test "stops before the expiry is reached, if there are no elements left in the delay stream" do
+      assert exponential_backoff() |> Stream.take(5) |> expiry(1_000) |> Enum.count() == 5
+    end
+
+    test "evaluates expiry at runtime of the stream, rather than execution time of expiry/3" do
+      delay_stream =
+        [50]
+        |> Stream.cycle()
+        |> expiry(75, 50)
+
+      assert Enum.count(delay_stream, fn delay -> :timer.sleep(delay) end) == 2
+
+      # If the expiry time is determined when `expiry/3` is executed, we will
+      # already be past it (due to all the sleeps performed previously) and jump
+      # out after the first run.
+      assert Enum.count(delay_stream, fn delay -> :timer.sleep(delay) end) == 2
+    end
   end
 
   describe "randomize/2" do