Skip to content

Commit a0a44fa

Browse files
amydevsamydevs
committed
lintfix
1 parent 4eb9370 commit a0a44fa

5 files changed

+66
-68
lines changed

‎src/WebSocketStream.ts

Lines changed: 27 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,11 @@
11
import type { StreamCodeToReason, StreamReasonToCode } from './types';
22
import type WebSocketConnection from './WebSocketConnection';
33
import type { StreamId, StreamMessage, VarInt } from './message';
4+
import {
5+
ReadableStream,
6+
WritableStream,
7+
CountQueuingStrategy,
8+
} from 'stream/web';
49
import { CreateDestroy, status } from '@matrixai/async-init/dist/CreateDestroy';
510
import Logger from '@matrixai/logger';
611
import { promise } from './utils';
@@ -13,12 +18,6 @@ import {
1318
StreamMessageType,
1419
StreamShutdown,
1520
} from './message';
16-
import {
17-
ReadableStream,
18-
WritableStream,
19-
CountQueuingStrategy,
20-
ReadableByteStreamController,
21-
} from 'stream/web';
2221
import WebSocketStreamQueue from './WebSocketStreamQueue';
2322

2423
interface WebSocketStream extends CreateDestroy {}
@@ -129,14 +128,14 @@ class WebSocketStream implements ReadableWritablePair<Uint8Array, Uint8Array> {
129128
return;
130129
}
131130

132-
// reset the promise before a read from the queue to wait until the queue has items
131+
// Reset the promise before a read from the queue to wait until the queue has items
133132
if (this.readableQueue.count === 0) {
134133
this.readableBufferReady.resolveP();
135134
this.readableBufferReady = promise<void>();
136135
}
137136
await this.readableBufferReady.p;
138137

139-
// data will be null in the case of stream destruction before the readable buffer is blocked
138+
// Data will be null in the case of stream destruction before the readable buffer is blocked
140139
// we're going to just enqueue an empty buffer in case it is null for some other reason, so that the next read is able to complete
141140
const data = this.readableQueue.dequeue();
142141
if (data == null) {
@@ -146,7 +145,9 @@ class WebSocketStream implements ReadableWritablePair<Uint8Array, Uint8Array> {
146145
const readBytes = data.length;
147146
controller.enqueue(data);
148147

149-
this.logger.debug(`${readBytes} bytes have been pushed onto stream buffer`)
148+
this.logger.debug(
149+
`${readBytes} bytes have been pushed onto stream buffer`,
150+
);
150151

151152
await this.streamSend({
152153
type: StreamMessageType.Ack,
@@ -158,9 +159,9 @@ class WebSocketStream implements ReadableWritablePair<Uint8Array, Uint8Array> {
158159
await this.signalReadableEnd(true, reason);
159160
},
160161
},
161-
{
162+
new CountQueuingStrategy({
162163
highWaterMark: 1,
163-
}
164+
}),
164165
);
165166

166167
this.writable = new WritableStream(
@@ -180,25 +181,28 @@ class WebSocketStream implements ReadableWritablePair<Uint8Array, Uint8Array> {
180181

181182
await this.writableDesiredSizeProm.p;
182183

183-
// chunking
184-
let data: Uint8Array;
184+
// Chunking
185+
// .subarray parameters begin and end are clamped to the size of the Uint8Array
186+
const data = chunk.subarray(0, this.writableDesiredSize);
185187
if (chunk.length > this.writableDesiredSize) {
186188
this.logger.debug(
187189
`this chunk will be split into sizes of ${this.writableDesiredSize} bytes`,
188190
);
189191
}
190-
// .subarray parameters begin and end are clamped to the size of the Uint8Array
191-
data = chunk.subarray(0, this.writableDesiredSize);
192192

193193
const bytesWritten = data.length;
194194
if (this.writableDesiredSize === bytesWritten) {
195-
this.logger.debug(`this chunk will trigger receiver to send an ACK`);
195+
this.logger.debug(
196+
`this chunk will trigger receiver to send an ACK`,
197+
);
196198
// Reset the promise to wait for another ACK
197199
this.writableDesiredSizeProm = promise();
198200
}
199201
// Decrement the desired size by the amount of bytes written
200202
this.writableDesiredSize -= bytesWritten;
201-
this.logger.debug(`writableDesiredSize is now ${this.writableDesiredSize} due to write`);
203+
this.logger.debug(
204+
`writableDesiredSize is now ${this.writableDesiredSize} due to write`,
205+
);
202206
await this.streamSend({
203207
type: StreamMessageType.Data,
204208
payload: data,
@@ -284,13 +288,16 @@ class WebSocketStream implements ReadableWritablePair<Uint8Array, Uint8Array> {
284288
if (parsedMessage.type === StreamMessageType.Ack) {
285289
this.writableDesiredSize += parsedMessage.payload;
286290
this.writableDesiredSizeProm.resolveP();
287-
this.logger.debug(`writableDesiredSize is now ${this.writableDesiredSize} due to ACK`);
291+
this.logger.debug(
292+
`writableDesiredSize is now ${this.writableDesiredSize} due to ACK`,
293+
);
288294
} else if (parsedMessage.type === StreamMessageType.Data) {
289295
if (this._readableEnded) {
290296
return;
291297
}
292298
if (
293-
parsedMessage.payload.length > (this.readableQueueBufferSize - this.readableQueue.length)
299+
parsedMessage.payload.length >
300+
this.readableQueueBufferSize - this.readableQueue.length
294301
) {
295302
await this.signalReadableEnd(
296303
true,
@@ -372,7 +379,7 @@ class WebSocketStream implements ReadableWritablePair<Uint8Array, Uint8Array> {
372379
this._readableEnded = true;
373380
// Resolve readable promise in case blocking
374381
this.readableBufferReady.resolveP();
375-
// clear the readable queue
382+
// Clear the readable queue
376383
this.readableQueue.clear();
377384
// Shutdown the write side of the other stream
378385
if (isError) {

‎src/WebSocketStreamQueue.ts

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ class WebSocketStreamQueue {
1919
* The combined length of the queued `Uint8Array`s.
2020
*/
2121
public get length(): Readonly<number> {
22-
return this._length
22+
return this._length;
2323
}
2424
/**
2525
* The number of queued `Uint8Array`.
@@ -35,19 +35,19 @@ class WebSocketStreamQueue {
3535
}
3636
public queue(data: Uint8Array): void {
3737
const item = {
38-
data
38+
data,
3939
};
40-
// if there is no head, then this is the first item in the queue
40+
// If there is no head, then this is the first item in the queue
4141
if (this.head == null) {
4242
this.head = item;
4343
}
44-
// if the tail exists, then set the next item on the tail to the new item
44+
// If the tail exists, then set the next item on the tail to the new item
4545
if (this.tail != null) {
4646
this.tail.next = item;
4747
}
48-
// set the tail to the new item
48+
// Set the tail to the new item
4949
this.tail = item;
50-
// update the byteLength, length, and count
50+
// Update the byteLength, length, and count
5151
this._byteLength += data.byteLength;
5252
this._length += data.length;
5353
this._count++;
@@ -57,15 +57,15 @@ class WebSocketStreamQueue {
5757
* If the queue is empty, then undefined is returned.
5858
*/
5959
public dequeue(): Uint8Array | undefined {
60-
// get the data of the head
60+
// Get the data of the head
6161
const oldData = this.head?.data;
6262
const newHead = this.head?.next;
63-
// if the head and the tail are the same, then the queue is either empty or only have one item
63+
// If the head and the tail are the same, then the queue is either empty or only have one item
6464
if (this.head === this.tail) {
6565
this.tail = undefined;
6666
}
6767
this.head = newHead;
68-
// decrement the count, but don't let it go below 0 in case the queue is empty
68+
// Decrement the count, but don't let it go below 0 in case the queue is empty
6969
this._count = this._count === 0 ? 0 : this._count - 1;
7070
this._byteLength -= oldData?.byteLength ?? 0;
7171
this._length -= oldData?.length ?? 0;
@@ -75,7 +75,7 @@ class WebSocketStreamQueue {
7575
this._byteLength = 0;
7676
this._length = 0;
7777
this._count = 0;
78-
// clearing head and tail should cause the garbage collector to clean up all the items in the queue
78+
// Clearing head and tail should cause the garbage collector to clean up all the items in the queue
7979
this.head = undefined;
8080
this.tail = undefined;
8181
}
@@ -84,6 +84,6 @@ class WebSocketStreamQueue {
8484
type WebSocketStreamQueueItem = {
8585
data: Uint8Array;
8686
next?: WebSocketStreamQueueItem;
87-
}
87+
};
8888

8989
export default WebSocketStreamQueue;

‎src/index.ts

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,4 +10,3 @@ export * as events from './events';
1010
export * as errors from './errors';
1111
export * as config from './config';
1212
export * as message from './message';
13-

‎tests/WebSocketStream.test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -517,7 +517,7 @@ describe(WebSocketStream.name, () => {
517517

518518
await Promise.all([writer.close(), stream1.writable.close()]);
519519
});
520-
// test('stream can error when blocked on data', async () => {
520+
// Test('stream can error when blocked on data', async () => {
521521
// const [stream1, stream2] = await createStreamPair();
522522

523523
// const message = new Uint8Array(STREAM_BUFFER_SIZE * 2);

‎tests/WebSocketStreamQueue.test.ts

Lines changed: 27 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,32 @@
1-
import WebSocketStreamQueue from "@/WebSocketStreamQueue";
2-
import { fc, testProp } from "@fast-check/jest";
1+
import { fc, testProp } from '@fast-check/jest';
2+
import WebSocketStreamQueue from '@/WebSocketStreamQueue';
33

44
describe(WebSocketStreamQueue.name, () => {
5-
testProp(
6-
"should queue items",
7-
[fc.array(fc.uint8Array())],
8-
(array) => {
9-
const queue = new WebSocketStreamQueue();
10-
let totalLength = 0;
11-
let totalByteLength = 0;
12-
for (const buffer of array) {
13-
queue.queue(buffer);
14-
totalByteLength += buffer.byteLength;
15-
totalLength += buffer.length;
16-
}
17-
expect(queue.count).toBe(array.length);
18-
expect(queue.byteLength).toBe(totalByteLength);
19-
expect(queue.length).toBe(totalLength);
5+
testProp('should queue items', [fc.array(fc.uint8Array())], (array) => {
6+
const queue = new WebSocketStreamQueue();
7+
let totalLength = 0;
8+
let totalByteLength = 0;
9+
for (const buffer of array) {
10+
queue.queue(buffer);
11+
totalByteLength += buffer.byteLength;
12+
totalLength += buffer.length;
2013
}
21-
);
22-
testProp(
23-
"should dequeue items",
24-
[fc.array(fc.uint8Array())],
25-
(array) => {
26-
const queue = new WebSocketStreamQueue();
27-
for (const buffer of array) {
28-
queue.queue(buffer);
29-
}
30-
const result: Array<Uint8Array> = [];
31-
for (let i = 0; i < array.length; i++) {
32-
result.push(queue.dequeue()!);
33-
}
34-
expect(result).toEqual(array);
35-
expect(queue.count).toBe(0)
36-
expect(queue.byteLength).toBe(0);
37-
expect(queue.length).toBe(0);
14+
expect(queue.count).toBe(array.length);
15+
expect(queue.byteLength).toBe(totalByteLength);
16+
expect(queue.length).toBe(totalLength);
17+
});
18+
testProp('should dequeue items', [fc.array(fc.uint8Array())], (array) => {
19+
const queue = new WebSocketStreamQueue();
20+
for (const buffer of array) {
21+
queue.queue(buffer);
3822
}
39-
);
23+
const result: Array<Uint8Array> = [];
24+
for (let i = 0; i < array.length; i++) {
25+
result.push(queue.dequeue()!);
26+
}
27+
expect(result).toEqual(array);
28+
expect(queue.count).toBe(0);
29+
expect(queue.byteLength).toBe(0);
30+
expect(queue.length).toBe(0);
31+
});
4032
});

0 commit comments

Comments
 (0)