diff --git a/test/feed_reader/streaming_test.exs b/test/feed_reader/streaming_test.exs
index a446ef6..16adc4d 100644
--- a/test/feed_reader/streaming_test.exs
+++ b/test/feed_reader/streaming_test.exs
@@ -2,6 +2,7 @@ defmodule FeedReader.StreamingTest do
use ExUnit.Case, async: true
alias FeedReader.{Feed, Item, Session}
+ import ExUnit.CaptureLog
@xml ~s(Examplehttps://example.com/- Firstone
- Secondtwo
)
@atom_xml ~s(Atom)
@@ -110,6 +111,41 @@ defmodule FeedReader.StreamingTest do
assert File.read!(path) == @xml
end
+ test "does not retain an event larger than the pending queue budget" do
+ path = write_temp(@xml)
+ on_exit(fn -> File.rm(path) end)
+
+ assert {:ok, %{events: events}} = FeedReader.stream_file(path, max_queue_bytes: 1)
+ assert [{:error, {:resource_limit, :max_queue_bytes, 1}}] = Enum.to_list(events)
+ end
+
+ test "cancelling enumeration stops before the next item is parsed" do
+ path = write_temp(@xml)
+ on_exit(fn -> File.rm(path) end)
+
+ assert {:ok, %{events: events}} = FeedReader.stream_file(path)
+
+ assert [{:feed, %Feed{}}, {:item, %Item{uuid: "one"}}] =
+ events |> Stream.take(2) |> Enum.to_list()
+ end
+
+ test "cleans up the parser when the consumer raises" do
+ path = write_temp(@xml)
+ on_exit(fn -> File.rm(path) end)
+
+ assert {:ok, %{events: events}} = FeedReader.stream_file(path)
+
+ _log =
+ capture_log(fn ->
+ {:ok, consumer} =
+ Task.start(fn -> Enum.each(events, fn _event -> raise "consumer failed" end) end)
+
+ monitor = Process.monitor(consumer)
+
+ assert_receive {:DOWN, ^monitor, :process, ^consumer, {%RuntimeError{}, _stack}}
+ end)
+ end
+
test "returns a stable error for a non-atom format option" do
path = write_temp(@xml)
on_exit(fn -> File.rm(path) end)
@@ -124,6 +160,16 @@ defmodule FeedReader.StreamingTest do
assert {:error, :unknown} = FeedReader.stream_file(path, max_detection_bytes: 0)
end
+ test "uses queue defaults when queue limits are invalid" do
+ path = write_temp(@xml)
+ on_exit(fn -> File.rm(path) end)
+
+ assert {:ok, %{events: events}} =
+ FeedReader.stream_file(path, max_item_bytes: :invalid, max_queue_bytes: :invalid)
+
+ assert {:feed, %Feed{title: "Example"}} = Enum.at(Enum.to_list(events), 0)
+ end
+
test "delivers a parser halt as a terminal error event" do
path = write_temp(@xml)
on_exit(fn -> File.rm(path) end)
@@ -152,6 +198,48 @@ defmodule FeedReader.StreamingTest do
assert [] = Enum.to_list(events)
end
+ test "delivers an abnormal parser exit as an error event" do
+ path = write_temp(@xml)
+ on_exit(fn -> File.rm(path) end)
+
+ assert {:ok, %{events: events}} =
+ FeedReader.stream_file(path,
+ format: :parser_halt,
+ formats: [FeedReader.TestSupport.ParserHaltFormat],
+ abnormal_exit: true
+ )
+
+ assert [{:error, :parser_crashed}] = Enum.to_list(events)
+ end
+
+ test "stops the parser when the consuming process dies" do
+ path = write_temp("")
+ on_exit(fn -> File.rm(path) end)
+ test_pid = self()
+
+ assert {:ok, %{events: events}} =
+ FeedReader.stream_file(path,
+ format: :external_test,
+ formats: [FeedReader.TestSupport.ExternalFormat],
+ notify_parser: test_pid,
+ emit_event: true
+ )
+
+ consumer =
+ spawn(fn ->
+ Enum.each(events, fn event ->
+ send(test_pid, {:event, event})
+ Process.sleep(:infinity)
+ end)
+ end)
+
+ assert_receive {:parser_pid, parser}, 500
+ assert_receive {:event, {:feed, %Feed{title: "External"}}}, 500
+ parser_monitor = Process.monitor(parser)
+ Process.exit(consumer, :kill)
+ assert_receive {:DOWN, ^parser_monitor, :process, ^parser, :normal}, 500
+ end
+
defp write_temp(content) do
path = Path.join(System.tmp_dir!(), "feed_reader-#{System.unique_integer([:positive])}.xml")
File.write!(path, content)
diff --git a/test/support/external_format.ex b/test/support/external_format.ex
index f3f1651..e9c94f9 100644
--- a/test/support/external_format.ex
+++ b/test/support/external_format.ex
@@ -15,7 +15,13 @@ defmodule FeedReader.TestSupport.ExternalFormat do
def handler, do: FeedReader.TestSupport.ExternalFormatHandler
@impl true
- def new_parser_state(event_sink, opts), do: %{event_sink: event_sink, opts: opts}
+ def new_parser_state(event_sink, opts) do
+ if notify = Keyword.get(opts, :notify_parser) do
+ send(notify, {:parser_pid, self()})
+ end
+
+ %{event_sink: event_sink, opts: opts}
+ end
@impl true
def normalize_feed(_state, _opts), do: %Feed{title: "External", type: :rss}
diff --git a/test/support/external_format_handler.ex b/test/support/external_format_handler.ex
index 1d70649..22b8591 100644
--- a/test/support/external_format_handler.ex
+++ b/test/support/external_format_handler.ex
@@ -2,6 +2,22 @@ defmodule FeedReader.TestSupport.ExternalFormatHandler do
@moduledoc false
@behaviour Saxy.Handler
+ alias FeedReader.Feed
+
@impl true
+ def handle_event(:start_document, _data, %{opts: opts} = state) do
+ if Keyword.get(opts, :emit_event), do: emit_event(state)
+
+ {:ok, state}
+ end
+
+ def handle_event(:end_document, _data, %{opts: opts} = state) do
+ if Keyword.get(opts, :emit_event), do: emit_event(state)
+
+ {:ok, state}
+ end
+
def handle_event(_event_type, _data, state), do: {:ok, state}
+
+ defp emit_event(state), do: state.event_sink.({:feed, %Feed{title: "External", type: :rss}})
end
diff --git a/test/support/parser_halt_format.ex b/test/support/parser_halt_format.ex
index 1058bfc..58da0d9 100644
--- a/test/support/parser_halt_format.ex
+++ b/test/support/parser_halt_format.ex
@@ -13,6 +13,7 @@ defmodule FeedReader.TestSupport.ParserHaltFormat do
@impl true
def new_parser_state(_event_sink, opts) do
if Keyword.get(opts, :normal_exit), do: exit(:normal)
+ if Keyword.get(opts, :abnormal_exit), do: exit(:parser_crashed)
nil
end