Removed AsyncIterableReader which can be replaced by readerFromIterable available in the std

This commit is contained in:
didier amyot committed 2023-06-13 07:43:07 -04:00
1 parent 789fb49e8b
commit f7a8d494f5
3 files changed
-111

No files matched your search

-66
View File
@@ -5,69 +5,3 @@ export async function asyncIterableToArray<T>(ite: AsyncIterable<T>): Promise<T[
}
return ret;
}
export function concatUint8Array(arr1: Uint8Array, arr2: Uint8Array): Uint8Array {
const mergedArray = new Uint8Array(arr1.length + arr2.length);
mergedArray.set(arr1);
mergedArray.set(arr2, arr1.length);
return mergedArray;
}
export class AsyncIterableReader implements Deno.Reader {
private readonly source: AsyncIterator<Uint8Array>;
private interReadBuffer = new Uint8Array();
private isSourceEmpty = false;
constructor(source: AsyncIterable<Uint8Array>) {
this.source = source[Symbol.asyncIterator]();
}
private async readMinimumOf(n: number): Promise<Uint8Array> {
let ret = new Uint8Array();
if(this.isSourceEmpty){
return ret;
}
let ite = await this.source.next();
while(!ite.done){
const piece = ite.value;
ret = concatUint8Array(ret, piece);
if(ret.length >= n){
return ret;
}
ite = await this.source.next();
}
this.isSourceEmpty = true;
return ret;
}
private transferData(source: Uint8Array, dest: Uint8Array): number {
const limit = Math.min(source.length, dest.length);
for(let i = 0; i < limit; i ++){
dest[i] = source[i];
}
return limit;
}
private readFromInternalReadBuffer(buffer: Uint8Array): number {
const n = this.transferData(this.interReadBuffer, buffer);
this.interReadBuffer = this.interReadBuffer.subarray(n);
return n;
}
async read(buffer: Uint8Array): Promise<number | null> {
if(this.isSourceEmpty && this.interReadBuffer.length < 1){
return null;
}
const internalReadBufferTransferdSize = this.readFromInternalReadBuffer(buffer);
if(internalReadBufferTransferdSize == buffer.length){
return internalReadBufferTransferdSize;
}
const data = await this.readMinimumOf(buffer.length - internalReadBufferTransferdSize);
const transferedSize = this.transferData(data, buffer.subarray(internalReadBufferTransferdSize));
this.interReadBuffer = data.subarray(transferedSize);
return internalReadBufferTransferdSize + transferedSize;
}
}
-1
View File
@@ -1,5 +1,4 @@
export function* randomNumbers(seed: number, max: number, maxResult: number|undefined = undefined) : Generator<number> {
const a = 48271;
const m = 2147483647;
-44
View File
@@ -1,44 +0,0 @@
import { readAll } from "https://deno.land/std@0.191.0/streams/read_all.ts";
import { assertEquals } from "https://deno.land/std/testing/asserts.ts";
import { AsyncIterableReader } from "../src/utils.ts";
import { randomNumbers, range } from "./utils.ts";
Deno.test("AsyncIterableReader", async (t) => {
async function* miniSource(){
yield Promise.resolve(new Uint8Array([1, 2, 3]));
yield Promise.resolve(new Uint8Array([4, 5, 6, 7]));
yield Promise.resolve(new Uint8Array([8, 9]));
}
async function* largeSource(size: number, seed = 77): AsyncIterable<Uint8Array>{
let val = 0
const random = randomNumbers(seed, size > 10 ? size / 4: size);
while(val < size){
const randomValue = random.next().value;
const newVal = Math.min(val + randomValue, size);
const ret = new Uint8Array(range(val, newVal));
yield Promise.resolve(ret);
val = newVal;
}
}
await t.step('read all mini source', async () => {
const reader = new AsyncIterableReader(miniSource());
const expt = new Uint8Array([1, 2 ,3 ,4 ,5, 6,7,8,9])
assertEquals(await readAll(reader), expt);
});
await t.step('read all large source', async () => {
// let's run the test 1000 times to make sure implementation is ok.
for(const i of randomNumbers(666, 100 * 1024, 1000)){
const reader = new AsyncIterableReader(largeSource(i));
const ret = await readAll(reader);
assertEquals(ret.length, i);
}
});
});