forked from JSKitty/scc-web3
-
Notifications
You must be signed in to change notification settings - Fork 38
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add Reader class and refactor shield sync (#471)
* Add Reader class and refactor shield sync * Comment --------- Co-authored-by: Alessandro Rezzi <[email protected]>
- Loading branch information
Showing
3 changed files
with
272 additions
and
62 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,98 @@ | ||
| export class Reader { | ||
| #i = 0; | ||
| #maxBytes = 0; | ||
| #availableBytes; | ||
| #done = false; | ||
| /** | ||
| * @type{()=>{} | null} Called when bytes are available. | ||
| * There can't be more than 1 awaiter | ||
| */ | ||
| #awaiter = null; | ||
|
|
||
| /** | ||
| * @returns {number} Content length if available, or an estimante | ||
| */ | ||
| get contentLength() { | ||
| return this.#availableBytes.length; | ||
| } | ||
|
|
||
| /** | ||
| * @returns {number} Number or bytes read | ||
| */ | ||
| get readBytes() { | ||
| return this.#i; | ||
| } | ||
| /** | ||
| * @param | ||
| */ | ||
| constructor(req) { | ||
| this.#availableBytes = new Uint8Array( | ||
| req.headers?.get('Content-Length') || 1024 | ||
| ); | ||
| const stream = req.body.getReader(); | ||
| (async () => { | ||
| while (true) { | ||
| const { done, value } = await stream.read(); | ||
| if (value) { | ||
| this.#appendBytes(value); | ||
| } | ||
| if (done) { | ||
| this.#done = true; | ||
| break; | ||
| } | ||
| } | ||
| })(); | ||
| } | ||
|
|
||
| #resizeArray(newLength) { | ||
| if (newLength <= this.#availableBytes.length) { | ||
| throw new Error( | ||
| 'New length must be greater than the current length.' | ||
| ); | ||
| } | ||
|
|
||
| const newArray = new Uint8Array(newLength); | ||
| newArray.set(this.#availableBytes); | ||
| this.#availableBytes = newArray; | ||
| } | ||
|
|
||
| #appendBytes(bytes) { | ||
| // If we have content-length, there should never be a need to | ||
| // resize | ||
| if (bytes.length + this.#maxBytes > this.#availableBytes.length) { | ||
| this.#resizeArray((bytes.length + this.#maxBytes) * 2); | ||
| } | ||
|
|
||
| this.#availableBytes.set(bytes, this.#maxBytes); | ||
| this.#maxBytes += bytes.length; | ||
| // Notify the awaiter if there is one | ||
| if (this.#awaiter) this.#awaiter(); | ||
| } | ||
|
|
||
| /** | ||
| * @param{number} byteLength | ||
| * @returns {Promise<Uint8Array | null>} bytes or null if there are no more bytes | ||
| */ | ||
| async read(byteLength) { | ||
| if (this.#awaiter) throw new Error('Called read more than once'); | ||
| while (true) { | ||
| if (this.#maxBytes - this.#i >= byteLength) { | ||
| this.#awaiter = null; | ||
| // We have enough bytes to respond | ||
| const res = this.#availableBytes.subarray( | ||
| this.#i, | ||
| this.#i + byteLength | ||
| ); | ||
| this.#i += byteLength; | ||
| return res; | ||
| } | ||
|
|
||
| // There are no more bytes to await, so we can return null | ||
| if (this.#done) return null; | ||
| // If we didn't respond, wait for the next batch of bytes, then try again | ||
| await new Promise((res) => { | ||
| this.#awaiter = res; | ||
| }); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| import { describe, it, expect } from 'vitest'; | ||
| import { Reader } from '../../scripts/reader.js'; | ||
|
|
||
| function createMockStream(chunks, contentLength = null) { | ||
| let i = 0; | ||
| return { | ||
| headers: { | ||
| get: (key) => (key === 'Content-Length' ? contentLength : null), | ||
| }, | ||
| body: { | ||
| getReader: () => { | ||
| return { | ||
| read: async () => { | ||
| if (i < chunks.length) { | ||
| const value = chunks[i]; | ||
| i++; | ||
| return { done: false, value }; | ||
| } | ||
| return { done: true, value: null }; | ||
| }, | ||
| }; | ||
| }, | ||
| }, | ||
| }; | ||
| } | ||
|
|
||
| describe('Reader without content length', () => { | ||
| it('should read bytes correctly when available', async () => { | ||
| const mockStream = createMockStream([ | ||
| new Uint8Array([1, 2, 3, 4]), | ||
| new Uint8Array([5, 6, 7, 8]), | ||
| ]); | ||
|
|
||
| const reader = new Reader(mockStream); | ||
|
|
||
| const result1 = await reader.read(4); | ||
| expect(result1).toEqual(new Uint8Array([1, 2, 3, 4])); | ||
|
|
||
| const result2 = await reader.read(4); | ||
| expect(result2).toEqual(new Uint8Array([5, 6, 7, 8])); | ||
|
|
||
| // Reads after the stream is done should yield null | ||
| expect(await reader.read(10)).toBe(null); | ||
| }); | ||
|
|
||
| it('should wait for more bytes if not enough are available', async () => { | ||
| const mockStream = createMockStream([ | ||
| new Uint8Array([1, 2, 3]), | ||
| new Uint8Array([4, 5, 6]), | ||
| ]); | ||
|
|
||
| const reader = new Reader(mockStream); | ||
|
|
||
| const result = await reader.read(6); | ||
| expect(result).toEqual(new Uint8Array([1, 2, 3, 4, 5, 6])); | ||
| // Reads after the stream is done should yield null | ||
| expect(await reader.read(1)).toBe(null); | ||
| }); | ||
|
|
||
| it('should throw an error if read is called multiple times concurrently', async () => { | ||
| const mockStream = createMockStream([new Uint8Array([1, 2, 3])]); | ||
|
|
||
| const reader = new Reader(mockStream); | ||
|
|
||
| const read1 = reader.read(2); | ||
| const read2 = reader.read(2); | ||
|
|
||
| await expect(read2).rejects.toThrow('Called read more than once'); | ||
| await expect(read1).resolves.toEqual(new Uint8Array([1, 2])); | ||
| }); | ||
|
|
||
| it('should handle reading less than available bytes', async () => { | ||
| const mockStream = createMockStream([new Uint8Array([1, 2, 3, 4, 5])]); | ||
|
|
||
| const reader = new Reader(mockStream); | ||
|
|
||
| const result1 = await reader.read(3); | ||
| expect(result1).toEqual(new Uint8Array([1, 2, 3])); | ||
|
|
||
| const result2 = await reader.read(2); | ||
| expect(result2).toEqual(new Uint8Array([4, 5])); | ||
| }); | ||
| }); | ||
|
|
||
| describe('Reader with Content-Length', () => { | ||
| it('should initialize buffer size based on Content-Length header', async () => { | ||
| const mockStream = createMockStream([], 2048); | ||
| const reader = new Reader(mockStream); | ||
|
|
||
| // Read some bytes to indirectly validate initialization | ||
| const readPromise = reader.read(0); // No bytes to read, but ensures no errors | ||
| await expect(readPromise).resolves.toEqual(new Uint8Array(0)); | ||
| }); | ||
|
|
||
| it('should work if Content-Length is not set', async () => { | ||
| const mockStream = createMockStream([]); | ||
| const reader = new Reader(mockStream); | ||
|
|
||
| // Read some bytes to validate no Content-Length doesn't break initialization | ||
| const readPromise = reader.read(0); | ||
| await expect(readPromise).resolves.toEqual(new Uint8Array(0)); | ||
| }); | ||
|
|
||
| it('should handle reading bytes when Content-Length is specified', async () => { | ||
| const mockStream = createMockStream( | ||
| [new Uint8Array([1, 2, 3, 4])], | ||
| 2048 // Content-Length | ||
| ); | ||
|
|
||
| const reader = new Reader(mockStream); | ||
|
|
||
| const result = await reader.read(4); | ||
| expect(result).toEqual(new Uint8Array([1, 2, 3, 4])); | ||
| }); | ||
|
|
||
| it('should resize the buffer if more bytes are received than Content-Length', async () => { | ||
| const mockStream = createMockStream( | ||
| [new Uint8Array([1, 2, 3, 4]), new Uint8Array([5, 6, 7, 8])], | ||
| 4 // Content-Length is smaller than total bytes received | ||
| ); | ||
|
|
||
| const reader = new Reader(mockStream); | ||
|
|
||
| const result1 = await reader.read(4); | ||
| expect(result1).toEqual(new Uint8Array([1, 2, 3, 4])); | ||
|
|
||
| const result2 = await reader.read(4); | ||
| expect(result2).toEqual(new Uint8Array([5, 6, 7, 8])); | ||
| }); | ||
| }); |