import type { Evt } from "./Evt.ts";import type { StatefulEvt, UnpackEvt, NonPostableEvtLike, StatefulReadonlyEvtLike } from "./types/index.ts";import type { PromiseOrNot } from "https://deno.land/x/tsafe@v1.3.4/lab/PromiseOrNot.ts";import { importProxy } from "./importProxy.ts";
export function asyncPipe<E extends NonPostableEvtLike<any>, U>( evt: E, asyncOp: (data: UnpackEvt<E>) => PromiseOrNot<[U] | null>): E extends StatefulReadonlyEvtLike<any> ? StatefulEvt<U | undefined> : Evt<U> {
const out = "state" in evt ? importProxy.Evt.create<UnpackEvt<E> | undefined>(undefined) : importProxy.Evt.create<UnpackEvt<E>>();
let currentCallCount = 0;
evt.attach(async (data: UnpackEvt<E>) => {
currentCallCount++;
const thisCallCount = currentCallCount;
const prOpResult = asyncOp(data);
let opResult: [U] | null;
if ( prOpResult !== null && "then" in prOpResult ) {
opResult = await prOpResult;
if ( "state" in evt && thisCallCount !== currentCallCount ) { return; }
} else {
opResult = prOpResult;
}
if (!opResult) {
return; }
out.post(opResult[0] as any);
});
return out as any;
}