Typescript implementation of the latest C# version of the Pipe and Filters approach

PipeLine.ts

/**
* Type of the message all pipeline messages must inherit from.
*
*/
export abstract class BaseMessage {
    cancellationToken: CancellationToken = new CancellationToken();
    onError = () => {};
    onStart = () => {};
    onComplete = () => {};
    onSuccess = () => {};
    onStep = (s: string) => {};
}

/**
* Type representing an interface of an executbale step.
*
*/
export interface Filter<T extends BaseMessage> {
    execute(msg: T): void;
}

/**
* Type representing an interface of a cross-cutting concern.
*
*/
export interface Aspect<T extends BaseMessage> {
    next: Aspect<T>;
    execute(msg: T): void;
}

/**
* Type representing the the ability to request a pipeline to stop executing.
*
*/
export class CancellationToken {
    stop: Boolean;
}

/**
* Type representing the context that all executable steps are run under.
*
*/
export class PipeLine<T extends BaseMessage> {
    _filters: Array<Filter<T>> = [];
    _finallyFilters: Array<Filter<T>> = [];
    _aspects: Array<Aspect<T>> = [];

    constructor() {
        this.execute = this.execute.bind(this);
        this._aspects.push(new ExecutionContext(this.execute));
    }

    addAspect(aspect: Aspect<T>): void {
        this._aspects.splice(this._aspects.length - 1, 0, aspect);
        this._aspects.reduce((a, b) => a.next = b);
    }

    register(filter: Filter<T>): void {
        this._filters.push(filter);
    }

    finally(filter: Filter<T>): void {
        this._finallyFilters.push(filter);
    }

    invoke(msg: T): void {
        this._aspects[0].execute(msg);
    }

    private execute(msg: T): void {

        if (msg.onStart != null) { msg.onStart(); }
        try {

            this._filters.forEach(f => {
                if (msg.cancellationToken.stop) { return; }
                f.execute(msg);
            });

            if (msg.onSuccess != null) { msg.onSuccess(); }
        } finally {

            this._finallyFilters.forEach(f => {
                f.execute(msg);
            });

            if (msg.onComplete != null) { msg.onComplete(); }
        }
    }
}

class ExecutionContext<T extends BaseMessage> implements Aspect<T>, Filter<T> {
    inner: any;
    next: Aspect<T>;

    constructor(action: Function) {
        this.inner = action;
    }

    execute(msg: T): void {
        this.inner(msg);
    }
}