HttpQueue

HttpQueue provides HTTP-specific features on top of the base queue: per-domain concurrency limiting, automatic rate limit header parsing, retry with exponential backoff, HTTP method-based priority, and domain pause/resume.

Constructor

new HttpQueue(config?: HttpQueueConfig)

HttpQueueConfig

Extends QueueConfig with HTTP-specific options:

Plus all QueueConfig options (concurrency, timeout, autoStart, etc.).

Example

import { HttpQueue } from 'rezo';

const queue = new HttpQueue({
  concurrency: 10,
  domainConcurrency: 2,
  requestsPerSecond: 5,
  respectRetryAfter: true,
  autoRetry: true,
  maxRetries: 3,
  retryStatusCodes: [429, 500, 502, 503, 504]
});

// Per-domain limits
const strictQueue = new HttpQueue({
  concurrency: 20,
  domainConcurrency: {
    'api.example.com': 5,
    'cdn.example.com': 10,
    'default': 2
  }
});

// Use with Rezo
const rezo = new Rezo({
  queue: new HttpQueue({ concurrency: 5, domainConcurrency: 2 })
});

Adding HTTP Tasks

addHttp(fn, options?)

Adds an HTTP-aware task to the queue with domain limiting and method-based priority.

addHttp<R = any>(fn: TaskFunction<R>, options?: HttpTaskOptions): Promise<R>

HttpTaskOptions

Extends TaskOptions with HTTP-specific fields:

Plus all TaskOptions (priority, timeout, id, signal).

Example

// Basic HTTP task
const data = await queue.addHttp(
  () => fetch('https://api.example.com/users').then(r => r.json()),
  { domain: 'api.example.com', method: 'GET' }
);

// POST with higher priority
await queue.addHttp(
  () => fetch('/api/data', { method: 'POST', body: payload }),
  { domain: 'api.example.com', method: 'POST', retry: 3 }
);

// With abort signal
const controller = new AbortController();
queue.addHttp(
  () => fetch('/api/long-task', { signal: controller.signal }),
  { domain: 'api.example.com', signal: controller.signal }
);

Domain Control

pauseDomain(domain)

Pauses all requests to a specific domain. Running tasks continue; no new tasks start for this domain.

pauseDomain(domain: string): void

resumeDomain(domain)

Resumes requests to a previously paused domain.

resumeDomain(domain: string): void

setDomainConcurrency(domain, limit)

Sets or updates the concurrency limit for a specific domain at runtime.

setDomainConcurrency(domain: string, limit: number): void

getDomainState(domain)

Returns the current state of a specific domain.

getDomainState(domain: string): DomainState

interface DomainState {
  pending: number;           // Tasks currently running for domain
  size: number;              // Tasks waiting for domain
  isPaused: boolean;         // Is domain paused
  rateLimitedUntil?: number; // Timestamp when rate limit expires
}

Example

// Pause a domain during maintenance
queue.pauseDomain('api.example.com');

// Resume when ready
queue.resumeDomain('api.example.com');

// Adjust concurrency at runtime
queue.setDomainConcurrency('cdn.example.com', 20);

// Check domain state
const state = queue.getDomainState('api.example.com');
console.log(`Pending: ${state.pending}, Queued: ${state.size}`);
if (state.rateLimitedUntil) {
  console.log(`Rate limited until: ${new Date(state.rateLimitedUntil)}`);
}

Rate Limiting

handleRateLimit(domain, retryAfter)

Manually triggers a rate limit for a domain. Pauses requests for the specified duration and emits a rateLimited event.

handleRateLimit(domain: string, retryAfter: number): void

Example

// Manual rate limit handling
queue.handleRateLimit('api.example.com', 60); // Pause for 60 seconds

// Automatic handling (when respectRetryAfter: true)
// The queue automatically parses Retry-After headers from 429 responses

HTTP Cancellation

cancelHttp(id)

Cancels a specific HTTP task by ID. Searches across all domain queues.

cancelHttp(id: string): boolean

clearHttp()

Clears all HTTP tasks across all domains.

clearHttp(): void

HTTP Statistics

httpStats

Returns HTTP-specific statistics including per-domain breakdowns.

get httpStats: HttpQueueStats

interface HttpQueueStats extends QueueStats {
  byDomain: Record<string, {
    pending: number;
    completed: number;
    failed: number;
    rateLimited: number;
  }>;
  retries: number;        // Total retries performed
  rateLimitHits: number;  // Total rate limit events
}

httpConfig

Read-only access to the resolved HTTP queue configuration.

readonly httpConfig: Required<HttpQueueConfig>

Example

const stats = queue.httpStats;
console.log('Total retries:', stats.retries);
console.log('Rate limit hits:', stats.rateLimitHits);

for (const [domain, domainStats] of Object.entries(stats.byDomain)) {
  console.log(`${domain}: ${domainStats.completed} completed, ${domainStats.failed} failed`);
}

Events

HttpQueue supports all base QueueEvents plus HTTP-specific events:

on(event, handler) / off(event, handler)

on<E extends keyof HttpQueueEvents>(event: E, handler: EventHandler<HttpQueueEvents[E]>): void
off<E extends keyof HttpQueueEvents>(event: E, handler: EventHandler<HttpQueueEvents[E]>): void

HttpQueueEvents

Plus all base events: add, start, completed, error, timeout, cancelled, active, idle, paused, resumed, next, empty.

Example

queue.on('rateLimited', ({ domain, retryAfter }) => {
  console.log(`${domain} rate limited for ${retryAfter}s`);
});

queue.on('domainAvailable', ({ domain }) => {
  console.log(`${domain} is available again`);
});

queue.on('retry', ({ id, attempt, error }) => {
  console.log(`Retrying task ${id} (attempt ${attempt}): ${error.message}`);
});

Lifecycle

destroy()

Destroys the queue: clears all HTTP tasks, removes event handlers, and calls parent destroy().

destroy(): void

HttpMethodPriority

Default priority values assigned to HTTP methods when no explicit priority is set.

const HttpMethodPriority: Record<string, number> = {
  HEAD: 100,
  GET: 75,
  OPTIONS: 50,
  POST: 50,
  PUT: 50,
  PATCH: 50,
  DELETE: 25
};

This means HEAD requests are prioritized over GET, which are prioritized over POST/PUT/PATCH, which are prioritized over DELETE. Override by passing an explicit priority in HttpTaskOptions.

Example

// HEAD runs first (priority 100), then GET (75), then DELETE (25)
queue.addHttp(() => fetch('/resource', { method: 'HEAD' }), { method: 'HEAD' });
queue.addHttp(() => fetch('/resource'), { method: 'GET' });
queue.addHttp(() => fetch('/resource', { method: 'DELETE' }), { method: 'DELETE' });

// Override default priority
queue.addHttp(() => fetch('/critical'), {
  method: 'DELETE',
  priority: 1000 // Run this DELETE first
});