import { Cancellation } from "@implab/core-amd/Cancellation";
import { CancelledError } from "@implab/core-amd/CancelledError";
import { Observable, Sink, Subscribable, observe } from "../observable";
import { Scope } from "../tsx/Scope";

/**
 * Creates a buffer with the specified length. The buffer will immediately
 * subscribe to the source observable and start accumulating values.
 * 
 * The resulting observable will repeat the buffered values for each new subscriber.
 * 
 * @param length The number of elements to store.
 * @param ct Cancellation token to unsubscribe from the original observable.
 * 
 */
export const buffer = (length: number, ct = Cancellation.none) => <T>(source: Subscribable<T>): Observable<T> => {
    type Status = "active" | "complete" | "error";

    // ring-buffer, wpos will rotate in range (0...length-1)
    let wpos = 0;
    const buffer: T[] = [];

    // writes the next value to the buffer
    const write = (value: T) => {
        buffer[wpos] = value;
        wpos = (wpos + 1) % length;
    };

    // reads contents of the buffer
    // cb will be called for each value in the buffer
    const read = (cb: (item: T) => void) => {
        const start = wpos + length - buffer.length;
        const end = wpos + length;

        for(let pos = start; pos < end; pos++ )
            cb(buffer[pos % length]);
    };

    let status: Status = "active";
    let lastError: unknown;
    let subscribers: Sink<T>[] = [];

    const scope = new Scope();

    // cleanup method to release resources held by this subscription
    const cleanup = (cb: (item: Sink<T>) => void) => {
        scope.destroy();
        const _subscribers = subscribers;
        subscribers = [];
        _subscribers.forEach(cb);
    };

    const sink: Sink<T> = {
        isClosed: () => status !== "active",
        complete: () => {
            if (status === "active") {
                status = "complete";
                cleanup(s => s.complete());
            }
        },
        error: e => {
            if (status === "active") {
                status = "error";
                lastError = e;
                cleanup(s => s.error(e));
            }
        },
        next: v => {
            if (status === "active") {
                write(v);
                const _subscribers = subscribers;
                _subscribers.forEach(s => s.next(v));
            }
        }
    };

    if (ct.isRequested()) {
        sink.error(new CancelledError("The operation was cancelled", ct));
    } else {
        scope.own(source.subscribe(sink));
        scope.own(ct.register(e => sink.error(e)));
    }

    return observe( s => {
        const _subscribers = subscribers;
        read(s.next);
        switch (status) {
            case "active":
                subscribers.push(s);
                return () => {
                    if (_subscribers === subscribers) {
                        const pos = subscribers.indexOf(s);
                        if (pos >= 0)
                            subscribers.splice(pos, 1);
                    }
                };
            case "complete":
                s.complete();
                break;
            case "error":
                s.error(lastError);
                break;
        }
    });
};