From ea6e919b3f77818baa3d1e5430b60b5011fabcbc Mon Sep 17 00:00:00 2001 From: Moises Macia Date: Mon, 8 Jun 2026 20:03:09 +0000 Subject: [PATCH] track feed fetch, parse and clean resources in sessions --- lib/feed_reader.ex | 255 +++++++++++++++++++++++++++-------- lib/feed_reader/collector.ex | 15 +++ lib/feed_reader/session.ex | 5 +- 3 files changed, 215 insertions(+), 60 deletions(-) diff --git a/lib/feed_reader.ex b/lib/feed_reader.ex index 5912490..1915be8 100644 --- a/lib/feed_reader.ex +++ b/lib/feed_reader.ex @@ -3,21 +3,101 @@ defmodule FeedReader do Standalone feed fetching and parsing library. This module exposes format detection, validation, local parsing, and streaming - helpers. + helpers. Remote sessions download and checksum into a temporary file before + exposing their one-pass event stream. """ @typedoc "`:rss`, `:atom`, `:rdf`, `:unknown`, or a custom format's atom ID." @type format_id :: atom() + @option_defaults [ + max_detection_bytes: 65_536, + chunk_size: 65_536, + max_xml_events: 1_000_000, + max_xml_depth: 256, + max_attributes_per_element: 128, + max_attribute_bytes: 65_536, + max_feed_metadata_bytes: 1_048_576, + max_item_bytes: 4_194_304, + http_retry_attempts: 5, + retry_delay: 0, + max_compressed_bytes: 52_428_800, + max_decoded_bytes: 52_428_800, + max_temp_bytes: 52_428_800, + max_expansion_ratio: 100 + ] + + @parsing_options_doc """ + - `:format` — format ID or `FeedReader.Format` module; defaults to detection. + - `:formats` — additional format modules, checked after RSS, Atom, and RDF. + - `:chunk_size` — positive XML input chunk size; defaults to + `#{@option_defaults[:chunk_size]}` bytes. + - `:max_detection_bytes` — positive detection prefix limit; defaults to + `#{@option_defaults[:max_detection_bytes]}` bytes. Invalid values disable + detection, so provide a valid explicit `:format` to parse in that case. + - `:feed_url` — source URL supplied to normalization; defaults to `nil` for + local sources and the final response URL for remote sources. + - `:max_xml_events`, `:max_xml_depth`, `:max_attributes_per_element`, + `:max_attribute_bytes`, `:max_feed_metadata_bytes`, and `:max_item_bytes` — + non-negative SAX and retained-data limits. Defaults are respectively + `#{@option_defaults[:max_xml_events]}`, `#{@option_defaults[:max_xml_depth]}`, + `#{@option_defaults[:max_attributes_per_element]}`, + `#{@option_defaults[:max_attribute_bytes]}`, + `#{@option_defaults[:max_feed_metadata_bytes]}`, and + `#{@option_defaults[:max_item_bytes]}`. Zero is valid; invalid values use + the defaults. + - `:max_queue_bytes` — non-negative maximum serialized size of one pending + event; defaults to `:max_item_bytes`. Zero is valid; invalid values use + that default. + """ + + @remote_options_doc """ + - `:http_client` — `FeedReader.Transport` implementation; defaults to + `FeedReader.ReqTransport`. + - `:http_options` — Req options forwarded to `Req.run/2`. The default transport + forces raw streaming and disables automatic decompression. `:headers` inside + this list replaces the default user-agent and accepted-compression headers. + - `:previous_etag` — binary ETag sent verbatim as `If-None-Match` on GET; a + 304 returns `:not_modified` without downloading a body. + - `:http_retry_attempts` — non-negative retries after the initial GET; + defaults to `#{@option_defaults[:http_retry_attempts]}`. The Req transport + retries status 400 and 500 and most transport errors; TLS and redirect + failures are terminal. Redirects are limited to five. + - `:retry_delay` — non-negative milliseconds or an arity-one delay function; + defaults to `#{@option_defaults[:retry_delay]}`. Invalid values use the + default. + - `:previous_checksum` — decoded-body checksum; a match returns + `:not_changed` and removes the downloaded file. + - `:max_compressed_bytes`, `:max_decoded_bytes`, and `:max_temp_bytes` — + non-negative raw-body, decoded-body, and temporary-file byte limits; + defaults are `#{@option_defaults[:max_compressed_bytes]}` bytes each. + - `:max_expansion_ratio` — non-negative decoded-to-compressed byte ratio; + defaults to `#{@option_defaults[:max_expansion_ratio]}`. Invalid limits use + their defaults. Checksums are computed over decoded bytes. + """ + + @collection_options_doc """ + - `:max_items` — required positive limit for accepted items; exceeding it + returns `{:error, {:max_items_exceeded, max_items}}` without partial data. + - `:item_filter` — optional one-argument function applied before counting and + retaining items; invalid values return `{:error, {:invalid_option, :item_filter}}`. + """ + + @format_options_doc """ + - `:max_detection_bytes` — positive byte limit for the inspected prefix; + defaults to `#{@option_defaults[:max_detection_bytes]}`. Invalid values yield + `:unknown`. + - `:formats` — custom format modules checked after RSS, Atom, and RDF in list + order. Duplicates are removed. Modules provide `matches_prefix?/1` and + `id/0`. + """ + @doc """ Detects a format from a bounded binary prefix. ## Options - - `:max_detection_bytes` — positive byte limit for the inspected prefix; - defaults to `65_536`. Invalid values yield `:unknown`. - - `:formats` — custom format modules checked after RSS, Atom, and RDF in list - order. Duplicates are removed. Modules provide `matches_prefix?/1` and - `id/0`. + + #{@format_options_doc} """ @spec detect_format(binary(), keyword()) :: format_id() def detect_format(content, opts \\ []) do @@ -28,11 +108,8 @@ defmodule FeedReader do Checks only the bounded-prefix signature for a format ID or module. ## Options - - `:max_detection_bytes` — positive byte limit for the inspected prefix; - defaults to `65_536`. Invalid values yield `false`. - - `:formats` — custom format modules used to resolve custom format IDs. - Modules provide `matches_prefix?/1` and `id/0`; passing the format module - directly also works. + + #{@format_options_doc} """ @spec validate(binary(), atom() | module(), keyword()) :: boolean() def validate(content, format, opts \\ []) do @@ -47,37 +124,7 @@ defmodule FeedReader do ## Options - - `:format` — an atom format ID (`:rss`, `:atom`, or `:rdf`) or a format - module. When omitted, the format is detected from the file prefix. - - `:formats` — a list of additional format modules. They are checked after - the built-in RSS, Atom, and RDF formats and may be selected with `:format`. - Custom modules must implement `FeedReader.Format`. - - `:chunk_size` — positive number of bytes supplied to the XML parser per - chunk. Defaults to `65_536`. - - `:max_detection_bytes` — positive byte limit for format detection. Defaults - to `65_536`. Invalid values produce `{:error, :unknown}` unless a valid - explicit `:format` is supplied. - - `:feed_url` — optional source URL passed to format normalizers and stored in - the normalized feed metadata. Defaults to `nil`. - - `:formats` — additional format modules checked after the built-in formats. - - `:max_xml_events` — non-negative maximum SAX events. Defaults to - `1_000_000`. - - `:max_xml_depth` — non-negative maximum XML nesting depth. Defaults to - `256`. - - `:max_attributes_per_element` — non-negative maximum attributes on one - element. Defaults to `128`. - - `:max_attribute_bytes` — non-negative maximum bytes used by one element's - attributes. Defaults to `65_536`. - - `:max_feed_metadata_bytes` — non-negative maximum retained feed metadata - bytes. Defaults to `1_048_576`. - - `:max_item_bytes` — non-negative maximum retained bytes for one item. - Defaults to `4_194_304`. - - `:max_queue_bytes` — non-negative maximum serialized size of one pending - event. Defaults to `:max_item_bytes`. - - Zero is valid for resource limits. Missing or invalid resource-limit values - use their defaults. Additional options are passed through to the selected - format implementation. + #{String.trim(@parsing_options_doc)} """ @spec stream_file(Path.t(), keyword()) :: {:ok, FeedReader.Session.t()} | {:error, term()} def stream_file(path, opts \\ []) do @@ -85,23 +132,113 @@ defmodule FeedReader do end @doc """ - Collects a caller-owned local XML file into a bounded parse result. + Downloads and checksums a feed before returning a lazy event session. - `:max_items` is required and must be a positive integer. The parser continues - through the end of the source so malformed tails are reported; an item - accepted beyond the limit returns `{:error, {:max_items_exceeded, max_items}}`. - `:item_filter`, when supplied, must be a one-argument function and is applied - before counting and collecting items. + The session emits feed metadata followed by items from one local XML pass. + Call `close/1` in an `after` block if enumeration may stop early. Matching + previous checksums or a conditional HTTP 304 return `:not_changed` or + `:not_modified` without a session. ## Options - Accepts all `stream_file/2` parsing options, plus: + #{String.trim(@remote_options_doc)} - - `:max_items` — required positive integer limit for accepted items. The - collector continues parsing after reaching the limit and returns - `{:error, {:max_items_exceeded, max_items}}` if another item is accepted. - - `:item_filter` — optional one-argument function applied before counting and - retaining an item. + #{String.trim(@parsing_options_doc)} + """ + @spec stream(String.t(), keyword()) :: + {:ok, FeedReader.Session.t() | :not_changed | :not_modified} | {:error, term()} + def stream(url, opts \\ []) do + case FeedReader.Downloader.download(url, opts) do + {:ok, %{path: _path, fetch: _fetch} = download} -> + create_session(download, opts) + + result -> + result + end + end + + defp create_session(%{path: path, fetch: fetch} = download, opts) do + parser_opts = Keyword.put_new(opts, :feed_url, fetch.final_url) + + case FeedReader.Parser.stream_file(path, parser_opts) do + {:ok, session} -> + events = + Stream.transform( + session.events, + fn -> :ok end, + fn event, state -> {[event], state} end, + fn _state -> + FeedReader.Downloader.close(download) + [] + end + ) + + {:ok, %FeedReader.Session{events: events, fetch: fetch, resource: download}} + + {:error, reason} -> + FeedReader.Downloader.close(download) + {:error, reason} + end + end + + @doc "Closes a remote session; safe after completion or repeated calls." + @spec close(FeedReader.Session.t()) :: :ok + def close(%FeedReader.Session{resource: %{path: _path} = resource}) do + FeedReader.Downloader.close(resource) + end + + def close(%FeedReader.Session{}), do: :ok + + @doc """ + Downloads and collects a feed, requiring a finite positive `:max_items`. + + Memory use is proportional to the number and size of collected items. The + collector scans through the terminal parser event and returns no partial + result on an item-limit breach or parse error. + + ## Options + + #{String.trim(@remote_options_doc)} + + #{String.trim(@parsing_options_doc)} + + #{String.trim(@collection_options_doc)} + """ + @spec fetch_and_parse(String.t(), keyword()) :: + {:ok, FeedReader.RemoteResult.t() | :not_changed | :not_modified} | {:error, term()} + def fetch_and_parse(url, opts \\ []) do + with {:ok, _max_items} <- FeedReader.Collector.max_items(opts), + {:ok, outcome} <- stream(url, opts) do + collect_remote_result(outcome, opts) + end + end + + defp collect_remote_result(outcome, _opts) when outcome in [:not_changed, :not_modified], + do: {:ok, outcome} + + defp collect_remote_result(%FeedReader.Session{} = session, opts) do + with {:ok, result} <- FeedReader.Collector.collect_session(session, opts) do + {:ok, + %FeedReader.RemoteResult{ + feed: result.feed, + items: result.items, + fetch: session.fetch + }} + end + end + + @doc """ + Collects a caller-owned local XML file into a bounded parse result. + + The parser continues through the end of the source so malformed tails are + reported. An item accepted beyond the limit returns + `{:error, {:max_items_exceeded, max_items}}` without partial data. + + ## Options + + #{String.trim(@parsing_options_doc)} + + #{String.trim(@collection_options_doc)} """ @spec parse_file(Path.t(), keyword()) :: {:ok, FeedReader.ParseResult.t()} | {:error, term()} def parse_file(path, opts \\ []) do @@ -113,8 +250,9 @@ defmodule FeedReader do ## Options - Accepts the same parsing and collection options as `parse_file/2`, including - required `:max_items` and optional `:item_filter`. + #{String.trim(@parsing_options_doc)} + + #{String.trim(@collection_options_doc)} """ @spec parse_binary(binary(), keyword()) :: {:ok, FeedReader.ParseResult.t()} | {:error, term()} @@ -135,8 +273,9 @@ defmodule FeedReader do ## Options - Accepts the same parsing and collection options as `parse_file/2`, including - required `:max_items` and optional `:item_filter`. + #{String.trim(@parsing_options_doc)} + + #{String.trim(@collection_options_doc)} """ @spec parse_stream(Enumerable.t(), keyword()) :: {:ok, FeedReader.ParseResult.t()} | {:error, term()} diff --git a/lib/feed_reader/collector.ex b/lib/feed_reader/collector.ex index 55680ac..9b36f60 100644 --- a/lib/feed_reader/collector.ex +++ b/lib/feed_reader/collector.ex @@ -30,6 +30,21 @@ defmodule FeedReader.Collector do end end + @doc "Collects a session and releases its owned source on every outcome." + @spec collect_session(FeedReader.Session.t(), keyword()) :: + {:ok, ParseResult.t()} | {:error, term()} + def collect_session(session, opts) do + collect_session_events(session, opts) + after + FeedReader.close(session) + end + + defp collect_session_events(session, opts) do + with {:ok, max_items} <- max_items(opts) do + collect_events(session.events, max_items, opts) + end + end + @doc """ Collects XML chunks from an enumerable. diff --git a/lib/feed_reader/session.ex b/lib/feed_reader/session.ex index f3976b8..3705d0a 100644 --- a/lib/feed_reader/session.ex +++ b/lib/feed_reader/session.ex @@ -17,8 +17,9 @@ defmodule FeedReader.Session do @type t :: %__MODULE__{ events: Enumerable.t(), - fetch: Fetch.t() | nil + fetch: Fetch.t() | nil, + resource: map() | nil } - defstruct events: nil, fetch: nil + defstruct events: nil, fetch: nil, resource: nil end