ObjectStreamWrapper

ObjectStreamWrapper

The storage abstraction the sort algorithms read and write through. A wrapper ferries objects to and from some backing store — a local file, an in-memory array, S3, an SSH pipe, a sharded volume, or a heterogeneous mix. The algorithms see only object streams; the wrapper hides the rest. This is what lets a sort exceed the capacity of any single disk.

Built-in implementations: MemoryWrapper, LocalFileWrapper.

Contract

class ObjectStreamWrapper<T> {
  openWriter(): ItemWriter<T>; // discards previous content
  openReader(): AsyncIterable<T>; // replays written items in order
  close(): Promise<void>; // release the current mode
  delete(): Promise<void>; // remove the backing storage
}

Mode-exclusive. A wrapper is idle, writing, or reading at any moment. openWriter() while reading — or openReader() while writing — throws; call close() first. openWriter() discards any previous content. The iterator returned from openReader()[Symbol.asyncIterator]() implements return(), so resources are released when a for await loop breaks early or throws.

Lifecycle. The algorithm delete()s wrappers it created via a factory (tmpDir / createWrapper), and close()s wrappers handed to it explicitly (files). keepTempFiles: true skips deletion.

ItemWriter

The push-side handle returned by openWriter():

class ItemWriter<T> {
  write(item: T): Promise<void>; // backpressure flows through the await
  writeAll(source: AsyncIterable<T> | Iterable<T>): Promise<void>; // drains source; does NOT end
  end(): Promise<void>; // idempotent
  get ended(): boolean;
}

writeAll deliberately does not call end(), so you can interleave write / writeAll and end once at the end. end() is idempotent; ended flips synchronously inside the first end().

consume

One-shot "drain then close" helper:

import {consume} from 'stream-sorting';

await consume(wrapper.openWriter(), source); // writeAll(source) then end()

Implementing a wrapper

Extend ObjectStreamWrapper (for instanceof checks and the shared writeAll default) or satisfy the contract structurally — TypeScript and JavaScript accept either. ItemWriter subclasses must override write, end, and the ended getter; writeAll has a default loop. See MemoryWrapper and LocalFileWrapper for reference implementations.

import {ItemWriter, ObjectStreamWrapper} from 'stream-sorting';

See also: sort, polyphaseSort.