Bedrock.DataPlane.Log.Shale.TransactionStreams (bedrock v0.6.0)

View Source

A module for handling transaction streams with operations like limiting, filtering, and halting based on conditions.

Summary

Functions

at_most(stream, limit)

Limits the number of transactions in the stream based on the given version.

from_file!(path_to_file)

Streams transactions from the segment.

from_list_of_transactions(transactions_fn)
from_segments(segments, target_version)
read_previous_version(path)

See Bedrock.DataPlane.Log.Shale.WalFormat.previous_version/1.

until_version(stream, last_version)

Functions

at_most(stream, limit)

@spec at_most(Enumerable.t(), pos_integer()) :: Enumerable.t()

Limits the number of transactions in the stream based on the given version.

from_file!(path_to_file)

@spec from_file!(path_to_file :: String.t()) ::
  Enumerable.t(
    {Bedrock.DataPlane.Transaction.encoded() | :eof | :corrupted,
     non_neg_integer()}
  )

Streams transactions from the segment.

This function returns a Stream that iterates through the transactions in the given segment. Each transaction is validated using its checksum (CRC32), and the stream yields either valid transactions, identifies end-of-file markers, or flags corrupted data. Offsets are tracked so that append operations can be performed safely.

from_list_of_transactions(transactions_fn)

@spec from_list_of_transactions((-> [Bedrock.DataPlane.Transaction.encoded()] | nil)) ::
  Enumerable.t(Bedrock.DataPlane.Transaction.encoded())

from_segments(segments, target_version)

@spec from_segments([Bedrock.DataPlane.Log.Shale.Segment.t()], Bedrock.version()) ::
  {:ok, Enumerable.t(Bedrock.DataPlane.Transaction.encoded())}
  | {:error, :not_found}

read_previous_version(path)

@spec read_previous_version(String.t()) ::
  {:ok, Bedrock.version()}
  | {:error, :unsupported_wal_format | :invalid_wal_format | File.posix()}

See Bedrock.DataPlane.Log.Shale.WalFormat.previous_version/1.

until_version(stream, last_version)

@spec until_version(Enumerable.t(), Bedrock.version()) :: Enumerable.t()