Streaming Guide

This guide covers efficient handling of large files using ObjectStoreX streaming capabilities.

Why Streaming?
Streaming Downloads
Streaming Uploads
Listing APIs
Streaming Lists
Range Reads
Best Practices
Performance Tuning

Why Streaming?

Streaming is essential for handling large files without loading them entirely into memory:

Memory efficient: Process files larger than available RAM
Faster time-to-first-byte: Start processing before full download completes
Better resource utilization: Constant memory usage regardless of file size
Scalability: Handle thousands of concurrent operations

Streaming Downloads

Download large files in chunks without loading the entire file into memory.

Basic Download Stream

alias ObjectStoreX.Stream

# Create a download stream
stream = Stream.download(store, "large-file.bin")

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

Process Chunks

# Count total bytes
total_bytes =
  Stream.download(store, "large-file.bin")
  |> Stream.map(&byte_size/1)
  |> Enum.sum()

IO.puts "Total size: #{total_bytes} bytes"

With Progress Tracking

defmodule Downloader do
  def download_with_progress(store, remote_path, local_path) do
    {:ok, meta} = ObjectStoreX.head(store, remote_path)
    total_size = meta.size
    downloaded = 0

    File.open!(local_path, [:write], fn file ->
      ObjectStoreX.Stream.download(store, remote_path)
      |> Stream.each(fn chunk ->
        IO.binwrite(file, chunk)
        downloaded = downloaded + byte_size(chunk)
        progress = Float.round(downloaded / total_size * 100, 1)
        IO.write("\rProgress: #{progress}%")
      end)
      |> Stream.run()
    end)

    IO.puts("\n✓ Download complete")
  end
end

Timeout Configuration

# Set custom timeout for each chunk (default: 30 seconds)
stream = Stream.download(store, "large-file.bin", timeout: 60_000)

Error Handling

try do
  Stream.download(store, "large-file.bin")
  |> Stream.into(File.stream!("output.bin"))
  |> Stream.run()
rescue
  error ->
    IO.puts "Download failed: #{inspect(error)}"
end

Streaming Uploads

Upload large files in chunks using multipart upload.

Basic Upload Stream

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

From Generated Data

# Upload randomly generated data
Stream.repeatedly(fn -> :crypto.strong_rand_bytes(1_048_576) end)
|> Stream.take(1000)  # 1GB total (1000 * 1MB)
|> ObjectStoreX.Stream.upload(store, "random-data.bin")

From External Source

# Upload from HTTP response
HTTPoison.get!("https://example.com/large-file.bin", stream_to: self())

stream = Stream.resource(
  fn -> :ok end,
  fn acc ->
    receive do
      %HTTPoison.AsyncChunk{chunk: data} -> {[data], acc}
      %HTTPoison.AsyncEnd{} -> {:halt, acc}
    after
      30_000 -> raise "Timeout"
    end
  end,
  fn _acc -> :ok end
)

ObjectStoreX.Stream.upload(stream, store, "downloaded-file.bin")

With Progress Tracking

defmodule Uploader do
  def upload_with_progress(local_path, store, remote_path, chunk_size \\ 10_485_760) do
    file_size = File.stat!(local_path).size
    uploaded = 0

    File.stream!(local_path, [], chunk_size)
    |> Stream.each(fn chunk ->
      uploaded = uploaded + byte_size(chunk)
      progress = Float.round(uploaded / file_size * 100, 1)
      IO.write("\rUploading: #{progress}%")
    end)
    |> ObjectStoreX.Stream.upload(store, remote_path)

    IO.puts("\n✓ Upload complete")
  end
end

Error Handling

case ObjectStoreX.Stream.upload(stream, store, "file.bin") do
  :ok ->
    IO.puts "✓ Upload successful"

  {:error, reason} ->
    IO.puts "✗ Upload failed: #{inspect(reason)}"
    # Multipart upload is automatically aborted on error
end

Listing APIs

ObjectStoreX offers three complementary listing contracts:

API	Scope	Result behavior
`ObjectStoreX.Stream.list_stream/2`	Recursive objects below a prefix	Lazy Elixir stream
`ObjectStoreX.list_with_delimiter/2`	Immediate objects and virtual folders	Complete level in one return value
`ObjectStoreX.list_with_delimiter_page/2`	Immediate objects and virtual folders	One bounded, resumable page

Choose bounded delimiter pages for paginated APIs and folder browsers. S3 (including Wasabi), Azure, and GCS issue one provider-native page request per call. Memory and local stores provide the same result semantics by materializing and deterministically slicing the complete immediate level.

{:ok, page} =
  ObjectStoreX.list_with_delimiter_page(store,
    prefix: "uploads/",
    max_keys: 100
  )

case page.next_page_token do
  nil -> :complete
  token ->
    ObjectStoreX.list_with_delimiter_page(store,
      prefix: "uploads/",
      max_keys: 100,
      page_token: token
    )
end

The limit counts objects and prefixes together. Treat tokens as opaque and reuse them unchanged with the same store, prefix, and page size. Native cloud ordering is provider-defined, and listings are not snapshots: concurrent mutation can change later pages.

Streaming Lists

List large numbers of recursive objects efficiently with automatic pagination. Unlike delimiter listings, this API emits objects from nested paths and does not return virtual folder prefixes.

Basic List Stream

# List all objects with prefix
ObjectStoreX.Stream.list_stream(store, prefix: "data/2025/")
|> Stream.map(fn meta -> meta.location end)
|> Enum.to_list()

Filter Results

# Find large files
large_files =
  ObjectStoreX.Stream.list_stream(store, prefix: "uploads/")
  |> Stream.filter(fn meta -> meta.size > 100_000_000 end)  # > 100MB
  |> Stream.map(fn meta -> {meta.location, meta.size} end)
  |> Enum.to_list()

Process in Batches

# Delete in batches of 1000
ObjectStoreX.Stream.list_stream(store, prefix: "temp/")
|> Stream.map(fn meta -> meta.location end)
|> Stream.chunk_every(1000)
|> Stream.each(fn batch ->
  {:ok, deleted, _failed} = ObjectStoreX.delete_many(store, batch)
  IO.puts "Deleted batch: #{deleted} objects"
end)
|> Stream.run()

Lazy Evaluation

# Only fetch first 10 objects (stops pagination early)
first_10 =
  ObjectStoreX.Stream.list_stream(store)
  |> Stream.take(10)
  |> Enum.to_list()

Aggregate Statistics

# Calculate total size of all objects
total_size =
  ObjectStoreX.Stream.list_stream(store, prefix: "backups/")
  |> Stream.map(fn meta -> meta.size end)
  |> Enum.sum()

IO.puts "Total backup size: #{total_size} bytes"

Range Reads

Read specific byte ranges without downloading entire files.

Single Range

# Read first 1000 bytes
{:ok, data, meta} = ObjectStoreX.get(store, "file.bin", range: {0, 999})

Multiple Ranges

# Read header and footer (e.g., Parquet file)
{:ok, [header, footer]} = ObjectStoreX.get_ranges(store, "data.parquet", [
  {0, 1000},           # Header
  {9_999_000, 10_000_000}  # Footer
])

# Parse metadata without downloading full file
parse_parquet_metadata(header, footer)

Video File Preview

# Extract video metadata from header
{:ok, [header, metadata_section]} = ObjectStoreX.get_ranges(store, "video.mp4", [
  {0, 100},     # File header
  {500, 2000}   # Metadata section
])

video_info = parse_video_header(header, metadata_section)

Parallel Range Reads

# Read multiple files in parallel
files = ["file1.bin", "file2.bin", "file3.bin"]

tasks = Enum.map(files, fn file ->
  Task.async(fn ->
    {:ok, data, _meta} = ObjectStoreX.get(store, file, range: {0, 1000})
    {file, data}
  end)
end)

results = Task.await_many(tasks)

Best Practices

1. Choose Appropriate Chunk Size

# For uploads, larger chunks = fewer requests
# Recommended: 10-50MB for cloud providers

# Small files (<10MB): Single put
:ok = ObjectStoreX.put(store, "small.txt", data)

# Large files (>10MB): Streaming
File.stream!("large.bin", [], 10_485_760)  # 10MB chunks
|> ObjectStoreX.Stream.upload(store, "large.bin")

2. Handle Backpressure

# Use Stream.chunk_every to batch operations
ObjectStoreX.Stream.list_stream(store)
|> Stream.chunk_every(100)
|> Stream.each(&process_batch/1)
|> Stream.run()

3. Implement Retry Logic

defmodule RetryUploader do
  def upload_with_retry(stream, store, path, retries \\ 3) do
    case ObjectStoreX.Stream.upload(stream, store, path) do
      :ok ->
        :ok

      {:error, reason} when retries > 0 ->
        if ObjectStoreX.Error.retryable?(reason) do
          :timer.sleep(1000)
          upload_with_retry(stream, store, path, retries - 1)
        else
          {:error, reason}
        end

      {:error, reason} ->
        {:error, reason}
    end
  end
end

4. Use Concurrent Streams

# Process multiple files concurrently
files = ["file1.bin", "file2.bin", "file3.bin"]

files
|> Task.async_stream(fn file ->
  ObjectStoreX.Stream.download(store, file)
  |> Stream.into(File.stream!("local-#{file}"))
  |> Stream.run()
end, max_concurrency: 10)
|> Stream.run()

5. Monitor Memory Usage

# Memory-efficient processing of large lists
ObjectStoreX.Stream.list_stream(store)
|> Stream.map(&process_object/1)
|> Stream.run()

# Avoid loading entire list into memory
# ❌ Don't do this:
all_objects = ObjectStoreX.Stream.list_stream(store) |> Enum.to_list()

Performance Tuning

Upload Performance

# Optimal chunk size depends on:
# - Network bandwidth
# - File size
# - Provider (S3 prefers 5MB-5GB chunks)

# For S3, use 10-50MB chunks
chunk_size = 10_485_760  # 10MB

File.stream!(path, [], chunk_size)
|> ObjectStoreX.Stream.upload(store, remote_path)

Download Performance

# Parallel downloads for multiple files
paths = ["file1.bin", "file2.bin", "file3.bin"]

paths
|> Task.async_stream(
  fn path ->
    ObjectStoreX.Stream.download(store, path)
    |> Stream.into(File.stream!("local-#{path}"))
    |> Stream.run()
  end,
  max_concurrency: 10,
  timeout: :infinity
)
|> Stream.run()

List Performance

# Use prefix to narrow results
ObjectStoreX.Stream.list_stream(store, prefix: "data/2025/01/")

# Process in batches to avoid memory issues
|> Stream.chunk_every(1000)
|> Stream.each(&process_batch/1)
|> Stream.run()

Real-World Examples

Backup System

defmodule BackupSystem do
  def backup_directory(local_dir, store, prefix) do
    File.ls!(local_dir)
    |> Task.async_stream(
      fn file ->
        local_path = Path.join(local_dir, file)
        remote_path = Path.join(prefix, file)

        File.stream!(local_path, [], 10_485_760)
        |> ObjectStoreX.Stream.upload(store, remote_path)
      end,
      max_concurrency: 5
    )
    |> Stream.run()
  end
end

Log Processor

defmodule LogProcessor do
  def process_logs(store, date) do
    prefix = "logs/#{date}/"

    ObjectStoreX.Stream.list_stream(store, prefix: prefix)
    |> Stream.flat_map(fn meta ->
      ObjectStoreX.Stream.download(store, meta.location)
    end)
    |> Stream.flat_map(&String.split(&1, "\n"))
    |> Stream.filter(&contains_error?/1)
    |> Enum.to_list()
  end

  defp contains_error?(line), do: String.contains?(line, "ERROR")
end

Data Migration

defmodule DataMigration do
  def migrate(source_store, dest_store, prefix) do
    ObjectStoreX.Stream.list_stream(source_store, prefix: prefix)
    |> Task.async_stream(
      fn meta ->
        # Stream from source to destination
        ObjectStoreX.Stream.download(source_store, meta.location)
        |> ObjectStoreX.Stream.upload(dest_store, meta.location)
      end,
      max_concurrency: 10,
      timeout: :infinity
    )
    |> Stream.run()
  end
end

Next Steps

← Previous Page Configuration Guide

Next Page → Distributed Systems Guide

Streaming Guide

Table of Contents

Why Streaming?

Streaming Downloads

Basic Download Stream

Process Chunks

With Progress Tracking

Timeout Configuration

Error Handling

Streaming Uploads

Basic Upload Stream

From Generated Data

From External Source

With Progress Tracking

Error Handling

Listing APIs

Streaming Lists

Basic List Stream

Filter Results

Process in Batches

Lazy Evaluation

Aggregate Statistics

Range Reads

Single Range

Multiple Ranges

Video File Preview

Parallel Range Reads

Best Practices

1. Choose Appropriate Chunk Size

2. Handle Backpressure

3. Implement Retry Logic

4. Use Concurrent Streams

5. Monitor Memory Usage

Performance Tuning

Upload Performance

Download Performance

List Performance

Real-World Examples

Backup System

Log Processor

Data Migration

Next Steps