AsyncPipe; table schemata; migrations; File logging

src/util/support/Pipe.ts

@@ -1,12 +1,14 @@
 /**
  * A closure that maps a given pipe item to a different type.
  */
+import {Awaitable} from './types'
+
 export type PipeOperator<T, T2> = (subject: T) => T2
 
 /**
  * A closure that maps a given pipe item to an item of the same type.
  */
-export type ReflexivePipeOperator<T> = (subject: T) => T
+export type ReflexivePipeOperator<T> = (subject: T) => T|void
 
 /**
  * A condition or condition-resolving function for pipe methods.
@@ -77,6 +79,15 @@ export class Pipe<T> {
         return new Pipe(op(this.subject))
     }
 
+    /**
+     * Like tap, but always returns the original pipe.
+     * @param op
+     */
+    peek<T2>(op: PipeOperator<T, T2>): this {
+        op(this.subject)
+        return this
+    }
+
     /**
      * If `check` is truthy, apply the given operator to the item in the pipe and return the result.
      * Otherwise, just return the current pipe unchanged.
@@ -86,7 +97,7 @@ export class Pipe<T> {
      */
     when(check: PipeCondition<T>, op: ReflexivePipeOperator<T>): Pipe<T> {
         if ( (typeof check === 'function' && check(this.subject)) || check ) {
-            return Pipe.wrap(op(this.subject))
+            Pipe.wrap(op(this.subject))
         }
 
         return this
@@ -104,7 +115,8 @@ export class Pipe<T> {
             return this
         }
 
-        return Pipe.wrap(op(this.subject))
+        Pipe.wrap(op(this.subject))
+        return this
     }
 
     /**
@@ -127,4 +139,134 @@ export class Pipe<T> {
     get(): T {
         return this.subject
     }
+
+    /**
+     * Get an AsyncPipe with the current item in the pipe.
+     */
+    async(): AsyncPipe<T> {
+        return AsyncPipe.wrap<T>(this.subject)
+    }
+}
+
+/**
+ * A subject function that yields the value in the AsyncPipe.
+ */
+export type AsyncPipeResolver<T> = () => Awaitable<T>
+
+/**
+ * A closure that maps a given pipe item to a different type.
+ */
+export type AsyncPipeOperator<T, T2> = (subject: T) => Awaitable<T2>
+
+/**
+ * A closure that maps a given pipe item to an item of the same type.
+ */
+export type ReflexiveAsyncPipeOperator<T> = (subject: T) => Awaitable<T|void>
+
+/**
+ * A condition or condition-resolving function for pipe methods.
+ */
+export type AsyncPipeCondition<T> = boolean | ((subject: T) => Awaitable<boolean>)
+
+/**
+ * An asynchronous version of the Pipe helper.
+ */
+export class AsyncPipe<T> {
+    /**
+     * Get an AsyncPipe with the given value in it.
+     * @param subject
+     */
+    static wrap<subjectType>(subject: subjectType): AsyncPipe<subjectType> {
+        return new AsyncPipe<subjectType>(() => subject)
+    }
+
+    constructor(
+        /** The current value resolver of the pipe. */
+        private subject: AsyncPipeResolver<T>,
+    ) {}
+
+    /**
+     * Apply a transformative operator to the pipe.
+     * @param op
+     */
+    tap<T2>(op: AsyncPipeOperator<T, T2>): AsyncPipe<T2> {
+        return new AsyncPipe<T2>(async () => op(await this.subject()))
+    }
+
+    /**
+     * Apply an operator to the pipe, but return the reference
+     * to the current pipe. The operator is resolved when the
+     * overall pipe is resolved.
+     * @param op
+     */
+    peek<T2>(op: AsyncPipeOperator<T, T2>): AsyncPipe<T> {
+        return new AsyncPipe<T>(async () => {
+            const subject = await this.subject()
+            await op(subject)
+            return subject
+        })
+    }
+
+    /**
+     * Apply an operator to the pipe, if the check condition passes.
+     * @param check
+     * @param op
+     */
+    when(check: AsyncPipeCondition<T>, op: ReflexiveAsyncPipeOperator<T>): AsyncPipe<T> {
+        return new AsyncPipe<T>(async () => {
+            let subject
+
+            if ( typeof check === 'function' ) {
+                check = await check(subject = await this.subject())
+            }
+
+            subject = subject ?? await this.subject()
+            if ( check ) {
+                return ((await op(subject)) ?? subject) as T
+            }
+
+            return subject as T
+        })
+    }
+
+    /**
+     * Apply an operator to the pipe, if the check condition fails.
+     * @param check
+     * @param op
+     */
+    unless(check: AsyncPipeCondition<T>, op: ReflexiveAsyncPipeOperator<T>): AsyncPipe<T> {
+        if ( typeof check === 'function' ) {
+            return this.when(async (subject: T) => !(await check(subject)), op)
+        }
+
+        return this.when(!check, op)
+    }
+
+    /**
+     * Alias of `unless()`.
+     * @param check
+     * @param op
+     */
+    whenNot(check: AsyncPipeCondition<T>, op: ReflexiveAsyncPipeOperator<T>): AsyncPipe<T> {
+        return this.unless(check, op)
+    }
+
+    /**
+     * Get the transformed value from the pipe.
+     */
+    async resolve(): Promise<T> {
+        return this.subject()
+    }
+
+    /**
+     * Resolve the value and return it in a sync `Pipe` instance.
+     */
+    async sync(): Promise<Pipe<T>> {
+        return Pipe.wrap<T>(await this.subject())
+    }
+
+    /** Get the transformed value from the pipe. Allows awaiting the pipe directly. */
+    then(): Promise<T> {
+        return this.resolve()
+    }
 }