ObjectStoreX.Stream (ObjectStoreX v0.2.0)

Streaming operations for large files.

This module provides streaming download capabilities that allow you to process large files without loading them entirely into memory.

Examples

# Download a large file and write to disk
stream = ObjectStoreX.Stream.download(store, "large-file.bin")

File.open!("output.bin", [:write], fn file ->
  stream
  |> Stream.each(&IO.binwrite(file, &1))
  |> Stream.run()
end)

# Process chunks
total_bytes =
  stream
  |> Stream.map(&byte_size/1)
  |> Enum.sum()

Summary

Types

path()

store()

Functions

download(store, path, opts \\ [])

Create an Elixir Stream for downloading a large object.

list_stream(store, opts \\ [])

List objects as a stream with automatic pagination.

upload(stream, store, path, opts \\ [])

Upload a large file using streaming multipart upload.

Types

path()

@type path() :: String.t()

store()

@type store() :: reference()

Functions

download(store, path, opts \\ [])

@spec download(store(), path(), keyword()) :: Enumerable.t()

Create an Elixir Stream for downloading a large object.

The stream will emit binary chunks as they are received from the object store. The chunks are yielded in order and the stream completes when the entire object has been downloaded.

Options

:timeout - Timeout in milliseconds for receiving each chunk (default: 30_000)

Examples

stream = ObjectStoreX.Stream.download(store, "large-file.bin")

# Write to file
File.open!("output.bin", [:write], fn file ->
  stream |> Stream.each(&IO.binwrite(file, &1)) |> Stream.run()
end)

# Count bytes
total_bytes = stream |> Stream.map(&byte_size/1) |> Enum.sum()

Error Handling

If an error occurs during streaming, the stream will raise an exception.

list_stream(store, opts \\ [])

@spec list_stream(
  store(),
  keyword()
) :: Enumerable.t()

List objects as a stream with automatic pagination.

Returns a stream that yields object metadata maps. The stream automatically handles pagination and will continue until all objects matching the prefix have been returned.

Options

:prefix - Optional prefix to filter objects (default: nil, lists all objects)
:timeout - Timeout in milliseconds for receiving each object (default: 30_000)

Examples

# List all objects with prefix
ObjectStoreX.Stream.list_stream(store, prefix: "data/2025/")
|> Stream.map(& &1.location)
|> Enum.take(100)

# Filter by size
ObjectStoreX.Stream.list_stream(store, prefix: "logs/")
|> Stream.filter(fn meta -> meta.size > 1_000_000 end)
|> Stream.map(& &1.location)
|> Enum.to_list()

# Process in batches
ObjectStoreX.Stream.list_stream(store)
|> Stream.chunk_every(100)
|> Stream.each(&process_batch/1)
|> Stream.run()

Metadata Structure

Each object metadata map contains:

:location - String path of the object
:size - Size in bytes
:last_modified - ISO8601 timestamp string
:etag - Optional ETag string
:version - Optional version string

Error Handling

If an error occurs during listing, the stream will raise an exception.

upload(stream, store, path, opts \\ [])

@spec upload(Enumerable.t(), store(), path(), keyword()) :: :ok | {:error, term()}

Upload a large file using streaming multipart upload.

The function consumes an Elixir Stream and uploads its data in chunks, using multipart upload behind the scenes. This allows uploading large files without loading them entirely into memory.

Options

None currently supported.

Examples

# Upload from file stream
File.stream!("large-file.bin", [], 10_485_760)  # 10MB chunks
|> ObjectStoreX.Stream.upload(store, "destination.bin")

# Upload from generated data
Stream.repeatedly(fn -> :crypto.strong_rand_bytes(1024) end)
|> Stream.take(10_000)  # ~10MB total
|> ObjectStoreX.Stream.upload(store, "random.dat")

Error Handling

If an error occurs during upload, the multipart upload will be aborted automatically and an error tuple will be returned.