From 789fb49e8beb42be565f62582b1896d9838f41bb Mon Sep 17 00:00:00 2001 From: didier amyot Date: Mon, 12 Jun 2023 19:31:07 -0400 Subject: [PATCH] First workable server. --- .gitignore | 2 + src/fileservice.ts | 108 ++++++----- src/restapi.ts | 32 ++-- src/utils.ts | 3 - tests/fileservice_test.ts | 16 +- .../deno.land/std@0.191.0/async/deferred.ts | 48 +++++ .../deno.land/std@0.191.0/streams/buffer.ts | 169 ++++++++++++++++++ .../std@0.191.0/streams/byte_slice_stream.ts | 45 +++++ vendor/deno.land/std@0.191.0/streams/copy.ts | 48 +++++ .../std@0.191.0/streams/delimiter_stream.ts | 114 ++++++++++++ .../streams/early_zip_readable_streams.ts | 34 ++++ .../streams/limited_bytes_transform_stream.ts | 37 ++++ .../streams/limited_transform_stream.ts | 33 ++++ .../streams/merge_readable_streams.ts | 42 +++++ vendor/deno.land/std@0.191.0/streams/mod.ts | 31 ++++ .../streams/readable_stream_from_iterable.ts | 63 +++++++ .../streams/readable_stream_from_reader.ts | 81 +++++++++ .../streams/reader_from_iterable.ts | 56 ++++++ .../streams/reader_from_stream_reader.ts | 43 +++++ .../streams/text_delimiter_stream.ts | 78 ++++++++ .../std@0.191.0/streams/text_line_stream.ts | 74 ++++++++ .../streams/to_transform_stream.ts | 75 ++++++++ .../streams/writable_stream_from_writer.ts | 52 ++++++ .../std@0.191.0/streams/write_all.ts | 65 +++++++ .../streams/writer_from_stream_writer.ts | 35 ++++ .../streams/zip_readable_streams.ts | 35 ++++ vendor/import_map.json | 4 +- 27 files changed, 1355 insertions(+), 68 deletions(-) create mode 100644 vendor/deno.land/std@0.191.0/async/deferred.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/buffer.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/byte_slice_stream.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/copy.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/delimiter_stream.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/early_zip_readable_streams.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/limited_bytes_transform_stream.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/limited_transform_stream.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/merge_readable_streams.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/mod.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/readable_stream_from_iterable.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/readable_stream_from_reader.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/reader_from_iterable.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/reader_from_stream_reader.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/text_delimiter_stream.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/text_line_stream.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/to_transform_stream.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/writable_stream_from_writer.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/write_all.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/writer_from_stream_writer.ts create mode 100644 vendor/deno.land/std@0.191.0/streams/zip_readable_streams.ts diff --git a/.gitignore b/.gitignore index 45dae1c..9908441 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ .vscode .direnv result +foo +playground.ts diff --git a/src/fileservice.ts b/src/fileservice.ts index 487ba88..39e5c62 100644 --- a/src/fileservice.ts +++ b/src/fileservice.ts @@ -1,7 +1,14 @@ -import { sha1 } from 'https://denopkg.com/chiefbiiko/sha1/mod.ts'; -import { iterateReader } from "https://deno.land/std@0.191.0/streams/iterate_reader.ts"; import { readAll } from "https://deno.land/std@0.191.0/streams/read_all.ts"; -import { readerFromIterable } from "https://deno.land/std@0.144.0/streams/conversion.ts"; +import { readerFromIterable, iterateReader } from "https://deno.land/std@0.191.0/streams/mod.ts"; +import { Buffer } from "https://deno.land/std@0.191.0/io/buffer.ts"; + +import { StringReader } from "https://deno.land/std@0.191.0/io/string_reader.ts"; + + +import { readerFromStreamReader, readableStreamFromReader } from "https://deno.land/std@0.191.0/streams/mod.ts"; + + + export interface FileService { @@ -29,13 +36,14 @@ export interface FileService { newPath: string, ): Promise; - deleteByPath(path: string): Promise; + deleteByPath(path: string): Promise; } interface SerializedFile { id: string; path: string; bodySize: number; + compression: string; } export interface FileMetaData { @@ -57,6 +65,14 @@ export class DenoKVBlobService { this.blobSize = blobSize; } + private async sha1(msgUint8: Uint8Array): Promise { + const hashBuffer = await crypto.subtle.digest("SHA-1", msgUint8); + const hashArray = Array.from(new Uint8Array(hashBuffer)); + return hashArray + .map((b) => b.toString(16).padStart(2, "0")) + .join(""); // convert bytes to hex string + } + private *splitText(body: Uint8Array): Generator { let blobCount = 0; let txt = body.slice( @@ -77,10 +93,10 @@ export class DenoKVBlobService { if(body.length > this.blobSize){ throw new Error('Persisted a too large blob'); } - const blob_id = sha1(body, 'utf8', 'hex'); // todo, should not be utf8 + + const blob_id = await this.sha1(body); await this.kv.set(['blob_index', mainIndex, index], blob_id); await this.kv.set(['blob', blob_id], body); - } async persistBlob(index: string, body: Uint8Array): Promise { @@ -98,7 +114,7 @@ export class DenoKVBlobService { let i = 0; let size = 0; for await (const chunk of iterateReader(body, {bufSize: this.blobSize})){ - this.unsafePersistBlob(index, i, chunk); + await this.unsafePersistBlob(index, i, chunk); size += chunk.length; i++; } @@ -162,36 +178,29 @@ export class DenoKVFileService implements FileService { constructor(kv: Deno.Kv) { this.kv = kv; - this.blobService = new DenoKVBlobService(kv, 500); + this.blobService = new DenoKVBlobService(kv, 32 * 1024); } - async createFromFullBody( + createFromFullBody( path: string, body: string | Uint8Array, ): Promise { - const newId = crypto.randomUUID(); - const bodyBytes: Uint8Array = body instanceof Uint8Array ? body : new TextEncoder().encode(body); - const size = await this.blobService.persistBlob(newId, bodyBytes); - - const serializedFile: SerializedFile = { - id: newId, - path, - bodySize: size, - }; - const idx: SerializedFilePathIndex = { fileId: newId }; - await this.kv.set(['file', newId], serializedFile); - await this.kv.set(['file_index_path', path], idx); - - return { id: newId, path, bodySize: size }; + if (body instanceof Uint8Array){ + return this.createFromReader(path, new Buffer(body)); + } + return this.createFromReader(path, new StringReader(body)); } - async createFromReader(path: string, body: Deno.Reader): Promise{ + async createFromReader(path: string, content: Deno.Reader): Promise{ const newId = crypto.randomUUID(); - const size = await this.blobService.persistReader(newId, body); + + const compressedContent = readableStreamFromReader(content).pipeThrough(new CompressionStream('gzip')).getReader() + const size = await this.blobService.persistReader(newId, readerFromStreamReader(compressedContent)); const serializedFile: SerializedFile = { id: newId, path, bodySize: size, + compression: 'gzip' }; const idx: SerializedFilePathIndex = { fileId: newId }; await this.kv.set(['file', newId], serializedFile); @@ -221,13 +230,13 @@ export class DenoKVFileService implements FileService { } } - async deleteByPath(path: string): Promise { + async deleteByPath(path: string): Promise { const idx = await this.kv.get([ 'file_index_path', path, ]); if (idx.value === null) { - return; + return false; } // todo, this will fail silently if transaction fails. we might want to put that into a loop. await this.kv.atomic() @@ -235,6 +244,22 @@ export class DenoKVFileService implements FileService { .delete(['file_index_path', path]) .delete(idx.key) .commit(); + return true; + } + + private async getSerializedFile(fileId: string): Promise{ + const ret = await this.kv.get(['file', fileId]); + return ret.value + } + + private decompressBody(serializedFile: SerializedFile, content: Deno.Reader): Deno.Reader { + if(serializedFile.compression == 'gzip'){ + const ret = readableStreamFromReader(content).pipeThrough(new DecompressionStream('gzip')).getReader(); + return readerFromStreamReader(ret); + + } + return content; + } async getFullBodyByPath(path: string): Promise { @@ -243,14 +268,8 @@ export class DenoKVFileService implements FileService { } async getRawBodyByPath(path: string): Promise { - const idx = await this.kv.get([ - 'file_index_path', - path, - ]); - if (idx.value === null) { - return null; - } - return this.blobService.readAllBlob(idx.value.fileId); + const ret = await this.getBodyAsReader(path); + return ret === null ? null : readAll(ret); } @@ -262,19 +281,23 @@ export class DenoKVFileService implements FileService { if (idx.value === null) { return null; } - return this.blobService.getBlob(idx.value.fileId); + const ser = await this.kv.get(['file', idx.value.fileId]); + if(ser.value === null){ + return null; + } + return this.decompressBody(ser.value, this.blobService.getBlob(idx.value.fileId)); } async getMetaDataByID(id: string): Promise { - const fileObject = await this.kv.get(['file', id]); - if (fileObject.value === null) { + const fileObject = await this.getSerializedFile(id); + if (fileObject === null) { return null; } return { - path: fileObject.value.path, - id: fileObject.value.id, - bodySize: fileObject.value.bodySize, + path: fileObject.path, + id: fileObject.id, + bodySize: fileObject.bodySize, }; } @@ -287,10 +310,11 @@ export class DenoKVFileService implements FileService { 'file_index_path', fileObject.value.path, ]); - const newFile: FileMetaData = { + const newFile: SerializedFile = { path: newPath, id: fileObject.value.id, bodySize: fileObject.value.bodySize, + compression: fileObject.value.compression, }; await this.kv.atomic() .check(fileObject) diff --git a/src/restapi.ts b/src/restapi.ts index ff1c802..e83cdec 100644 --- a/src/restapi.ts +++ b/src/restapi.ts @@ -1,10 +1,7 @@ import { Context, Router } from "https://deno.land/x/oak/mod.ts"; import { FileService } from "./fileservice.ts"; -import { readerFromStreamReader } from "https://deno.land/std@0.189.0/streams/mod.ts"; +import { readerFromStreamReader, readableStreamFromReader, TextLineStream } from "https://deno.land/std@0.191.0/streams/mod.ts"; import { asyncIterableToArray } from "./utils.ts"; -import { StringReader } from "https://deno.land/std@0.178.0/io/string_reader.ts"; -import { isReader } from "https://deno.land/x/oak@v12.5.0/util.ts"; -import { createMockNext } from "https://deno.land/x/oak@v12.1.0/testing.ts"; type Handler = (ctx: Context) => void; @@ -17,7 +14,6 @@ function listFileHandlerFactory(fileService: FileService): Handler { }; } - function createFileHandlerFactory(fileService: FileService): Handler { return async (ctx: Context) => { const untypedCtx: any = ctx; @@ -26,33 +22,43 @@ function createFileHandlerFactory(fileService: FileService): Handler { const reqBody = await ctx.request.body({ type: "stream" }).value; const reader = readerFromStreamReader(reqBody.getReader()) - fileService.createFromReader(path, reader); + await fileService.createFromReader(path, reader); }; } - function getFileHandlerFactory(fileService: FileService): Handler { return async (ctx: Context) => { const untypedCtx: any = ctx; - const fileId = untypedCtx.params.file_id; + const fileId = untypedCtx.params.path; const reader = await fileService.getBodyAsReader(fileId); if(reader == null){ ctx.response.status = 404; return; } ctx.response.status = 200; - ctx.response.body = reader; - ctx.response.type = "text/plain"; - // ctx.send(reader) + ctx.response.body = reader; + }; +} + +function deleteFileHandlerFactory(fileService: FileService): Handler { + return async (ctx: Context) => { + const untypedCtx: any = ctx; + const path = untypedCtx.params.path; + const isDeleted = await fileService.deleteByPath(path); + if(!isDeleted){ + ctx.response.status = 404; + return; + } + ctx.response.status = 200; }; } export function buildAPI(router: Router, fileService: FileService): Router { - router.get('/files', listFileHandlerFactory(fileService)); - router.get('/files/:file_id', getFileHandlerFactory(fileService)); + router.get('/files/:path', getFileHandlerFactory(fileService)); + router.delete('/file/:path', deleteFileHandlerFactory(fileService)); router.post('/files', createFileHandlerFactory(fileService)); return router; diff --git a/src/utils.ts b/src/utils.ts index f4d826e..d83f22b 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -18,7 +18,6 @@ export class AsyncIterableReader implements Deno.Reader { private readonly source: AsyncIterator; private interReadBuffer = new Uint8Array(); private isSourceEmpty = false; - private shit = 0; constructor(source: AsyncIterable) { this.source = source[Symbol.asyncIterator](); } @@ -58,8 +57,6 @@ export class AsyncIterableReader implements Deno.Reader { } async read(buffer: Uint8Array): Promise { - this.shit += 1; - console.log("reading \> " + this.shit); if(this.isSourceEmpty && this.interReadBuffer.length < 1){ return null; } diff --git a/tests/fileservice_test.ts b/tests/fileservice_test.ts index 7cb5e2f..c000ac3 100644 --- a/tests/fileservice_test.ts +++ b/tests/fileservice_test.ts @@ -8,6 +8,7 @@ import { } from "../src/fileservice.ts"; import { asyncIterableToArray } from "../src/utils.ts"; import { generateLargeText } from "./utils.ts"; +import { StringReader } from "https://deno.land/std@0.191.0/io/mod.ts"; @@ -51,7 +52,7 @@ Deno.test("DenoKVBlobService happy path", async (t) => { kv.close(); }); -Deno.test("DenoKVFileService happy path", async (t) => { +Deno.test("DenoKVFileService happy path xxx ", async (t) => { const kv = await Deno.openKv(":memory:"); const fs: FileService = new DenoKVFileService(kv); const fileName1 = "didier/file1.txt"; @@ -64,7 +65,7 @@ Deno.test("DenoKVFileService happy path", async (t) => { file2 = await fs.createFromFullBody(fileName2, "le geant vert ☀️"); assertEquals(file1.path, fileName1); - assertEquals(file2.bodySize, 20); + assertEquals(file2.bodySize, 47); }); await t.step("Get file metadata", async () => { @@ -99,11 +100,12 @@ Deno.test("DenoKVFileService happy path", async (t) => { await t.step("Create File from reader", async () => { const newPath = "boris/streamed-file.txt"; - const file = await Deno.open('tests/examples/teapot.obj'); - const ret = await fs.createFromReader(newPath, file); - assertEquals(ret.bodySize, 51493); - file.close(); + const ret = await fs.createFromReader(newPath, new StringReader("hello world 18")); + assertEquals(ret.bodySize, 40); + + const s = await fs.getRawBodyByPath(newPath); + assertEquals(new TextDecoder().decode(s!), "hello world 18") }); await t.step("Rename file", async () => { @@ -128,7 +130,7 @@ Deno.test("DenoKVFileService happy path with byte array", async (t) => { const file = await fs.createFromFullBody("hello", inputBody); assertEquals(file.path, "hello"); - assertEquals(file.bodySize, inputBody.length); + assertEquals(file.bodySize, 27); const bodyString = await fs.getFullBodyByPath("hello"); assertEquals(bodyString, new TextDecoder().decode(inputBody)); diff --git a/vendor/deno.land/std@0.191.0/async/deferred.ts b/vendor/deno.land/std@0.191.0/async/deferred.ts new file mode 100644 index 0000000..cd6c0aa --- /dev/null +++ b/vendor/deno.land/std@0.191.0/async/deferred.ts @@ -0,0 +1,48 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +// TODO(ry) It'd be better to make Deferred a class that inherits from +// Promise, rather than an interface. This is possible in ES2016, however +// typescript produces broken code when targeting ES5 code. +// See https://github.com/Microsoft/TypeScript/issues/15202 +// At the time of writing, the github issue is closed but the problem remains. +export interface Deferred extends Promise { + readonly state: "pending" | "fulfilled" | "rejected"; + resolve(value?: T | PromiseLike): void; + // deno-lint-ignore no-explicit-any + reject(reason?: any): void; +} + +/** + * Creates a Promise with the `reject` and `resolve` functions placed as methods + * on the promise object itself. + * + * @example + * ```typescript + * import { deferred } from "https://deno.land/std@$STD_VERSION/async/deferred.ts"; + * + * const p = deferred(); + * // ... + * p.resolve(42); + * ``` + */ +export function deferred(): Deferred { + let methods; + let state = "pending"; + const promise = new Promise((resolve, reject) => { + methods = { + async resolve(value: T | PromiseLike) { + await value; + state = "fulfilled"; + resolve(value); + }, + // deno-lint-ignore no-explicit-any + reject(reason?: any) { + state = "rejected"; + reject(reason); + }, + }; + }); + Object.defineProperty(promise, "state", { get: () => state }); + return Object.assign(promise, methods) as Deferred; +} diff --git a/vendor/deno.land/std@0.191.0/streams/buffer.ts b/vendor/deno.land/std@0.191.0/streams/buffer.ts new file mode 100644 index 0000000..74f4633 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/buffer.ts @@ -0,0 +1,169 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import { assert } from "../_util/asserts.ts"; +import { copy } from "../bytes/copy.ts"; + +const MAX_SIZE = 2 ** 32 - 2; +const DEFAULT_CHUNK_SIZE = 16_640; + +/** A variable-sized buffer of bytes with `read()` and `write()` methods. + * + * Buffer is almost always used with some I/O like files and sockets. It allows + * one to buffer up a download from a socket. Buffer grows and shrinks as + * necessary. + * + * Buffer is NOT the same thing as Node's Buffer. Node's Buffer was created in + * 2009 before JavaScript had the concept of ArrayBuffers. It's simply a + * non-standard ArrayBuffer. + * + * ArrayBuffer is a fixed memory allocation. Buffer is implemented on top of + * ArrayBuffer. + * + * Based on [Go Buffer](https://golang.org/pkg/bytes/#Buffer). */ +export class Buffer { + #buf: Uint8Array; // contents are the bytes buf[off : len(buf)] + #off = 0; // read at buf[off], write at buf[buf.byteLength] + #readable: ReadableStream = new ReadableStream({ + type: "bytes", + pull: (controller) => { + const view = new Uint8Array(controller.byobRequest!.view!.buffer); + if (this.empty()) { + // Buffer is empty, reset to recover space. + this.reset(); + controller.close(); + controller.byobRequest!.respond(0); + return; + } + const nread = copy(this.#buf.subarray(this.#off), view); + this.#off += nread; + controller.byobRequest!.respond(nread); + }, + autoAllocateChunkSize: DEFAULT_CHUNK_SIZE, + }); + get readable() { + return this.#readable; + } + #writable = new WritableStream({ + write: (chunk) => { + const m = this.#grow(chunk.byteLength); + copy(chunk, this.#buf, m); + }, + }); + get writable() { + return this.#writable; + } + + constructor(ab?: ArrayBufferLike | ArrayLike) { + this.#buf = ab === undefined ? new Uint8Array(0) : new Uint8Array(ab); + } + + /** Returns a slice holding the unread portion of the buffer. + * + * The slice is valid for use only until the next buffer modification (that + * is, only until the next call to a method like `read()`, `write()`, + * `reset()`, or `truncate()`). If `options.copy` is false the slice aliases + * the buffer content at least until the next buffer modification, so + * immediate changes to the slice will affect the result of future reads. + */ + bytes(options = { copy: true }): Uint8Array { + if (options.copy === false) return this.#buf.subarray(this.#off); + return this.#buf.slice(this.#off); + } + + /** Returns whether the unread portion of the buffer is empty. */ + empty(): boolean { + return this.#buf.byteLength <= this.#off; + } + + /** A read only number of bytes of the unread portion of the buffer. */ + get length(): number { + return this.#buf.byteLength - this.#off; + } + + /** The read only capacity of the buffer's underlying byte slice, that is, + * the total space allocated for the buffer's data. */ + get capacity(): number { + return this.#buf.buffer.byteLength; + } + + /** Discards all but the first `n` unread bytes from the buffer but + * continues to use the same allocated storage. It throws if `n` is + * negative or greater than the length of the buffer. */ + truncate(n: number) { + if (n === 0) { + this.reset(); + return; + } + if (n < 0 || n > this.length) { + throw Error("bytes.Buffer: truncation out of range"); + } + this.#reslice(this.#off + n); + } + + reset() { + this.#reslice(0); + this.#off = 0; + } + + #tryGrowByReslice(n: number) { + const l = this.#buf.byteLength; + if (n <= this.capacity - l) { + this.#reslice(l + n); + return l; + } + return -1; + } + + #reslice(len: number) { + assert(len <= this.#buf.buffer.byteLength); + this.#buf = new Uint8Array(this.#buf.buffer, 0, len); + } + + #grow(n: number) { + const m = this.length; + // If buffer is empty, reset to recover space. + if (m === 0 && this.#off !== 0) { + this.reset(); + } + // Fast: Try to grow by means of a reslice. + const i = this.#tryGrowByReslice(n); + if (i >= 0) { + return i; + } + const c = this.capacity; + if (n <= Math.floor(c / 2) - m) { + // We can slide things down instead of allocating a new + // ArrayBuffer. We only need m+n <= c to slide, but + // we instead let capacity get twice as large so we + // don't spend all our time copying. + copy(this.#buf.subarray(this.#off), this.#buf); + } else if (c + n > MAX_SIZE) { + throw new Error("The buffer cannot be grown beyond the maximum size."); + } else { + // Not enough space anywhere, we need to allocate. + const buf = new Uint8Array(Math.min(2 * c + n, MAX_SIZE)); + copy(this.#buf.subarray(this.#off), buf); + this.#buf = buf; + } + // Restore this.#off and len(this.#buf). + this.#off = 0; + this.#reslice(Math.min(m + n, MAX_SIZE)); + return m; + } + + /** Grows the buffer's capacity, if necessary, to guarantee space for + * another `n` bytes. After `.grow(n)`, at least `n` bytes can be written to + * the buffer without another allocation. If `n` is negative, `.grow()` will + * throw. If the buffer can't grow it will throw an error. + * + * Based on Go Lang's + * [Buffer.Grow](https://golang.org/pkg/bytes/#Buffer.Grow). */ + grow(n: number) { + if (n < 0) { + throw Error("Buffer.grow: negative count"); + } + const m = this.#grow(n); + this.#reslice(m); + } +} diff --git a/vendor/deno.land/std@0.191.0/streams/byte_slice_stream.ts b/vendor/deno.land/std@0.191.0/streams/byte_slice_stream.ts new file mode 100644 index 0000000..545d870 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/byte_slice_stream.ts @@ -0,0 +1,45 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import { assert } from "../_util/asserts.ts"; + +/** + * A transform stream that only transforms from the zero-indexed `start` and `end` bytes (both inclusive). + * + * @example + * ```ts + * import { ByteSliceStream } from "https://deno.land/std@$STD_VERSION/streams/byte_slice_stream.ts"; + * const response = await fetch("https://example.com"); + * const rangedStream = response.body! + * .pipeThrough(new ByteSliceStream(3, 8)); + * ``` + */ +export class ByteSliceStream extends TransformStream { + #offsetStart = 0; + #offsetEnd = 0; + + constructor(start = 0, end = Infinity) { + super({ + start: () => { + assert(start >= 0, "`start` must be greater than 0"); + end += 1; + }, + transform: (chunk, controller) => { + this.#offsetStart = this.#offsetEnd; + this.#offsetEnd += chunk.byteLength; + if (this.#offsetEnd > start) { + if (this.#offsetStart < start) { + chunk = chunk.slice(start - this.#offsetStart); + } + if (this.#offsetEnd >= end) { + chunk = chunk.slice(0, chunk.byteLength - this.#offsetEnd + end); + controller.enqueue(chunk); + controller.terminate(); + } else { + controller.enqueue(chunk); + } + } + }, + }); + } +} diff --git a/vendor/deno.land/std@0.191.0/streams/copy.ts b/vendor/deno.land/std@0.191.0/streams/copy.ts new file mode 100644 index 0000000..a2bea5c --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/copy.ts @@ -0,0 +1,48 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import { DEFAULT_BUFFER_SIZE } from "./_common.ts"; +import type { Reader, Writer } from "../types.d.ts"; + +/** Copies from `src` to `dst` until either EOF (`null`) is read from `src` or + * an error occurs. It resolves to the number of bytes copied or rejects with + * the first error encountered while copying. + * + * ```ts + * import { copy } from "https://deno.land/std@$STD_VERSION/streams/copy.ts"; + * + * const source = await Deno.open("my_file.txt"); + * const bytesCopied1 = await copy(source, Deno.stdout); + * const destination = await Deno.create("my_file_2.txt"); + * const bytesCopied2 = await copy(source, destination); + * ``` + * + * @param src The source to copy from + * @param dst The destination to copy to + * @param options Can be used to tune size of the buffer. Default size is 32kB + */ +export async function copy( + src: Reader, + dst: Writer, + options?: { + bufSize?: number; + }, +): Promise { + let n = 0; + const bufSize = options?.bufSize ?? DEFAULT_BUFFER_SIZE; + const b = new Uint8Array(bufSize); + let gotEOF = false; + while (gotEOF === false) { + const result = await src.read(b); + if (result === null) { + gotEOF = true; + } else { + let nwritten = 0; + while (nwritten < result) { + nwritten += await dst.write(b.subarray(nwritten, result)); + } + n += nwritten; + } + } + return n; +} diff --git a/vendor/deno.land/std@0.191.0/streams/delimiter_stream.ts b/vendor/deno.land/std@0.191.0/streams/delimiter_stream.ts new file mode 100644 index 0000000..eb89f2c --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/delimiter_stream.ts @@ -0,0 +1,114 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import { BytesList } from "../bytes/bytes_list.ts"; +import { createLPS } from "./_common.ts"; + +/** Disposition of the delimiter. */ +export type DelimiterDisposition = + /** Include delimiter in the found chunk. */ + | "suffix" + /** Include delimiter in the subsequent chunk. */ + | "prefix" + /** Discard the delimiter. */ + | "discard" // delimiter discarded +; + +export interface DelimiterStreamOptions { + /** Disposition of the delimiter. */ + disposition?: DelimiterDisposition; +} + +/** + * Divide a stream into chunks delimited by a given byte sequence. + * + * @example + * Divide a CSV stream by commas, discarding the commas: + * ```ts + * import { DelimiterStream } from "https://deno.land/std@$STD_VERSION/streams/delimiter_stream.ts"; + * const res = await fetch("https://example.com/data.csv"); + * const parts = res.body! + * .pipeThrough(new DelimiterStream(new TextEncoder().encode(","))) + * .pipeThrough(new TextDecoderStream()); + * ``` + * + * @example + * Divide a stream after semi-colons, keeping the semi-colons in the output: + * ```ts + * import { DelimiterStream } from "https://deno.land/std@$STD_VERSION/streams/delimiter_stream.ts"; + * const res = await fetch("https://example.com/file.js"); + * const parts = res.body! + * .pipeThrough( + * new DelimiterStream( + * new TextEncoder().encode(";"), + * { disposition: "suffix" }, + * ) + * ) + * .pipeThrough(new TextDecoderStream()); + * ``` + * + * @param delimiter Delimiter byte sequence + * @param options Options for the transform stream + * @returns Transform stream + */ +export class DelimiterStream extends TransformStream { + #bufs = new BytesList(); + #delimiter: Uint8Array; + #inspectIndex = 0; + #matchIndex = 0; + #delimLen: number; + #delimLPS: Uint8Array; + #disp?: DelimiterDisposition; + + constructor( + delimiter: Uint8Array, + options?: DelimiterStreamOptions, + ) { + super({ + transform: (chunk, controller) => { + this.#handle(chunk, controller); + }, + flush: (controller) => { + controller.enqueue(this.#bufs.concat()); + }, + }); + + this.#delimiter = delimiter; + this.#delimLen = delimiter.length; + this.#delimLPS = createLPS(delimiter); + this.#disp = options?.disposition ?? "discard"; + } + + #handle( + chunk: Uint8Array, + controller: TransformStreamDefaultController, + ) { + this.#bufs.add(chunk); + let localIndex = 0; + while (this.#inspectIndex < this.#bufs.size()) { + if (chunk[localIndex] === this.#delimiter[this.#matchIndex]) { + this.#inspectIndex++; + localIndex++; + this.#matchIndex++; + if (this.#matchIndex === this.#delimLen) { + // Full match + const start = this.#inspectIndex - this.#delimLen; + const end = this.#disp == "suffix" ? this.#inspectIndex : start; + const copy = this.#bufs.slice(0, end); + controller.enqueue(copy); + const shift = this.#disp == "prefix" ? start : this.#inspectIndex; + this.#bufs.shift(shift); + this.#inspectIndex = this.#disp == "prefix" ? this.#delimLen : 0; + this.#matchIndex = 0; + } + } else { + if (this.#matchIndex === 0) { + this.#inspectIndex++; + localIndex++; + } else { + this.#matchIndex = this.#delimLPS[this.#matchIndex - 1]; + } + } + } + } +} diff --git a/vendor/deno.land/std@0.191.0/streams/early_zip_readable_streams.ts b/vendor/deno.land/std@0.191.0/streams/early_zip_readable_streams.ts new file mode 100644 index 0000000..fd27a52 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/early_zip_readable_streams.ts @@ -0,0 +1,34 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +/** + * Merge multiple streams into a single one, taking order into account, and each stream + * will wait for a chunk to enqueue before the next stream can append another chunk. + * If a stream ends before other ones, the others will be cancelled. + */ +export function earlyZipReadableStreams( + ...streams: ReadableStream[] +): ReadableStream { + const readers = streams.map((s) => s.getReader()); + return new ReadableStream({ + async start(controller) { + try { + loop: + while (true) { + for (const reader of readers) { + const { value, done } = await reader.read(); + if (!done) { + controller.enqueue(value!); + } else { + await Promise.all(readers.map((reader) => reader.cancel())); + break loop; + } + } + } + controller.close(); + } catch (e) { + controller.error(e); + } + }, + }); +} diff --git a/vendor/deno.land/std@0.191.0/streams/limited_bytes_transform_stream.ts b/vendor/deno.land/std@0.191.0/streams/limited_bytes_transform_stream.ts new file mode 100644 index 0000000..6835efb --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/limited_bytes_transform_stream.ts @@ -0,0 +1,37 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +/** A TransformStream that will only read & enqueue `size` amount of bytes. + * This operation is chunk based and not BYOB based, + * and as such will read more than needed. + * + * if options.error is set, then instead of terminating the stream, + * an error will be thrown. + * + * ```ts + * import { LimitedBytesTransformStream } from "https://deno.land/std@$STD_VERSION/streams/limited_bytes_transform_stream.ts"; + * const res = await fetch("https://example.com"); + * const parts = res.body! + * .pipeThrough(new LimitedBytesTransformStream(512 * 1024)); + * ``` + */ +export class LimitedBytesTransformStream + extends TransformStream { + #read = 0; + constructor(size: number, options: { error?: boolean } = {}) { + super({ + transform: (chunk, controller) => { + if ((this.#read + chunk.byteLength) > size) { + if (options.error) { + throw new RangeError(`Exceeded byte size limit of '${size}'`); + } else { + controller.terminate(); + } + } else { + this.#read += chunk.byteLength; + controller.enqueue(chunk); + } + }, + }); + } +} diff --git a/vendor/deno.land/std@0.191.0/streams/limited_transform_stream.ts b/vendor/deno.land/std@0.191.0/streams/limited_transform_stream.ts new file mode 100644 index 0000000..a85303d --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/limited_transform_stream.ts @@ -0,0 +1,33 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +/** A TransformStream that will only read & enqueue `size` amount of chunks. + * + * if options.error is set, then instead of terminating the stream, + * an error will be thrown. + * + * ```ts + * import { LimitedTransformStream } from "https://deno.land/std@$STD_VERSION/streams/limited_transform_stream.ts"; + * const res = await fetch("https://example.com"); + * const parts = res.body!.pipeThrough(new LimitedTransformStream(50)); + * ``` + */ +export class LimitedTransformStream extends TransformStream { + #read = 0; + constructor(size: number, options: { error?: boolean } = {}) { + super({ + transform: (chunk, controller) => { + if ((this.#read + 1) > size) { + if (options.error) { + throw new RangeError(`Exceeded chunk limit of '${size}'`); + } else { + controller.terminate(); + } + } else { + this.#read++; + controller.enqueue(chunk); + } + }, + }); + } +} diff --git a/vendor/deno.land/std@0.191.0/streams/merge_readable_streams.ts b/vendor/deno.land/std@0.191.0/streams/merge_readable_streams.ts new file mode 100644 index 0000000..297f2c8 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/merge_readable_streams.ts @@ -0,0 +1,42 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. + +import { deferred } from "../async/deferred.ts"; + +/** + * Merge multiple streams into a single one, not taking order into account. + * If a stream ends before other ones, the other will continue adding data, + * and the finished one will not add any more data. + */ +export function mergeReadableStreams( + ...streams: ReadableStream[] +): ReadableStream { + const resolvePromises = streams.map(() => deferred()); + return new ReadableStream({ + start(controller) { + let mustClose = false; + Promise.all(resolvePromises) + .then(() => { + controller.close(); + }) + .catch((error) => { + mustClose = true; + controller.error(error); + }); + for (const [key, stream] of Object.entries(streams)) { + (async () => { + try { + for await (const data of stream) { + if (mustClose) { + break; + } + controller.enqueue(data); + } + resolvePromises[+key].resolve(); + } catch (error) { + resolvePromises[+key].reject(error); + } + })(); + } + }, + }); +} diff --git a/vendor/deno.land/std@0.191.0/streams/mod.ts b/vendor/deno.land/std@0.191.0/streams/mod.ts new file mode 100644 index 0000000..b5ba3f6 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/mod.ts @@ -0,0 +1,31 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +/** + * Utilities for working with the + * [Streams API](https://developer.mozilla.org/en-US/docs/Web/API/Streams_API). + * + * Includes buffering and conversion. + * + * @module + */ + +export * from "./buffer.ts"; +export * from "./byte_slice_stream.ts"; +export * from "./copy.ts"; +export * from "./delimiter_stream.ts"; +export * from "./early_zip_readable_streams.ts"; +export * from "./iterate_reader.ts"; +export * from "./limited_bytes_transform_stream.ts"; +export * from "./limited_transform_stream.ts"; +export * from "./merge_readable_streams.ts"; +export * from "./read_all.ts"; +export * from "./readable_stream_from_iterable.ts"; +export * from "./readable_stream_from_reader.ts"; +export * from "./reader_from_iterable.ts"; +export * from "./reader_from_stream_reader.ts"; +export * from "./text_delimiter_stream.ts"; +export * from "./text_line_stream.ts"; +export * from "./to_transform_stream.ts"; +export * from "./writable_stream_from_writer.ts"; +export * from "./write_all.ts"; +export * from "./writer_from_stream_writer.ts"; +export * from "./zip_readable_streams.ts"; diff --git a/vendor/deno.land/std@0.191.0/streams/readable_stream_from_iterable.ts b/vendor/deno.land/std@0.191.0/streams/readable_stream_from_iterable.ts new file mode 100644 index 0000000..80851e4 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/readable_stream_from_iterable.ts @@ -0,0 +1,63 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +/** Create a `ReadableStream` from any kind of iterable. + * + * ```ts + * import { readableStreamFromIterable } from "https://deno.land/std@$STD_VERSION/streams/readable_stream_from_iterable.ts"; + * + * const r1 = readableStreamFromIterable(["foo, bar, baz"]); + * const r2 = readableStreamFromIterable(async function* () { + * await new Promise(((r) => setTimeout(r, 1000))); + * yield "foo"; + * await new Promise(((r) => setTimeout(r, 1000))); + * yield "bar"; + * await new Promise(((r) => setTimeout(r, 1000))); + * yield "baz"; + * }()); + * ``` + * + * If the produced iterator (`iterable[Symbol.asyncIterator]()` or + * `iterable[Symbol.iterator]()`) is a generator, or more specifically is found + * to have a `.throw()` method on it, that will be called upon + * `readableStream.cancel()`. This is the case for the second input type above: + * + * ```ts + * import { readableStreamFromIterable } from "https://deno.land/std@$STD_VERSION/streams/readable_stream_from_iterable.ts"; + * + * const r3 = readableStreamFromIterable(async function* () { + * try { + * yield "foo"; + * } catch (error) { + * console.log(error); // Error: Cancelled by consumer. + * } + * }()); + * const reader = r3.getReader(); + * console.log(await reader.read()); // { value: "foo", done: false } + * await reader.cancel(new Error("Cancelled by consumer.")); + * ``` + */ +export function readableStreamFromIterable( + iterable: Iterable | AsyncIterable, +): ReadableStream { + const iterator: Iterator | AsyncIterator = + (iterable as AsyncIterable)[Symbol.asyncIterator]?.() ?? + (iterable as Iterable)[Symbol.iterator]?.(); + return new ReadableStream({ + async pull(controller) { + const { value, done } = await iterator.next(); + if (done) { + controller.close(); + } else { + controller.enqueue(value); + } + }, + async cancel(reason) { + if (typeof iterator.throw == "function") { + try { + await iterator.throw(reason); + } catch { /* `iterator.throw()` always throws on site. We catch it. */ } + } + }, + }); +} diff --git a/vendor/deno.land/std@0.191.0/streams/readable_stream_from_reader.ts b/vendor/deno.land/std@0.191.0/streams/readable_stream_from_reader.ts new file mode 100644 index 0000000..e987082 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/readable_stream_from_reader.ts @@ -0,0 +1,81 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import { DEFAULT_CHUNK_SIZE } from "./_common.ts"; +import type { Closer, Reader } from "../types.d.ts"; + +function isCloser(value: unknown): value is Closer { + return typeof value === "object" && value != null && "close" in value && + // deno-lint-ignore no-explicit-any + typeof (value as Record)["close"] === "function"; +} + +export interface ReadableStreamFromReaderOptions { + /** If the `reader` is also a `Closer`, automatically close the `reader` + * when `EOF` is encountered, or a read error occurs. + * + * @default {true} + */ + autoClose?: boolean; + + /** The size of chunks to allocate to read, the default is ~16KiB, which is + * the maximum size that Deno operations can currently support. */ + chunkSize?: number; + + /** The queuing strategy to create the `ReadableStream` with. */ + strategy?: { highWaterMark?: number | undefined; size?: undefined }; +} + +/** + * Create a `ReadableStream` from a `Reader`. + * + * When the pull algorithm is called on the stream, a chunk from the reader + * will be read. When `null` is returned from the reader, the stream will be + * closed along with the reader (if it is also a `Closer`). + * + * An example converting a `Deno.FsFile` into a readable stream: + * + * ```ts + * import { readableStreamFromReader } from "https://deno.land/std@$STD_VERSION/streams/readable_stream_from_reader.ts"; + * + * const file = await Deno.open("./file.txt", { read: true }); + * const fileStream = readableStreamFromReader(file); + * ``` + */ +export function readableStreamFromReader( + reader: Reader | (Reader & Closer), + options: ReadableStreamFromReaderOptions = {}, +): ReadableStream { + const { + autoClose = true, + chunkSize = DEFAULT_CHUNK_SIZE, + strategy, + } = options; + + return new ReadableStream({ + async pull(controller) { + const chunk = new Uint8Array(chunkSize); + try { + const read = await reader.read(chunk); + if (read === null) { + if (isCloser(reader) && autoClose) { + reader.close(); + } + controller.close(); + return; + } + controller.enqueue(chunk.subarray(0, read)); + } catch (e) { + controller.error(e); + if (isCloser(reader)) { + reader.close(); + } + } + }, + cancel() { + if (isCloser(reader) && autoClose) { + reader.close(); + } + }, + }, strategy); +} diff --git a/vendor/deno.land/std@0.191.0/streams/reader_from_iterable.ts b/vendor/deno.land/std@0.191.0/streams/reader_from_iterable.ts new file mode 100644 index 0000000..f63523e --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/reader_from_iterable.ts @@ -0,0 +1,56 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import { Buffer } from "../io/buffer.ts"; +import { writeAll } from "./write_all.ts"; +import { Reader } from "../types.d.ts"; + +/** Create a `Reader` from an iterable of `Uint8Array`s. + * + * ```ts + * import { readerFromIterable } from "https://deno.land/std@$STD_VERSION/streams/reader_from_iterable.ts"; + * import { copy } from "https://deno.land/std@$STD_VERSION/streams/copy.ts"; + * + * const file = await Deno.open("metrics.txt", { write: true }); + * const reader = readerFromIterable((async function* () { + * while (true) { + * await new Promise((r) => setTimeout(r, 1000)); + * const message = `data: ${JSON.stringify(Deno.metrics())}\n\n`; + * yield new TextEncoder().encode(message); + * } + * })()); + * await copy(reader, file); + * ``` + */ +export function readerFromIterable( + iterable: Iterable | AsyncIterable, +): Reader { + const iterator: Iterator | AsyncIterator = + (iterable as AsyncIterable)[Symbol.asyncIterator]?.() ?? + (iterable as Iterable)[Symbol.iterator]?.(); + const buffer = new Buffer(); + return { + async read(p: Uint8Array): Promise { + if (buffer.length == 0) { + const result = await iterator.next(); + if (result.done) { + return null; + } else { + if (result.value.byteLength <= p.byteLength) { + p.set(result.value); + return result.value.byteLength; + } + p.set(result.value.subarray(0, p.byteLength)); + await writeAll(buffer, result.value.subarray(p.byteLength)); + return p.byteLength; + } + } else { + const n = await buffer.read(p); + if (n == null) { + return this.read(p); + } + return n; + } + }, + }; +} diff --git a/vendor/deno.land/std@0.191.0/streams/reader_from_stream_reader.ts b/vendor/deno.land/std@0.191.0/streams/reader_from_stream_reader.ts new file mode 100644 index 0000000..e1d3113 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/reader_from_stream_reader.ts @@ -0,0 +1,43 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import { Buffer } from "../io/buffer.ts"; +import { writeAll } from "./write_all.ts"; +import type { Reader } from "../types.d.ts"; + +/** + * Create a `Reader` from a `ReadableStreamDefaultReader`. + * + * @example + * ```ts + * import { copy } from "https://deno.land/std@$STD_VERSION/streams/copy.ts"; + * import { readerFromStreamReader } from "https://deno.land/std@$STD_VERSION/streams/reader_from_stream_reader.ts"; + * + * const res = await fetch("https://deno.land"); + * const file = await Deno.open("./deno.land.html", { create: true, write: true }); + * + * const reader = readerFromStreamReader(res.body!.getReader()); + * await copy(reader, file); + * file.close(); + * ``` + */ +export function readerFromStreamReader( + streamReader: ReadableStreamDefaultReader, +): Reader { + const buffer = new Buffer(); + + return { + async read(p: Uint8Array): Promise { + if (buffer.empty()) { + const res = await streamReader.read(); + if (res.done) { + return null; // EOF + } + + await writeAll(buffer, res.value); + } + + return buffer.read(p); + }, + }; +} diff --git a/vendor/deno.land/std@0.191.0/streams/text_delimiter_stream.ts b/vendor/deno.land/std@0.191.0/streams/text_delimiter_stream.ts new file mode 100644 index 0000000..85d7725 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/text_delimiter_stream.ts @@ -0,0 +1,78 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import { createLPS } from "./_common.ts"; + +import type { + DelimiterDisposition, + DelimiterStreamOptions, +} from "./delimiter_stream.ts"; + +/** Transform a stream into a stream where each chunk is divided by a given delimiter. + * + * ```ts + * import { TextDelimiterStream } from "https://deno.land/std@$STD_VERSION/streams/text_delimiter_stream.ts"; + * const res = await fetch("https://example.com"); + * const parts = res.body! + * .pipeThrough(new TextDecoderStream()) + * .pipeThrough(new TextDelimiterStream("foo")); + * ``` + */ +export class TextDelimiterStream extends TransformStream { + #buf = ""; + #delimiter: string; + #inspectIndex = 0; + #matchIndex = 0; + #delimLPS: Uint8Array; + #disp: DelimiterDisposition; + + constructor(delimiter: string, options?: DelimiterStreamOptions) { + super({ + transform: (chunk, controller) => { + this.#handle(chunk, controller); + }, + flush: (controller) => { + controller.enqueue(this.#buf); + }, + }); + + this.#delimiter = delimiter; + this.#delimLPS = createLPS(new TextEncoder().encode(delimiter)); + this.#disp = options?.disposition ?? "discard"; + } + + #handle( + chunk: string, + controller: TransformStreamDefaultController, + ) { + this.#buf += chunk; + let localIndex = 0; + while (this.#inspectIndex < this.#buf.length) { + if (chunk[localIndex] === this.#delimiter[this.#matchIndex]) { + this.#inspectIndex++; + localIndex++; + this.#matchIndex++; + if (this.#matchIndex === this.#delimiter.length) { + // Full match + const start = this.#inspectIndex - this.#delimiter.length; + const end = this.#disp === "suffix" ? this.#inspectIndex : start; + const copy = this.#buf.slice(0, end); + controller.enqueue(copy); + const shift = this.#disp == "prefix" ? start : this.#inspectIndex; + this.#buf = this.#buf.slice(shift); + this.#inspectIndex = this.#disp == "prefix" + ? this.#delimiter.length + : 0; + this.#matchIndex = 0; + } + } else { + if (this.#matchIndex === 0) { + this.#inspectIndex++; + localIndex++; + } else { + this.#matchIndex = this.#delimLPS[this.#matchIndex - 1]; + } + } + } + } +} diff --git a/vendor/deno.land/std@0.191.0/streams/text_line_stream.ts b/vendor/deno.land/std@0.191.0/streams/text_line_stream.ts new file mode 100644 index 0000000..2d6c093 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/text_line_stream.ts @@ -0,0 +1,74 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +interface TextLineStreamOptions { + /** Allow splitting by solo \r */ + allowCR: boolean; +} + +/** Transform a stream into a stream where each chunk is divided by a newline, + * be it `\n` or `\r\n`. `\r` can be enabled via the `allowCR` option. + * + * ```ts + * import { TextLineStream } from "https://deno.land/std@$STD_VERSION/streams/text_line_stream.ts"; + * const res = await fetch("https://example.com"); + * const lines = res.body! + * .pipeThrough(new TextDecoderStream()) + * .pipeThrough(new TextLineStream()); + * ``` + */ +export class TextLineStream extends TransformStream { + readonly #allowCR: boolean; + #buf = ""; + + constructor(options?: TextLineStreamOptions) { + super({ + transform: (chunk, controller) => this.#handle(chunk, controller), + flush: (controller) => { + if (this.#buf.length > 0) { + if ( + this.#allowCR && + this.#buf[this.#buf.length - 1] === "\r" + ) controller.enqueue(this.#buf.slice(0, -1)); + else controller.enqueue(this.#buf); + } + }, + }); + this.#allowCR = options?.allowCR ?? false; + } + + #handle(chunk: string, controller: TransformStreamDefaultController) { + chunk = this.#buf + chunk; + + for (;;) { + const lfIndex = chunk.indexOf("\n"); + + if (this.#allowCR) { + const crIndex = chunk.indexOf("\r"); + + if ( + crIndex !== -1 && crIndex !== (chunk.length - 1) && + (lfIndex === -1 || (lfIndex - 1) > crIndex) + ) { + controller.enqueue(chunk.slice(0, crIndex)); + chunk = chunk.slice(crIndex + 1); + continue; + } + } + + if (lfIndex !== -1) { + let crOrLfIndex = lfIndex; + if (chunk[lfIndex - 1] === "\r") { + crOrLfIndex--; + } + controller.enqueue(chunk.slice(0, crOrLfIndex)); + chunk = chunk.slice(lfIndex + 1); + continue; + } + + break; + } + + this.#buf = chunk; + } +} diff --git a/vendor/deno.land/std@0.191.0/streams/to_transform_stream.ts b/vendor/deno.land/std@0.191.0/streams/to_transform_stream.ts new file mode 100644 index 0000000..5003c72 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/to_transform_stream.ts @@ -0,0 +1,75 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +/** + * Convert the generator function into a TransformStream. + * + * ```ts + * import { readableStreamFromIterable } from "https://deno.land/std@$STD_VERSION/streams/readable_stream_from_iterable.ts"; + * import { toTransformStream } from "https://deno.land/std@$STD_VERSION/streams/to_transform_stream.ts"; + * + * const readable = readableStreamFromIterable([0, 1, 2]) + * .pipeThrough(toTransformStream(async function* (src) { + * for await (const chunk of src) { + * yield chunk * 100; + * } + * })); + * + * for await (const chunk of readable) { + * console.log(chunk); + * } + * // output: 0, 100, 200 + * ``` + * + * @param transformer A function to transform. + * @param writableStrategy An object that optionally defines a queuing strategy for the stream. + * @param readableStrategy An object that optionally defines a queuing strategy for the stream. + */ +export function toTransformStream( + transformer: (src: ReadableStream) => Iterable | AsyncIterable, + writableStrategy?: QueuingStrategy, + readableStrategy?: QueuingStrategy, +): TransformStream { + const { + writable, + readable, + } = new TransformStream(undefined, writableStrategy); + + const iterable = transformer(readable); + const iterator: Iterator | AsyncIterator = + (iterable as AsyncIterable)[Symbol.asyncIterator]?.() ?? + (iterable as Iterable)[Symbol.iterator]?.(); + return { + writable, + readable: new ReadableStream({ + async pull(controller) { + let result: IteratorResult; + try { + result = await iterator.next(); + } catch (error) { + // Propagate error to stream from iterator + // If the stream status is "errored", it will be thrown, but ignore. + await readable.cancel(error).catch(() => {}); + controller.error(error); + return; + } + if (result.done) { + controller.close(); + return; + } + controller.enqueue(result.value); + }, + async cancel(reason) { + // Propagate cancellation to readable and iterator + if (typeof iterator.throw == "function") { + try { + await iterator.throw(reason); + } catch { + /* `iterator.throw()` always throws on site. We catch it. */ + } + } + await readable.cancel(reason); + }, + }, readableStrategy), + }; +} diff --git a/vendor/deno.land/std@0.191.0/streams/writable_stream_from_writer.ts b/vendor/deno.land/std@0.191.0/streams/writable_stream_from_writer.ts new file mode 100644 index 0000000..ffdd90e --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/writable_stream_from_writer.ts @@ -0,0 +1,52 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import { writeAll } from "./write_all.ts"; +import type { Closer, Writer } from "../types.d.ts"; + +function isCloser(value: unknown): value is Closer { + return typeof value === "object" && value != null && "close" in value && + // deno-lint-ignore no-explicit-any + typeof (value as Record)["close"] === "function"; +} + +export interface WritableStreamFromWriterOptions { + /** + * If the `writer` is also a `Closer`, automatically close the `writer` + * when the stream is closed, aborted, or a write error occurs. + * + * @default {true} + */ + autoClose?: boolean; +} + +/** Create a `WritableStream` from a `Writer`. */ +export function writableStreamFromWriter( + writer: Writer, + options: WritableStreamFromWriterOptions = {}, +): WritableStream { + const { autoClose = true } = options; + + return new WritableStream({ + async write(chunk, controller) { + try { + await writeAll(writer, chunk); + } catch (e) { + controller.error(e); + if (isCloser(writer) && autoClose) { + writer.close(); + } + } + }, + close() { + if (isCloser(writer) && autoClose) { + writer.close(); + } + }, + abort() { + if (isCloser(writer) && autoClose) { + writer.close(); + } + }, + }); +} diff --git a/vendor/deno.land/std@0.191.0/streams/write_all.ts b/vendor/deno.land/std@0.191.0/streams/write_all.ts new file mode 100644 index 0000000..c23a837 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/write_all.ts @@ -0,0 +1,65 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import type { Writer, WriterSync } from "../types.d.ts"; + +/** Write all the content of the array buffer (`arr`) to the writer (`w`). + * + * ```ts + * import { Buffer } from "https://deno.land/std@$STD_VERSION/io/buffer.ts"; + * import { writeAll } from "https://deno.land/std@$STD_VERSION/streams/write_all.ts"; + + * // Example writing to stdout + * let contentBytes = new TextEncoder().encode("Hello World"); + * await writeAll(Deno.stdout, contentBytes); + * + * // Example writing to file + * contentBytes = new TextEncoder().encode("Hello World"); + * const file = await Deno.open('test.file', {write: true}); + * await writeAll(file, contentBytes); + * file.close(); + * + * // Example writing to buffer + * contentBytes = new TextEncoder().encode("Hello World"); + * const writer = new Buffer(); + * await writeAll(writer, contentBytes); + * console.log(writer.bytes().length); // 11 + * ``` + */ +export async function writeAll(w: Writer, arr: Uint8Array) { + let nwritten = 0; + while (nwritten < arr.length) { + nwritten += await w.write(arr.subarray(nwritten)); + } +} + +/** Synchronously write all the content of the array buffer (`arr`) to the + * writer (`w`). + * + * ```ts + * import { Buffer } from "https://deno.land/std@$STD_VERSION/io/buffer.ts"; + * import { writeAllSync } from "https://deno.land/std@$STD_VERSION/streams/write_all.ts"; + * + * // Example writing to stdout + * let contentBytes = new TextEncoder().encode("Hello World"); + * writeAllSync(Deno.stdout, contentBytes); + * + * // Example writing to file + * contentBytes = new TextEncoder().encode("Hello World"); + * const file = Deno.openSync('test.file', {write: true}); + * writeAllSync(file, contentBytes); + * file.close(); + * + * // Example writing to buffer + * contentBytes = new TextEncoder().encode("Hello World"); + * const writer = new Buffer(); + * writeAllSync(writer, contentBytes); + * console.log(writer.bytes().length); // 11 + * ``` + */ +export function writeAllSync(w: WriterSync, arr: Uint8Array) { + let nwritten = 0; + while (nwritten < arr.length) { + nwritten += w.writeSync(arr.subarray(nwritten)); + } +} diff --git a/vendor/deno.land/std@0.191.0/streams/writer_from_stream_writer.ts b/vendor/deno.land/std@0.191.0/streams/writer_from_stream_writer.ts new file mode 100644 index 0000000..bbedf63 --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/writer_from_stream_writer.ts @@ -0,0 +1,35 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +import type { Writer } from "../types.d.ts"; + +/** Create a `Writer` from a `WritableStreamDefaultWriter`. + * + * @example + * ```ts + * import { copy } from "https://deno.land/std@$STD_VERSION/streams/copy.ts"; + * import { writerFromStreamWriter } from "https://deno.land/std@$STD_VERSION/streams/writer_from_stream_writer.ts"; + * + * const file = await Deno.open("./deno.land.html", { read: true }); + * + * const writableStream = new WritableStream({ + * write(chunk): void { + * console.log(chunk); + * }, + * }); + * const writer = writerFromStreamWriter(writableStream.getWriter()); + * await copy(file, writer); + * file.close(); + * ``` + */ +export function writerFromStreamWriter( + streamWriter: WritableStreamDefaultWriter, +): Writer { + return { + async write(p: Uint8Array): Promise { + await streamWriter.ready; + await streamWriter.write(p); + return p.length; + }, + }; +} diff --git a/vendor/deno.land/std@0.191.0/streams/zip_readable_streams.ts b/vendor/deno.land/std@0.191.0/streams/zip_readable_streams.ts new file mode 100644 index 0000000..90ee9db --- /dev/null +++ b/vendor/deno.land/std@0.191.0/streams/zip_readable_streams.ts @@ -0,0 +1,35 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. +// This module is browser compatible. + +/** + * Merge multiple streams into a single one, taking order into account, and each stream + * will wait for a chunk to enqueue before the next stream can append another chunk. + * If a stream ends before other ones, the others will continue adding data in order, + * and the finished one will not add any more data. + */ +export function zipReadableStreams( + ...streams: ReadableStream[] +): ReadableStream { + const readers = streams.map((s) => s.getReader()); + return new ReadableStream({ + async start(controller) { + try { + let resolved = 0; + while (resolved != streams.length) { + for (const [key, reader] of Object.entries(readers)) { + const { value, done } = await reader.read(); + if (!done) { + controller.enqueue(value!); + } else { + resolved++; + readers.splice(+key, 1); + } + } + } + controller.close(); + } catch (e) { + controller.error(e); + } + }, + }); +} diff --git a/vendor/import_map.json b/vendor/import_map.json index 5d2a804..52786a6 100644 --- a/vendor/import_map.json +++ b/vendor/import_map.json @@ -1,9 +1,7 @@ { "imports": { - "https://denopkg.com/chiefbiiko/sha1/mod.ts": "./denopkg.com/chiefbiiko/sha1@master/mod.ts", "https://deno.land/x/oak/mod.ts": "./deno.land/x/oak@v12.1.0/mod.ts", "https://deno.land/std/testing/asserts.ts": "./deno.land/std@0.191.0/testing/asserts.ts", - "https://deno.land/": "./deno.land/", - "https://denopkg.com/": "./denopkg.com/" + "https://deno.land/": "./deno.land/" } }