track feed fetch, parse and clean resources in sessions

This commit is contained in:
mmacia committed 2026-06-08 20:03:09 +00:00
1 parent 0cddbfc76f
commit ea6e919b3f
3 files changed
+215 -60

No files matched your search

+197 -58
View File
@@ -3,21 +3,101 @@ defmodule FeedReader do
Standalone feed fetching and parsing library. Standalone feed fetching and parsing library.
This module exposes format detection, validation, local parsing, and streaming 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." @typedoc "`:rss`, `:atom`, `:rdf`, `:unknown`, or a custom format's atom ID."
@type format_id :: atom() @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 """ @doc """
Detects a format from a bounded binary prefix. Detects a format from a bounded binary prefix.
## Options ## Options
- `:max_detection_bytes` — positive byte limit for the inspected prefix;
defaults to `65_536`. Invalid values yield `:unknown`. #{@format_options_doc}
- `:formats` — custom format modules checked after RSS, Atom, and RDF in list
order. Duplicates are removed. Modules provide `matches_prefix?/1` and
`id/0`.
""" """
@spec detect_format(binary(), keyword()) :: format_id() @spec detect_format(binary(), keyword()) :: format_id()
def detect_format(content, opts \\ []) do 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. Checks only the bounded-prefix signature for a format ID or module.
## Options ## Options
- `:max_detection_bytes` — positive byte limit for the inspected prefix;
defaults to `65_536`. Invalid values yield `false`. #{@format_options_doc}
- `: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.
""" """
@spec validate(binary(), atom() | module(), keyword()) :: boolean() @spec validate(binary(), atom() | module(), keyword()) :: boolean()
def validate(content, format, opts \\ []) do def validate(content, format, opts \\ []) do
@@ -47,37 +124,7 @@ defmodule FeedReader do
## Options ## Options
- `:format` — an atom format ID (`:rss`, `:atom`, or `:rdf`) or a format #{String.trim(@parsing_options_doc)}
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.
""" """
@spec stream_file(Path.t(), keyword()) :: {:ok, FeedReader.Session.t()} | {:error, term()} @spec stream_file(Path.t(), keyword()) :: {:ok, FeedReader.Session.t()} | {:error, term()}
def stream_file(path, opts \\ []) do def stream_file(path, opts \\ []) do
@@ -85,23 +132,113 @@ defmodule FeedReader do
end end
@doc """ @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 The session emits feed metadata followed by items from one local XML pass.
through the end of the source so malformed tails are reported; an item Call `close/1` in an `after` block if enumeration may stop early. Matching
accepted beyond the limit returns `{:error, {:max_items_exceeded, max_items}}`. previous checksums or a conditional HTTP 304 return `:not_changed` or
`:item_filter`, when supplied, must be a one-argument function and is applied `:not_modified` without a session.
before counting and collecting items.
## Options ## Options
Accepts all `stream_file/2` parsing options, plus: #{String.trim(@remote_options_doc)}
- `:max_items` — required positive integer limit for accepted items. The #{String.trim(@parsing_options_doc)}
collector continues parsing after reaching the limit and returns """
`{:error, {:max_items_exceeded, max_items}}` if another item is accepted. @spec stream(String.t(), keyword()) ::
- `:item_filter` — optional one-argument function applied before counting and {:ok, FeedReader.Session.t() | :not_changed | :not_modified} | {:error, term()}
retaining an item. 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()} @spec parse_file(Path.t(), keyword()) :: {:ok, FeedReader.ParseResult.t()} | {:error, term()}
def parse_file(path, opts \\ []) do def parse_file(path, opts \\ []) do
@@ -113,8 +250,9 @@ defmodule FeedReader do
## Options ## Options
Accepts the same parsing and collection options as `parse_file/2`, including #{String.trim(@parsing_options_doc)}
required `:max_items` and optional `:item_filter`.
#{String.trim(@collection_options_doc)}
""" """
@spec parse_binary(binary(), keyword()) :: @spec parse_binary(binary(), keyword()) ::
{:ok, FeedReader.ParseResult.t()} | {:error, term()} {:ok, FeedReader.ParseResult.t()} | {:error, term()}
@@ -135,8 +273,9 @@ defmodule FeedReader do
## Options ## Options
Accepts the same parsing and collection options as `parse_file/2`, including #{String.trim(@parsing_options_doc)}
required `:max_items` and optional `:item_filter`.
#{String.trim(@collection_options_doc)}
""" """
@spec parse_stream(Enumerable.t(), keyword()) :: @spec parse_stream(Enumerable.t(), keyword()) ::
{:ok, FeedReader.ParseResult.t()} | {:error, term()} {:ok, FeedReader.ParseResult.t()} | {:error, term()}
+15
View File
@@ -30,6 +30,21 @@ defmodule FeedReader.Collector do
end end
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 """ @doc """
Collects XML chunks from an enumerable. Collects XML chunks from an enumerable.
+3 -2
View File
@@ -17,8 +17,9 @@ defmodule FeedReader.Session do
@type t :: %__MODULE__{ @type t :: %__MODULE__{
events: Enumerable.t(), 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 end