Add last() method

src/index.spec.ts

@@ -18,6 +18,7 @@ import {
     duplex,
     child,
     reduce,
+    last,
 } from ".";
 
 test.cb("fromArray() streams array elements in flowing mode", t => {
@@ -1065,3 +1066,14 @@ test.cb(
         source.push(null);
     },
 );
+
+test("last() yields the last chunk streamed by the given readable stream", async t => {
+    const source = new Readable({ objectMode: true });
+    const lastPromise = last(source);
+    source.push("ab");
+    source.push("cd");
+    source.push("ef");
+    source.push(null);
+    const lastChunk = await lastPromise;
+    expect(lastChunk).to.equal("ef");
+});

src/index.ts

@@ -427,3 +427,16 @@ export function duplex(writable: Writable, readable: Readable) {
 export function child(childProcess: ChildProcess) {
     return duplex(childProcess.stdin, childProcess.stdout);
 }
+
+/**
+ * Resolve the last streamed chunk of the given readable stream, after it has ended
+ * @param readable The readable stream to wait on
+ */
+export function last<T>(readable: Readable): Promise<T | null> {
+    let lastChunk: T | null = null;
+    return new Promise((resolve, reject) => {
+        readable
+            .on("data", chunk => (lastChunk = chunk))
+            .on("end", () => resolve(lastChunk));
+    });
+}