Hyperse
  • Pricing
  • FAQs
Newsletter

Subscribe to our newsletter

Receive weekly updates with latest collections and exclusive offers, straight to your inbox.

Hyperse

Headless commerce solutions for brands, developers, and modern ecommerce teams.

Explore

  • Home
  • About Us
  • Partners
  • FAQs

Legal & resources

  • Proposal
  • Privacy Policy
  • Terms & Conditions

© 2026 Hyperse. All rights reserved.

  1. Back to Popular Libraries
  2. @hyperse/pipeline
Pipeline

@hyperse/pipeline

A composable pipeline engine designed for efficient data processing and task orchestration. Enables chaining of operations with high throughput and minimal overhead.

middlewarepipelinedata-transformernode-pipelinehypersets-pipelinets-pipetyped-chianer

Quick start

git clone https://github.com/hyperse-io/pipeline.git
loading...

On this page

  • @hyperse/pipeline
    • Features
      • 🚀 Middleware Engine
      • ⚡ Reactive Pipeline
      • 🔧 Advanced Features
    • Installation
    • API Reference
      • Middleware Engine
        • Pipeline<T>
        • Middleware<T>
      • Reactive Pipeline
        • pipe(...functions)
        • Returned Function Methods
        • exitPipe<T>(value: T)
        • isExitPipeValue<T>(value: unknown)
        • pipeContext<T, L, R, C>(fn: (ctx: C) => (r: T, l: L) => R)
    • Usage Examples
      • 1. Middleware Engine
        • Basic Middleware Usage
        • Error Handling
      • 2. Reactive Pipeline
        • Basic Pipeline
        • Pipeline with Arguments
        • Early Exit
        • Context Injection
        • Dynamic Pipeline Modification
        • Promise Support
        • Complex Promise Pipeline
      • 3. Monad Integration
        • Either Support
        • Maybe Support
        • Purify-ts Integration
      • 4. Advanced Patterns
        • Nested Pipelines
        • Context with Nested Pipelines
        • Error Handling in Pipeline
    • Type Safety
    • Performance Considerations
    • Browser Support
    • Development
    • License

@hyperse/pipeline

build stable version GitHub top language Licence

A powerful TypeScript middleware engine and reactive pipeline library that provides two main paradigms for handling data flow and middleware patterns.

Features

🚀 Middleware Engine

  • Type-safe middleware composition with full TypeScript support
  • Async/await support for modern JavaScript patterns
  • Error handling middleware with automatic error propagation
  • Context sharing between middleware functions
  • Chaining and composition of middleware functions

⚡ Reactive Pipeline

  • Functional programming patterns with pipe composition
  • Promise and async support throughout the pipeline
  • Early exit capabilities with exitPipe
  • Global context injection with pipeContext
  • Dynamic pipeline modification with replace and replaceUndo
  • Monad support for Either, Maybe, and Validation types
  • Library integrations with fp-ts, monet, and purify-ts

🔧 Advanced Features

  • Type inference with up to 10 function compositions
  • Multiple parameter support (0-3 parameters per function)
  • Monad composition for functional programming patterns
  • Before/after hooks for pipeline lifecycle management
  • Memory efficient with WeakMap and WeakSet usage

Installation

npm install @hyperse/pipeline
# or
yarn add @hyperse/pipeline

API Reference

Middleware Engine

Pipeline<T>

A middleware container and invoker for handling context-based operations.

Parameters:

  • T: The context type that will be shared between middleware functions

Methods:

  • use(...middlewares): Add middleware functions to the pipeline
  • execute(context): Execute the middleware chain with the given context

Middleware<T>

Type definition for middleware functions.

Reactive Pipeline

pipe(...functions)

Creates a function that pipes input through a series of functions.

Features:

  • Supports up to 10 function compositions
  • Handles 0-3 parameters per function
  • Returns a callable function with additional methods

Returned Function Methods

exitPipe<T>(value: T)

Exits the pipeline early and returns a wrapped value.

function exitPipevalue  ExitPipeReturnValue

isExitPipeValue<T>(value: unknown)

Type guard to check if a value is an exit pipe return value.

pipeContext<T, L, R, C>(fn: (ctx: C) => (r: T, l: L) => R)

Creates a context-aware function for the pipeline.

Usage Examples

1. Middleware Engine

Basic Middleware Usage

Error Handling

2. Reactive Pipeline

Basic Pipeline

Pipeline with Arguments

Early Exit

Context Injection

Dynamic Pipeline Modification

Promise Support

Complex Promise Pipeline

3. Monad Integration

Either Support

Maybe Support

Purify-ts Integration

4. Advanced Patterns

Nested Pipelines

Context with Nested Pipelines

Error Handling in Pipeline

Type Safety

The library provides comprehensive TypeScript support with:

  • Generic type inference for context and return types
  • Function signature preservation across pipeline compositions
  • Monad type detection for automatic composition
  • Parameter count validation (0-3 parameters per function)
  • Return type inference with exit pipe detection

Performance Considerations

  • Lazy evaluation - Pipelines are only executed when called
  • Memory efficient - Uses WeakMap and WeakSet for internal state
  • Minimal overhead - Direct function composition without unnecessary wrappers
  • Async optimization - Efficient Promise handling and composition

Browser Support

The library is built with modern JavaScript features and supports:

  • Node.js 16+
  • Modern browsers with ES2020+ support
  • TypeScript 4.5+

Development

# Install dependencies
yarn

License

MIT License - see LICENSE file for details.

class Pipeline<T> {
constructor(...middlewares: Middleware<T>[]);
use(...mw: Middleware<T>[]): Pipeline<T>;
execute(context: T): Promise<void>;
}
type Middleware<T> = (
context: T,
next: Next<T>,
error?: Error
) => Promise<void> | void;
type Next<T> = (error?: Error) => Promise<Middleware<T>> | Promise<void>;
function pipe<T1, T2, ..., T10>(
fn0: () => PipeFuncReturn<T1, L1, R1>,
fn1: (x: T1, l: L1) => PipeFuncReturn<T2, L2, R2>,
// ... up to 10 functions
): PipeReturn
interface PipeFunction {
(...args: unknown[]): Promise<unknown>;
replace(replacements: Array<[number, Function]>): PipeFunction;
replaceUndo(): PipeFunction;
context(ctx: unknown): PipeFunction;
}
<T>
(
:
T
)
:
<
T
>
;
function isExitPipeValue<T>(value: unknown): value is ExitPipeReturnValue<T>;
function pipeContext<T, L, R, C>(
fn: (ctx: C) => (r: T, l: L) => R
): (r: T, l: L) => R;
import { Pipeline } from '@hyperse/pipeline';
interface Context {
[key: string]: any;
}
// Create a pipeline with initial middleware
const engine = new Pipeline<Context>((ctx, next) => {
ctx.foobar = 'baz';
await next();
});
// Add more middleware
engine.use(async (ctx, next) => {
await new Promise((res) => setTimeout(() => res, 2000));
ctx.another = 123;
await next();
});
// Execute the pipeline
(async () => {
const context: Context = {};
await engine.execute(context);
console.log(context);
// => { foobar: "baz", another: 123 }
})();
import { Pipeline } from '@hyperse/pipeline';
const engine = new Pipeline<Context>((ctx, next) => {
ctx.foobar = 'baz';
await next();
});
engine.use((ctx, next) => {
await new Promise((resolve, reject) => {
fs.readFile(ctx.path, {}, (err, file) => {
if (err) {
reject(err);
return;
}
// ... Do something
resolve();
});
});
await next();
});
// Error handling middleware
engine.use((ctx, next, error) => {
if (error) console.error(error.message);
await next();
});
import { pipe } from '@hyperse/pipeline';
const result = await pipe(
() => 123,
(n) => n + 1
)();
// result = 124
const result = await pipe(
(n: number) => n + 1,
(n) => n * 2
)(123);
// result = 248
import { pipe, exitPipe, isExitPipeValue } from '@hyperse/pipeline';
const result = await pipe(
() => 123,
(n) => exitPipe(n + 1),
() => 'qwe'
)();
if (isExitPipeValue(result)) {
console.log(result.r); // 124
}
import { pipe, pipeContext } from '@hyperse/pipeline';
const result = await pipe(
() => 123,
pipeContext((ctx: { n: number }) => (n) => n + ctx.n),
(n) => n + 1
).context({ n: 1 })();
// result = 125
const fn = pipe(
() => 123,
(n) => n + 1
);
// Replace a function in the pipeline
fn.replace([[0, () => 124]]);
const result1 = await fn(); // 125
// Undo the replacement
fn.replaceUndo();
const result2 = await fn(); // 124
const result = await pipe(
() => Promise.resolve(123),
(n) => n + 1
)();
// result = 124
const result = await pipe(
() => 123,
(n) => Promise.resolve(exitPipe(n + 1)),
() => 'qwe'
)();
if (isExitPipeValue(result)) {
console.log(result.r); // 124
}
import { Either } from 'monet';
const result = await pipe(
() => Either.left(123),
(n, l) => l + 1 // l contains the left value
)();
// result = 124
import { Maybe } from 'monet';
const result = await pipe(
() => Maybe.Some(123),
(n) => n + 1
)();
// result = 124
import { Either } from 'purify-ts';
const result = await pipe(
() => Either.Left(123),
(n, l) => l + 1
)();
// result = 124
const result = await pipe(
() => 123,
pipe(
(n: number) => n + 1,
(n) => exitPipe(n + 1),
() => true
),
() => 'qwe'
)();
if (isExitPipeValue(result)) {
console.log(result.r); // 125
}
const result = await pipe(
() => 123,
pipe(
(n: number) => n + 1,
pipeContext((ctx: { n: number }) => (n) => n + ctx.n)
),
(n) => n + 1
).context({ n: 1 })();
// result = 127
try {
const result = await pipe(
async () => {
throw new Error('Pipeline error');
},
(n) => n + 1
)();
} catch (error) {
console.error(error.message); // "Pipeline error"
}
install
# Run tests
yarn test
# Build the project
yarn build