Rivulet.Http

Name: Rivulet.Http
Author: Jeffeek

Parallel HTTP operations with automatic retries, resilient downloads, and HttpClientFactory integration.

Built on top of Rivulet.Core, this package provides HTTP-aware parallel operators that automatically handle transient failures, respect rate limits, and support resumable downloads.

Installation

dotnet add package Rivulet.Http

Requires Rivulet.Core (automatically included).

Quick Start

Parallel HTTP GET Requests

Fetch multiple URLs in parallel with automatic retry for transient HTTP errors:

using Rivulet.Http;

var urls = new[]
{
    new Uri("https://api.example.com/users/1"),
    new Uri("https://api.example.com/users/2"),
    new Uri("https://api.example.com/users/3")
};

var responses = await urls.GetParallelAsync(
    httpClient,
    new HttpOptions
    {
        ParallelOptions = new ParallelOptionsRivulet
        {
            MaxDegreeOfParallelism = 10,
            MaxRetries = 3
        }
    });

foreach (var response in responses)
{
    var content = await response.Content.ReadAsStringAsync();
    Console.WriteLine(content);
    response.Dispose();
}

Get String Content Directly

Automatically read response content as strings:

var urls = new[]
{
    new Uri("https://api.example.com/data/1"),
    new Uri("https://api.example.com/data/2")
};

var contents = await urls.GetStringParallelAsync(httpClient);

foreach (var content in contents)
{
    Console.WriteLine(content);
}

Parallel POST Requests

Send multiple POST requests in parallel:

var requests = users.Select(user => (
    uri: new Uri("https://api.example.com/users"),
    content: new StringContent(
        JsonSerializer.Serialize(user),
        Encoding.UTF8,
        "application/json")
));

var responses = await requests.PostParallelAsync(
    httpClient,
    new HttpOptions
    {
        ParallelOptions = new ParallelOptionsRivulet
        {
            MaxDegreeOfParallelism = 5,
            ErrorMode = ErrorMode.CollectAndContinue
        }
    });

HttpClientFactory Integration

Use named or typed HttpClient instances with IHttpClientFactory:

using Microsoft.Extensions.DependencyInjection;

var services = new ServiceCollection();
services.AddHttpClient("api", client =>
{
    client.BaseAddress = new Uri("https://api.example.com");
    client.DefaultRequestHeaders.Add("Authorization", "Bearer token");
});

var provider = services.BuildServiceProvider();
var factory = provider.GetRequiredService<IHttpClientFactory>();

var urls = new[]
{
    new Uri("/endpoint1", UriKind.Relative),
    new Uri("/endpoint2", UriKind.Relative)
};

var results = await urls.GetStringParallelAsync(
    factory,
    clientName: "api",
    new HttpOptions
    {
        ParallelOptions = new ParallelOptionsRivulet
        {
            MaxDegreeOfParallelism = 10
        }
    });

Resilient Downloads with Resume Support

Download files in parallel with automatic resume on failure:

var downloads = new[]
{
    (uri: new Uri("https://example.com/file1.zip"), path: "downloads/file1.zip"),
    (uri: new Uri("https://example.com/file2.zip"), path: "downloads/file2.zip"),
    (uri: new Uri("https://example.com/file3.zip"), path: "downloads/file3.zip")
};

var results = await downloads.DownloadParallelAsync(
    httpClient,
    new StreamingDownloadOptions
    {
        EnableResume = true,
        ValidateContentLength = true,
        OnProgressAsync = (uri, downloaded, total) =>
        {
            var percent = total.HasValue ? (downloaded * 100.0 / total.Value) : 0;
            Console.WriteLine($"{uri}: {downloaded:N0}/{total:N0} bytes ({percent:F1}%)");
            return ValueTask.CompletedTask;
        },
        HttpOptions = new HttpOptions
        {
            ParallelOptions = new ParallelOptionsRivulet
            {
                MaxDegreeOfParallelism = 3,
                MaxRetries = 5
            }
        }
    });

foreach (var (uri, filePath, bytesDownloaded) in results)
{
    Console.WriteLine($"Downloaded {uri} to {filePath}: {bytesDownloaded:N0} bytes");
}

Stream Download to Custom Destination

Download directly to any Stream:

using var memoryStream = new MemoryStream();

var bytesDownloaded = await HttpStreamingExtensions.DownloadToStreamAsync(
    new Uri("https://example.com/data.json"),
    memoryStream,
    httpClient,
    new StreamingDownloadOptions
    {
        BufferSize = 8192,
        OnProgressAsync = (uri, downloaded, total) =>
        {
            Console.WriteLine($"Downloaded: {downloaded:N0} bytes");
            return ValueTask.CompletedTask;
        }
    });

memoryStream.Position = 0;
var data = await JsonSerializer.DeserializeAsync<MyData>(memoryStream);

Automatic Retry Handling

Rivulet.Http automatically retries transient HTTP errors:

408 Request Timeout
429 Too Many Requests
500 Internal Server Error
502 Bad Gateway
503 Service Unavailable
504 Gateway Timeout

Respect Retry-After Headers

Automatically respects Retry-After headers from rate-limited APIs:

var options = new HttpOptions
{
    RespectRetryAfterHeader = true, // Default: true
    ParallelOptions = new ParallelOptionsRivulet
    {
        MaxRetries = 5,
        BaseDelay = TimeSpan.FromSeconds(1)
    }
};

var responses = await urls.GetParallelAsync(httpClient, options);

When a server returns 429 Too Many Requests or 503 Service Unavailable with a Retry-After header, Rivulet.Http will wait the specified duration before retrying instead of using the configured backoff strategy.

Error Handling

HTTP-Specific Error Callbacks

Get notified of HTTP errors with status codes:

var options = new HttpOptions
{
    OnHttpErrorAsync = (uri, statusCode, exception) =>
    {
        Console.WriteLine($"Error fetching {uri}: {statusCode} - {exception.Message}");
        return ValueTask.CompletedTask;
    },
    ParallelOptions = new ParallelOptionsRivulet
    {
        ErrorMode = ErrorMode.CollectAndContinue,
        MaxRetries = 3
    }
};

var responses = await urls.GetParallelAsync(httpClient, options);

Custom Retriable Status Codes

Customize which HTTP status codes should trigger retries:

var options = new HttpOptions
{
    RetriableStatusCodes = new HashSet<HttpStatusCode>
    {
        HttpStatusCode.TooManyRequests,
        HttpStatusCode.ServiceUnavailable,
        HttpStatusCode.RequestTimeout
    },
    ParallelOptions = new ParallelOptionsRivulet
    {
        MaxRetries = 3,
        BackoffStrategy = BackoffStrategy.ExponentialJitter
    }
};

Advanced Features

Rate Limiting

Control request rate to respect API limits:

var options = new HttpOptions
{
    ParallelOptions = new ParallelOptionsRivulet
    {
        MaxDegreeOfParallelism = 10,
        RateLimit = new RateLimitOptions
        {
            TokensPerSecond = 100,
            BurstCapacity = 200
        }
    }
};

var responses = await urls.GetParallelAsync(httpClient, options);

Circuit Breaker

Prevent cascading failures with circuit breaker pattern:

var options = new HttpOptions
{
    ParallelOptions = new ParallelOptionsRivulet
    {
        CircuitBreaker = new CircuitBreakerOptions
        {
            FailureThreshold = 5,
            SuccessThreshold = 2,
            OpenTimeout = TimeSpan.FromSeconds(30)
        }
    }
};

Progress Tracking

Monitor download progress for long-running operations:

var options = new StreamingDownloadOptions
{
    ProgressInterval = TimeSpan.FromSeconds(1),
    OnProgressAsync = (uri, downloaded, total) =>
    {
        var percent = total.HasValue ? (downloaded * 100.0 / total.Value) : 0;
        Console.WriteLine($"Progress: {percent:F1}%");
        return ValueTask.CompletedTask;
    },
    OnResumeAsync = (uri, offset) =>
    {
        Console.WriteLine($"Resuming download from byte {offset:N0}");
        return ValueTask.CompletedTask;
    },
    OnCompleteAsync = (uri, path, bytes) =>
    {
        Console.WriteLine($"Download complete: {bytes:N0} bytes saved to {path}");
        return ValueTask.CompletedTask;
    }
};

HTTP Methods Supported

All standard HTTP methods with parallel operators:

GET: GetParallelAsync, GetStringParallelAsync, GetByteArrayParallelAsync
POST: PostParallelAsync
PUT: PutParallelAsync
DELETE: DeleteParallelAsync

All methods support both HttpClient and IHttpClientFactory.

Configuration Options

HttpOptions

HTTP-specific configuration:

var options = new HttpOptions
{
    RequestTimeout = TimeSpan.FromSeconds(30),
    RespectRetryAfterHeader = true,
    RetriableStatusCodes = new HashSet<HttpStatusCode> { /* ... */ },
    BufferSize = 81920,
    OnHttpErrorAsync = async (uri, status, ex) => { /* ... */ },
    ParallelOptions = new ParallelOptionsRivulet
    {
        MaxDegreeOfParallelism = 10,
        MaxRetries = 3,
        BaseDelay = TimeSpan.FromMilliseconds(100),
        BackoffStrategy = BackoffStrategy.ExponentialJitter,
        ErrorMode = ErrorMode.CollectAndContinue
    }
};

StreamingDownloadOptions

Download-specific configuration:

var options = new StreamingDownloadOptions
{
    EnableResume = true,
    ValidateContentLength = true,
    OverwriteExisting = false,
    BufferSize = 81920,
    ProgressInterval = TimeSpan.FromSeconds(1),
    HttpOptions = new HttpOptions { /* ... */ }
};

Best Practices

Dispose HttpResponseMessage: Always dispose responses when using GetParallelAsync, PostParallelAsync, etc.
Use String/ByteArray Variants: Use GetStringParallelAsync or GetByteArrayParallelAsync for automatic disposal
Set MaxDegreeOfParallelism: Always limit concurrency to avoid overwhelming servers
Enable Resume for Large Files: Use EnableResume = true for downloads that might be interrupted
Respect Rate Limits: Use RespectRetryAfterHeader = true and configure RateLimit for APIs
Use HttpClientFactory: Prefer IHttpClientFactory over creating HttpClient instances manually

Performance

Rivulet.Http is designed for high-throughput scenarios:

Bounded Concurrency: Prevents resource exhaustion
Backpressure: Channel-based buffering prevents memory issues
Zero Allocations: Uses ValueTask<T> in hot paths
Streaming Downloads: Efficient memory usage for large files
Resume Support: Saves bandwidth by resuming interrupted downloads

Examples

See the samples directory for complete working examples including:

API scraping with rate limiting
Bulk data export/import
Image downloading with progress tracking
Resilient file synchronization

License

MIT License - see LICENSE file for details

Made with ❤️ by Jeffeek | NuGet | GitHub

Rivulet.Http

Parallel HTTP operations with automatic retries, resilient downloads, and HttpClientFactory integration.

Built on top of Rivulet.Core, this package provides HTTP-aware parallel operators that automatically handle transient failures, respect rate limits, and support resumable downloads.

Installation

dotnet add package Rivulet.Http

Requires Rivulet.Core (automatically included).

Quick Start

Parallel HTTP GET Requests

Fetch multiple URLs in parallel with automatic retry for transient HTTP errors:

using Rivulet.Http;

var urls = new[]
{
    new Uri("https://api.example.com/users/1"),
    new Uri("https://api.example.com/users/2"),
    new Uri("https://api.example.com/users/3")
};

var responses = await urls.GetParallelAsync(
    httpClient,
    new HttpOptions
    {
        ParallelOptions = new ParallelOptionsRivulet
        {
            MaxDegreeOfParallelism = 10,
            MaxRetries = 3
        }
    });

foreach (var response in responses)
{
    var content = await response.Content.ReadAsStringAsync();
    Console.WriteLine(content);
    response.Dispose();
}

Get String Content Directly

Automatically read response content as strings:

var urls = new[]
{
    new Uri("https://api.example.com/data/1"),
    new Uri("https://api.example.com/data/2")
};

var contents = await urls.GetStringParallelAsync(httpClient);

foreach (var content in contents)
{
    Console.WriteLine(content);
}

Parallel POST Requests

Send multiple POST requests in parallel:

var requests = users.Select(user => (
    uri: new Uri("https://api.example.com/users"),
    content: new StringContent(
        JsonSerializer.Serialize(user),
        Encoding.UTF8,
        "application/json")
));

var responses = await requests.PostParallelAsync(
    httpClient,
    new HttpOptions
    {
        ParallelOptions = new ParallelOptionsRivulet
        {
            MaxDegreeOfParallelism = 5,
            ErrorMode = ErrorMode.CollectAndContinue
        }
    });

HttpClientFactory Integration

Use named or typed HttpClient instances with IHttpClientFactory:

using Microsoft.Extensions.DependencyInjection;

var services = new ServiceCollection();
services.AddHttpClient("api", client =>
{
    client.BaseAddress = new Uri("https://api.example.com");
    client.DefaultRequestHeaders.Add("Authorization", "Bearer token");
});

var provider = services.BuildServiceProvider();
var factory = provider.GetRequiredService<IHttpClientFactory>();

var urls = new[]
{
    new Uri("/endpoint1", UriKind.Relative),
    new Uri("/endpoint2", UriKind.Relative)
};

var results = await urls.GetStringParallelAsync(
    factory,
    clientName: "api",
    new HttpOptions
    {
        ParallelOptions = new ParallelOptionsRivulet
        {
            MaxDegreeOfParallelism = 10
        }
    });

Resilient Downloads with Resume Support

Download files in parallel with automatic resume on failure:

var downloads = new[]
{
    (uri: new Uri("https://example.com/file1.zip"), path: "downloads/file1.zip"),
    (uri: new Uri("https://example.com/file2.zip"), path: "downloads/file2.zip"),
    (uri: new Uri("https://example.com/file3.zip"), path: "downloads/file3.zip")
};

var results = await downloads.DownloadParallelAsync(
    httpClient,
    new StreamingDownloadOptions
    {
        EnableResume = true,
        ValidateContentLength = true,
        OnProgressAsync = (uri, downloaded, total) =>
        {
            var percent = total.HasValue ? (downloaded * 100.0 / total.Value) : 0;
            Console.WriteLine($"{uri}: {downloaded:N0}/{total:N0} bytes ({percent:F1}%)");
            return ValueTask.CompletedTask;
        },
        HttpOptions = new HttpOptions
        {
            ParallelOptions = new ParallelOptionsRivulet
            {
                MaxDegreeOfParallelism = 3,
                MaxRetries = 5
            }
        }
    });

foreach (var (uri, filePath, bytesDownloaded) in results)
{
    Console.WriteLine($"Downloaded {uri} to {filePath}: {bytesDownloaded:N0} bytes");
}

Stream Download to Custom Destination

Download directly to any Stream:

using var memoryStream = new MemoryStream();

var bytesDownloaded = await HttpStreamingExtensions.DownloadToStreamAsync(
    new Uri("https://example.com/data.json"),
    memoryStream,
    httpClient,
    new StreamingDownloadOptions
    {
        BufferSize = 8192,
        OnProgressAsync = (uri, downloaded, total) =>
        {
            Console.WriteLine($"Downloaded: {downloaded:N0} bytes");
            return ValueTask.CompletedTask;
        }
    });

memoryStream.Position = 0;
var data = await JsonSerializer.DeserializeAsync<MyData>(memoryStream);

Automatic Retry Handling

Rivulet.Http automatically retries transient HTTP errors:

408 Request Timeout
429 Too Many Requests
500 Internal Server Error
502 Bad Gateway
503 Service Unavailable
504 Gateway Timeout

Respect Retry-After Headers

Automatically respects Retry-After headers from rate-limited APIs:

var options = new HttpOptions
{
    RespectRetryAfterHeader = true, // Default: true
    ParallelOptions = new ParallelOptionsRivulet
    {
        MaxRetries = 5,
        BaseDelay = TimeSpan.FromSeconds(1)
    }
};

var responses = await urls.GetParallelAsync(httpClient, options);

Error Handling

HTTP-Specific Error Callbacks

Get notified of HTTP errors with status codes:

var options = new HttpOptions
{
    OnHttpErrorAsync = (uri, statusCode, exception) =>
    {
        Console.WriteLine($"Error fetching {uri}: {statusCode} - {exception.Message}");
        return ValueTask.CompletedTask;
    },
    ParallelOptions = new ParallelOptionsRivulet
    {
        ErrorMode = ErrorMode.CollectAndContinue,
        MaxRetries = 3
    }
};

var responses = await urls.GetParallelAsync(httpClient, options);

Custom Retriable Status Codes

Customize which HTTP status codes should trigger retries:

var options = new HttpOptions
{
    RetriableStatusCodes = new HashSet<HttpStatusCode>
    {
        HttpStatusCode.TooManyRequests,
        HttpStatusCode.ServiceUnavailable,
        HttpStatusCode.RequestTimeout
    },
    ParallelOptions = new ParallelOptionsRivulet
    {
        MaxRetries = 3,
        BackoffStrategy = BackoffStrategy.ExponentialJitter
    }
};

Advanced Features

Rate Limiting

Control request rate to respect API limits:

var options = new HttpOptions
{
    ParallelOptions = new ParallelOptionsRivulet
    {
        MaxDegreeOfParallelism = 10,
        RateLimit = new RateLimitOptions
        {
            TokensPerSecond = 100,
            BurstCapacity = 200
        }
    }
};

var responses = await urls.GetParallelAsync(httpClient, options);

Circuit Breaker

Prevent cascading failures with circuit breaker pattern:

var options = new HttpOptions
{
    ParallelOptions = new ParallelOptionsRivulet
    {
        CircuitBreaker = new CircuitBreakerOptions
        {
            FailureThreshold = 5,
            SuccessThreshold = 2,
            OpenTimeout = TimeSpan.FromSeconds(30)
        }
    }
};

Progress Tracking

Monitor download progress for long-running operations:

var options = new StreamingDownloadOptions
{
    ProgressInterval = TimeSpan.FromSeconds(1),
    OnProgressAsync = (uri, downloaded, total) =>
    {
        var percent = total.HasValue ? (downloaded * 100.0 / total.Value) : 0;
        Console.WriteLine($"Progress: {percent:F1}%");
        return ValueTask.CompletedTask;
    },
    OnResumeAsync = (uri, offset) =>
    {
        Console.WriteLine($"Resuming download from byte {offset:N0}");
        return ValueTask.CompletedTask;
    },
    OnCompleteAsync = (uri, path, bytes) =>
    {
        Console.WriteLine($"Download complete: {bytes:N0} bytes saved to {path}");
        return ValueTask.CompletedTask;
    }
};

HTTP Methods Supported

All standard HTTP methods with parallel operators:

GET: GetParallelAsync, GetStringParallelAsync, GetByteArrayParallelAsync
POST: PostParallelAsync
PUT: PutParallelAsync
DELETE: DeleteParallelAsync

All methods support both HttpClient and IHttpClientFactory.

Configuration Options

HttpOptions

HTTP-specific configuration:

var options = new HttpOptions
{
    RequestTimeout = TimeSpan.FromSeconds(30),
    RespectRetryAfterHeader = true,
    RetriableStatusCodes = new HashSet<HttpStatusCode> { /* ... */ },
    BufferSize = 81920,
    OnHttpErrorAsync = async (uri, status, ex) => { /* ... */ },
    ParallelOptions = new ParallelOptionsRivulet
    {
        MaxDegreeOfParallelism = 10,
        MaxRetries = 3,
        BaseDelay = TimeSpan.FromMilliseconds(100),
        BackoffStrategy = BackoffStrategy.ExponentialJitter,
        ErrorMode = ErrorMode.CollectAndContinue
    }
};

StreamingDownloadOptions

Download-specific configuration:

var options = new StreamingDownloadOptions
{
    EnableResume = true,
    ValidateContentLength = true,
    OverwriteExisting = false,
    BufferSize = 81920,
    ProgressInterval = TimeSpan.FromSeconds(1),
    HttpOptions = new HttpOptions { /* ... */ }
};

Best Practices

Dispose HttpResponseMessage: Always dispose responses when using GetParallelAsync, PostParallelAsync, etc.
Use String/ByteArray Variants: Use GetStringParallelAsync or GetByteArrayParallelAsync for automatic disposal
Set MaxDegreeOfParallelism: Always limit concurrency to avoid overwhelming servers
Enable Resume for Large Files: Use EnableResume = true for downloads that might be interrupted
Respect Rate Limits: Use RespectRetryAfterHeader = true and configure RateLimit for APIs
Use HttpClientFactory: Prefer IHttpClientFactory over creating HttpClient instances manually

Performance

Rivulet.Http is designed for high-throughput scenarios:

Bounded Concurrency: Prevents resource exhaustion
Backpressure: Channel-based buffering prevents memory issues
Zero Allocations: Uses ValueTask<T> in hot paths
Streaming Downloads: Efficient memory usage for large files
Resume Support: Saves bandwidth by resuming interrupted downloads

Examples

See the samples directory for complete working examples including:

API scraping with rate limiting
Bulk data export/import
Image downloading with progress tracking
Resilient file synchronization

License

MIT License - see LICENSE file for details

Made with ❤️ by Jeffeek | NuGet | GitHub

Jeffeek/Rivulet.Httpv1.3.0

Get Started

Readme

Rivulet.Http

Installation

Quick Start

Parallel HTTP GET Requests

Get String Content Directly

Parallel POST Requests

HttpClientFactory Integration

Resilient Downloads with Resume Support

Stream Download to Custom Destination

Automatic Retry Handling

Respect Retry-After Headers

Error Handling

HTTP-Specific Error Callbacks

Custom Retriable Status Codes

Advanced Features

Rate Limiting

Circuit Breaker

Progress Tracking

HTTP Methods Supported

Configuration Options

HttpOptions

StreamingDownloadOptions

Best Practices

Performance

Examples

License

Jeffeek/Rivulet.Httpv1.3.0

Get Started

Readme

Rivulet.Http

Installation

Quick Start

Parallel HTTP GET Requests

Get String Content Directly

Parallel POST Requests

HttpClientFactory Integration

Resilient Downloads with Resume Support

Stream Download to Custom Destination

Automatic Retry Handling

Respect Retry-After Headers

Error Handling

HTTP-Specific Error Callbacks

Custom Retriable Status Codes

Advanced Features

Rate Limiting

Circuit Breaker

Progress Tracking

HTTP Methods Supported

Configuration Options

HttpOptions

StreamingDownloadOptions

Best Practices

Performance

Examples

License