First workable server.

This commit is contained in:
didier amyot committed 2023-06-13 07:37:41 -04:00
1 parent 0d627c9051
commit 789fb49e8b
27 files changed
+1355 -68

No files matched your search

+2
View File
@@ -1,3 +1,5 @@
.vscode .vscode
.direnv .direnv
result result
foo
playground.ts
+66 -42
View File
@@ -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 { 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 { export interface FileService {
@@ -29,13 +36,14 @@ export interface FileService {
newPath: string, newPath: string,
): Promise<FileMetaData | null>; ): Promise<FileMetaData | null>;
deleteByPath(path: string): Promise<void>; deleteByPath(path: string): Promise<boolean>;
} }
interface SerializedFile { interface SerializedFile {
id: string; id: string;
path: string; path: string;
bodySize: number; bodySize: number;
compression: string;
} }
export interface FileMetaData { export interface FileMetaData {
@@ -57,6 +65,14 @@ export class DenoKVBlobService {
this.blobSize = blobSize; this.blobSize = blobSize;
} }
private async sha1(msgUint8: Uint8Array): Promise<string> {
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<Uint8Array> { private *splitText(body: Uint8Array): Generator<Uint8Array> {
let blobCount = 0; let blobCount = 0;
let txt = body.slice( let txt = body.slice(
@@ -77,10 +93,10 @@ export class DenoKVBlobService {
if(body.length > this.blobSize){ if(body.length > this.blobSize){
throw new Error('Persisted a too large blob'); 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_index', mainIndex, index], blob_id);
await this.kv.set(['blob', blob_id], body); await this.kv.set(['blob', blob_id], body);
} }
async persistBlob(index: string, body: Uint8Array): Promise<number> { async persistBlob(index: string, body: Uint8Array): Promise<number> {
@@ -98,7 +114,7 @@ export class DenoKVBlobService {
let i = 0; let i = 0;
let size = 0; let size = 0;
for await (const chunk of iterateReader(body, {bufSize: this.blobSize})){ for await (const chunk of iterateReader(body, {bufSize: this.blobSize})){
this.unsafePersistBlob(index, i, chunk); await this.unsafePersistBlob(index, i, chunk);
size += chunk.length; size += chunk.length;
i++; i++;
} }
@@ -162,36 +178,29 @@ export class DenoKVFileService implements FileService {
constructor(kv: Deno.Kv) { constructor(kv: Deno.Kv) {
this.kv = kv; this.kv = kv;
this.blobService = new DenoKVBlobService(kv, 500); this.blobService = new DenoKVBlobService(kv, 32 * 1024);
} }
async createFromFullBody( createFromFullBody(
path: string, path: string,
body: string | Uint8Array, body: string | Uint8Array,
): Promise<FileMetaData> { ): Promise<FileMetaData> {
const newId = crypto.randomUUID(); if (body instanceof Uint8Array){
const bodyBytes: Uint8Array = body instanceof Uint8Array ? body : new TextEncoder().encode(body); return this.createFromReader(path, new Buffer(body));
const size = await this.blobService.persistBlob(newId, bodyBytes); }
return this.createFromReader(path, new StringReader(body));
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 };
} }
async createFromReader(path: string, body: Deno.Reader): Promise<FileMetaData>{ async createFromReader(path: string, content: Deno.Reader): Promise<FileMetaData>{
const newId = crypto.randomUUID(); 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 = { const serializedFile: SerializedFile = {
id: newId, id: newId,
path, path,
bodySize: size, bodySize: size,
compression: 'gzip'
}; };
const idx: SerializedFilePathIndex = { fileId: newId }; const idx: SerializedFilePathIndex = { fileId: newId };
await this.kv.set(['file', newId], serializedFile); await this.kv.set(['file', newId], serializedFile);
@@ -221,13 +230,13 @@ export class DenoKVFileService implements FileService {
} }
} }
async deleteByPath(path: string): Promise<void> { async deleteByPath(path: string): Promise<boolean> {
const idx = await this.kv.get<SerializedFilePathIndex>([ const idx = await this.kv.get<SerializedFilePathIndex>([
'file_index_path', 'file_index_path',
path, path,
]); ]);
if (idx.value === null) { if (idx.value === null) {
return; return false;
} }
// todo, this will fail silently if transaction fails. we might want to put that into a loop. // todo, this will fail silently if transaction fails. we might want to put that into a loop.
await this.kv.atomic() await this.kv.atomic()
@@ -235,6 +244,22 @@ export class DenoKVFileService implements FileService {
.delete(['file_index_path', path]) .delete(['file_index_path', path])
.delete(idx.key) .delete(idx.key)
.commit(); .commit();
return true;
}
private async getSerializedFile(fileId: string): Promise<SerializedFile | null>{
const ret = await this.kv.get<SerializedFile>(['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<string | null> { async getFullBodyByPath(path: string): Promise<string | null> {
@@ -243,14 +268,8 @@ export class DenoKVFileService implements FileService {
} }
async getRawBodyByPath(path: string): Promise<Uint8Array | null> { async getRawBodyByPath(path: string): Promise<Uint8Array | null> {
const idx = await this.kv.get<SerializedFilePathIndex>([ const ret = await this.getBodyAsReader(path);
'file_index_path', return ret === null ? null : readAll(ret);
path,
]);
if (idx.value === null) {
return null;
}
return this.blobService.readAllBlob(idx.value.fileId);
} }
@@ -262,19 +281,23 @@ export class DenoKVFileService implements FileService {
if (idx.value === null) { if (idx.value === null) {
return null; return null;
} }
return this.blobService.getBlob(idx.value.fileId); const ser = await this.kv.get<SerializedFile>(['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<FileMetaData | null> { async getMetaDataByID(id: string): Promise<FileMetaData | null> {
const fileObject = await this.kv.get<SerializedFile>(['file', id]); const fileObject = await this.getSerializedFile(id);
if (fileObject.value === null) { if (fileObject === null) {
return null; return null;
} }
return { return {
path: fileObject.value.path, path: fileObject.path,
id: fileObject.value.id, id: fileObject.id,
bodySize: fileObject.value.bodySize, bodySize: fileObject.bodySize,
}; };
} }
@@ -287,10 +310,11 @@ export class DenoKVFileService implements FileService {
'file_index_path', 'file_index_path',
fileObject.value.path, fileObject.value.path,
]); ]);
const newFile: FileMetaData = { const newFile: SerializedFile = {
path: newPath, path: newPath,
id: fileObject.value.id, id: fileObject.value.id,
bodySize: fileObject.value.bodySize, bodySize: fileObject.value.bodySize,
compression: fileObject.value.compression,
}; };
await this.kv.atomic() await this.kv.atomic()
.check(fileObject) .check(fileObject)
+19 -13
View File
@@ -1,10 +1,7 @@
import { Context, Router } from "https://deno.land/x/oak/mod.ts"; import { Context, Router } from "https://deno.land/x/oak/mod.ts";
import { FileService } from "./fileservice.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 { 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; type Handler = (ctx: Context) => void;
@@ -17,7 +14,6 @@ function listFileHandlerFactory(fileService: FileService): Handler {
}; };
} }
function createFileHandlerFactory(fileService: FileService): Handler { function createFileHandlerFactory(fileService: FileService): Handler {
return async (ctx: Context) => { return async (ctx: Context) => {
const untypedCtx: any = ctx; const untypedCtx: any = ctx;
@@ -26,33 +22,43 @@ function createFileHandlerFactory(fileService: FileService): Handler {
const reqBody = await ctx.request.body({ type: "stream" }).value; const reqBody = await ctx.request.body({ type: "stream" }).value;
const reader = readerFromStreamReader(reqBody.getReader()) const reader = readerFromStreamReader(reqBody.getReader())
fileService.createFromReader(path, reader); await fileService.createFromReader(path, reader);
}; };
} }
function getFileHandlerFactory(fileService: FileService): Handler { function getFileHandlerFactory(fileService: FileService): Handler {
return async (ctx: Context) => { return async (ctx: Context) => {
const untypedCtx: any = ctx; const untypedCtx: any = ctx;
const fileId = untypedCtx.params.file_id; const fileId = untypedCtx.params.path;
const reader = await fileService.getBodyAsReader(fileId); const reader = await fileService.getBodyAsReader(fileId);
if(reader == null){ if(reader == null){
ctx.response.status = 404; ctx.response.status = 404;
return; return;
} }
ctx.response.status = 200; ctx.response.status = 200;
ctx.response.body = reader; ctx.response.body = reader;
ctx.response.type = "text/plain"; };
// ctx.send(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 { export function buildAPI(router: Router, fileService: FileService): Router {
router.get('/files', listFileHandlerFactory(fileService)); 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)); router.post('/files', createFileHandlerFactory(fileService));
return router; return router;
-3
View File
@@ -18,7 +18,6 @@ export class AsyncIterableReader implements Deno.Reader {
private readonly source: AsyncIterator<Uint8Array>; private readonly source: AsyncIterator<Uint8Array>;
private interReadBuffer = new Uint8Array(); private interReadBuffer = new Uint8Array();
private isSourceEmpty = false; private isSourceEmpty = false;
private shit = 0;
constructor(source: AsyncIterable<Uint8Array>) { constructor(source: AsyncIterable<Uint8Array>) {
this.source = source[Symbol.asyncIterator](); this.source = source[Symbol.asyncIterator]();
} }
@@ -58,8 +57,6 @@ export class AsyncIterableReader implements Deno.Reader {
} }
async read(buffer: Uint8Array): Promise<number | null> { async read(buffer: Uint8Array): Promise<number | null> {
this.shit += 1;
console.log("reading \> " + this.shit);
if(this.isSourceEmpty && this.interReadBuffer.length < 1){ if(this.isSourceEmpty && this.interReadBuffer.length < 1){
return null; return null;
} }
+9 -7
View File
@@ -8,6 +8,7 @@ import {
} from "../src/fileservice.ts"; } from "../src/fileservice.ts";
import { asyncIterableToArray } from "../src/utils.ts"; import { asyncIterableToArray } from "../src/utils.ts";
import { generateLargeText } from "./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(); kv.close();
}); });
Deno.test("DenoKVFileService happy path", async (t) => { Deno.test("DenoKVFileService happy path xxx ", async (t) => {
const kv = await Deno.openKv(":memory:"); const kv = await Deno.openKv(":memory:");
const fs: FileService = new DenoKVFileService(kv); const fs: FileService = new DenoKVFileService(kv);
const fileName1 = "didier/file1.txt"; const fileName1 = "didier/file1.txt";
@@ -64,7 +65,7 @@ Deno.test("DenoKVFileService happy path", async (t) => {
file2 = await fs.createFromFullBody(fileName2, "le geant vert ☀️"); file2 = await fs.createFromFullBody(fileName2, "le geant vert ☀️");
assertEquals(file1.path, fileName1); assertEquals(file1.path, fileName1);
assertEquals(file2.bodySize, 20); assertEquals(file2.bodySize, 47);
}); });
await t.step("Get file metadata", async () => { 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 () => { await t.step("Create File from reader", async () => {
const newPath = "boris/streamed-file.txt"; const newPath = "boris/streamed-file.txt";
const file = await Deno.open('tests/examples/teapot.obj');
const ret = await fs.createFromReader(newPath, file); const ret = await fs.createFromReader(newPath, new StringReader("hello world 18"));
assertEquals(ret.bodySize, 51493); assertEquals(ret.bodySize, 40);
file.close();
const s = await fs.getRawBodyByPath(newPath);
assertEquals(new TextDecoder().decode(s!), "hello world 18")
}); });
await t.step("Rename file", async () => { 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); const file = await fs.createFromFullBody("hello", inputBody);
assertEquals(file.path, "hello"); assertEquals(file.path, "hello");
assertEquals(file.bodySize, inputBody.length); assertEquals(file.bodySize, 27);
const bodyString = await fs.getFullBodyByPath("hello"); const bodyString = await fs.getFullBodyByPath("hello");
assertEquals(bodyString, new TextDecoder().decode(inputBody)); assertEquals(bodyString, new TextDecoder().decode(inputBody));
+48
View File
@@ -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<T> extends Promise<T> {
readonly state: "pending" | "fulfilled" | "rejected";
resolve(value?: T | PromiseLike<T>): 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<number>();
* // ...
* p.resolve(42);
* ```
*/
export function deferred<T>(): Deferred<T> {
let methods;
let state = "pending";
const promise = new Promise<T>((resolve, reject) => {
methods = {
async resolve(value: T | PromiseLike<T>) {
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<T>;
}
+169
View File
@@ -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<Uint8Array> = 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<Uint8Array>({
write: (chunk) => {
const m = this.#grow(chunk.byteLength);
copy(chunk, this.#buf, m);
},
});
get writable() {
return this.#writable;
}
constructor(ab?: ArrayBufferLike | ArrayLike<number>) {
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);
}
}
@@ -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<Uint8Array, Uint8Array> {
#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);
}
}
},
});
}
}
+48
View File
@@ -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<number> {
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;
}
+114
View File
@@ -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<Uint8Array, Uint8Array> {
#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<Uint8Array>,
) {
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];
}
}
}
}
}
@@ -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<T>(
...streams: ReadableStream<T>[]
): ReadableStream<T> {
const readers = streams.map((s) => s.getReader());
return new ReadableStream<T>({
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);
}
},
});
}
@@ -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<Uint8Array, Uint8Array> {
#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);
}
},
});
}
}
@@ -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<T> extends TransformStream<T, T> {
#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);
}
},
});
}
}
@@ -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<T>(
...streams: ReadableStream<T>[]
): ReadableStream<T> {
const resolvePromises = streams.map(() => deferred<void>());
return new ReadableStream<T>({
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);
}
})();
}
},
});
}
+31
View File
@@ -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";
@@ -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<T>(
iterable: Iterable<T> | AsyncIterable<T>,
): ReadableStream<T> {
const iterator: Iterator<T> | AsyncIterator<T> =
(iterable as AsyncIterable<T>)[Symbol.asyncIterator]?.() ??
(iterable as Iterable<T>)[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. */ }
}
},
});
}
@@ -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<string, any>)["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<Uint8Array>` 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<Uint8Array> {
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);
}
@@ -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<Uint8Array> | AsyncIterable<Uint8Array>,
): Reader {
const iterator: Iterator<Uint8Array> | AsyncIterator<Uint8Array> =
(iterable as AsyncIterable<Uint8Array>)[Symbol.asyncIterator]?.() ??
(iterable as Iterable<Uint8Array>)[Symbol.iterator]?.();
const buffer = new Buffer();
return {
async read(p: Uint8Array): Promise<number | null> {
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;
}
},
};
}
@@ -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<Uint8Array>,
): Reader {
const buffer = new Buffer();
return {
async read(p: Uint8Array): Promise<number | null> {
if (buffer.empty()) {
const res = await streamReader.read();
if (res.done) {
return null; // EOF
}
await writeAll(buffer, res.value);
}
return buffer.read(p);
},
};
}
@@ -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<string, string> {
#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<string>,
) {
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];
}
}
}
}
}
@@ -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<string, string> {
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<string>) {
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;
}
}
@@ -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<I, O>(
transformer: (src: ReadableStream<I>) => Iterable<O> | AsyncIterable<O>,
writableStrategy?: QueuingStrategy<I>,
readableStrategy?: QueuingStrategy<O>,
): TransformStream<I, O> {
const {
writable,
readable,
} = new TransformStream<I, I>(undefined, writableStrategy);
const iterable = transformer(readable);
const iterator: Iterator<O> | AsyncIterator<O> =
(iterable as AsyncIterable<O>)[Symbol.asyncIterator]?.() ??
(iterable as Iterable<O>)[Symbol.iterator]?.();
return {
writable,
readable: new ReadableStream<O>({
async pull(controller) {
let result: IteratorResult<O>;
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),
};
}
@@ -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<string, any>)["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<Uint8Array> {
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();
}
},
});
}
+65
View File
@@ -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));
}
}
@@ -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<Uint8Array>,
): Writer {
return {
async write(p: Uint8Array): Promise<number> {
await streamWriter.ready;
await streamWriter.write(p);
return p.length;
},
};
}
@@ -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<T>(
...streams: ReadableStream<T>[]
): ReadableStream<T> {
const readers = streams.map((s) => s.getReader());
return new ReadableStream<T>({
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);
}
},
});
}
+1 -3
View File
@@ -1,9 +1,7 @@
{ {
"imports": { "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/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/std/testing/asserts.ts": "./deno.land/std@0.191.0/testing/asserts.ts",
"https://deno.land/": "./deno.land/", "https://deno.land/": "./deno.land/"
"https://denopkg.com/": "./denopkg.com/"
} }
} }