Skip to content

Commit 782e76f

Browse files
wu-huiwu-hui
committed
Address comments
1 parent aa455bf commit 782e76f

File tree

3 files changed

+101
-79
lines changed

3 files changed

+101
-79
lines changed

‎packages/firestore/src/protos/firestore_proto_api.d.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -184,7 +184,7 @@ export declare namespace firestoreV1ApiClientInterfaces {
184184
interface Document {
185185
name?: string;
186186
fields?: ApiClientObjectMap<Value>;
187-
createTime?: string;
187+
createTime?: Timestamp;
188188
updateTime?: Timestamp;
189189
}
190190
interface DocumentChange {

‎packages/firestore/src/util/bundle_reader.ts

Lines changed: 76 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ import {
1919
BundleElement,
2020
BundleMetadata
2121
} from '../protos/firestore_bundle_proto';
22+
import { Deferred } from './promise';
23+
import { AsyncQueue } from './async_queue';
2224

2325
/**
2426
* A complete element in the bundle stream, together with the byte length it
@@ -40,7 +42,8 @@ export class SizedBundleElement {
4042
* Create a `ReadableStream` from a underlying buffer.
4143
*
4244
* @param data: Underlying buffer.
43-
* @param bytesPerRead: How many bytes to read from the underlying buffer from each read through the stream.
45+
* @param bytesPerRead: How many bytes to read from the underlying buffer from
46+
* each read through the stream.
4447
*/
4548
export function toReadableStream(
4649
data: Uint8Array | ArrayBuffer,
@@ -66,19 +69,25 @@ export function toReadableStream(
6669
* elements out of the underlying content.
6770
*/
6871
export class BundleReader {
69-
// Cached bundle metadata.
70-
private metadata?: BundleMetadata | null;
71-
// The reader instance of the given ReadableStream.
72+
/** Cached bundle metadata. */
73+
private metadata: Deferred<BundleMetadata> = new Deferred<BundleMetadata>();
74+
/** The reader instance of the given ReadableStream. */
7275
private reader: ReadableStreamDefaultReader;
73-
// Internal buffer to hold bundle content, accumulating incomplete element content.
76+
/**
77+
* Internal buffer to hold bundle content, accumulating incomplete element
78+
* content.
79+
*/
7480
private buffer: Uint8Array = new Uint8Array();
81+
/** The decoder used to parse binary data into strings. */
7582
private textDecoder = new TextDecoder('utf-8');
7683

7784
constructor(
7885
private bundleStream:
7986
| ReadableStream<Uint8Array | ArrayBuffer>
8087
| Uint8Array
81-
| ArrayBuffer
88+
| ArrayBuffer,
89+
/** Async queue used to perform bundle reading. */
90+
private asyncQueue: AsyncQueue = new AsyncQueue()
8291
) {
8392
if (
8493
bundleStream instanceof Uint8Array ||
@@ -87,56 +96,54 @@ export class BundleReader {
8796
this.bundleStream = toReadableStream(bundleStream);
8897
}
8998
this.reader = (this.bundleStream as ReadableStream).getReader();
99+
100+
this.nextElement().then(
101+
element => {
102+
if (element && element.isBundleMetadata()) {
103+
this.metadata.resolve(element.payload.metadata!);
104+
} else {
105+
const payload = (element || { payload: null }).payload;
106+
this.metadata.reject(
107+
new Error(`The first element of the bundle is not a metadata, it is
108+
${JSON.stringify(payload)}`)
109+
);
110+
}
111+
},
112+
error => {
113+
this.metadata.reject(error);
114+
}
115+
);
90116
}
91117

92118
/**
93119
* Returns the metadata of the bundle.
94120
*/
95121
async getMetadata(): Promise<BundleMetadata> {
96-
if (!this.metadata) {
97-
await this.nextElement();
98-
}
99-
100-
return this.metadata!;
122+
return this.metadata.promise;
101123
}
102124

103125
/**
104126
* Returns the next BundleElement (together with its byte size in the bundle)
105127
* that has not been read from underlying ReadableStream. Returns null if we
106128
* have reached the end of the stream.
107-
*
108-
* Throws an error if the first element is not a BundleMetadata.
109129
*/
110130
async nextElement(): Promise<SizedBundleElement | null> {
111-
const element = await this.readNextElement();
112-
if (!element) {
113-
return element;
114-
}
115-
116-
if (!this.metadata) {
117-
if (element.isBundleMetadata()) {
118-
this.metadata = element.payload.metadata;
119-
} else {
120-
this.raiseError(
121-
`The first element of the bundle is not a metadata, it is ${JSON.stringify(
122-
element.payload
123-
)}`
124-
);
125-
}
126-
}
127-
128-
return element;
131+
// Ensures `nextElementImpl` calls are executed sequentially before they
132+
// modifies internal buffer.
133+
return this.asyncQueue.enqueue(() => this.nextElementImpl());
129134
}
130135

131136
/**
132-
* Reads from the head of internal buffer, and pulling more data from underlying stream if a complete element
133-
* cannot be found, until an element(including the prefixed length and the JSON string) is found.
137+
* Reads from the head of internal buffer, and pulling more data from
138+
* underlying stream if a complete element cannot be found, until an
139+
* element(including the prefixed length and the JSON string) is found.
134140
*
135141
* Once a complete element is read, it is dropped from internal buffer.
136142
*
137-
* Returns either the bundled element, or null if we have reached the end of the stream.
143+
* Returns either the bundled element, or null if we have reached the end of
144+
* the stream.
138145
*/
139-
private async readNextElement(): Promise<SizedBundleElement | null> {
146+
private async nextElementImpl(): Promise<SizedBundleElement | null> {
140147
const lengthBuffer = await this.readLength();
141148
if (lengthBuffer === null) {
142149
return null;
@@ -148,29 +155,30 @@ export class BundleReader {
148155
this.raiseError(`length string (${lengthString}) is not valid number`);
149156
}
150157

151-
const jsonString = await this.readJsonString(lengthBuffer.length, length);
152-
// Update the internal buffer to drop the read length and json string.
153-
this.buffer = this.buffer.slice(lengthBuffer.length + length);
158+
const jsonString = await this.readJsonString(length);
154159

155160
return new SizedBundleElement(
156161
JSON.parse(jsonString),
157162
lengthBuffer.length + length
158163
);
159164
}
160165

161-
// First index of '{' from the underlying buffer.
166+
/** First index of '{' from the underlying buffer. */
162167
private indexOfOpenBracket(): number {
163168
return this.buffer.findIndex(v => v === '{'.charCodeAt(0));
164169
}
165170

166-
// Reads from the beginning of the internal buffer, until the first '{', and return
167-
// the content.
168-
// If reached end of the stream, returns a null.
171+
/**
172+
* Reads from the beginning of the internal buffer, until the first '{', and
173+
* return the content.
174+
*
175+
* If reached end of the stream, returns a null.
176+
*/
169177
private async readLength(): Promise<Uint8Array | null> {
170178
let position = this.indexOfOpenBracket();
171179
while (position < 0) {
172-
const bytesRead = await this.pullMoreDataToBuffer();
173-
if (bytesRead < 0) {
180+
const done = await this.pullMoreDataToBuffer();
181+
if (done) {
174182
if (this.buffer.length === 0) {
175183
return null;
176184
}
@@ -186,22 +194,30 @@ export class BundleReader {
186194
}
187195
}
188196

189-
return this.buffer.slice(0, position);
197+
const result = this.buffer.slice(0, position);
198+
// Update the internal buffer to drop the read length.
199+
this.buffer = this.buffer.slice(position);
200+
return result;
190201
}
191202

192-
// Reads from a specified position from the internal buffer, for a specified
193-
// number of bytes, pulling more data from the underlying stream if needed.
194-
//
195-
// Returns a string decoded from the read bytes.
196-
private async readJsonString(start: number, length: number): Promise<string> {
197-
while (this.buffer.length < start + length) {
198-
const bytesRead = await this.pullMoreDataToBuffer();
199-
if (bytesRead < 0) {
203+
/**
204+
* Reads from a specified position from the internal buffer, for a specified
205+
* number of bytes, pulling more data from the underlying stream if needed.
206+
*
207+
* Returns a string decoded from the read bytes.
208+
*/
209+
private async readJsonString(length: number): Promise<string> {
210+
while (this.buffer.length < length) {
211+
const done = await this.pullMoreDataToBuffer();
212+
if (done) {
200213
this.raiseError('Reached the end of bundle when more is expected.');
201214
}
202215
}
203216

204-
return this.textDecoder.decode(this.buffer.slice(start, start + length));
217+
const result = this.textDecoder.decode(this.buffer.slice(0, length));
218+
// Update the internal buffer to drop the read json string.
219+
this.buffer = this.buffer.slice(length);
220+
return result;
205221
}
206222

207223
private raiseError(message: string): void {
@@ -210,20 +226,20 @@ export class BundleReader {
210226
throw new Error(message);
211227
}
212228

213-
// Pulls more data from underlying stream to internal buffer.
214-
// Returns a boolean indicating whether the stream is finished.
215-
private async pullMoreDataToBuffer(): Promise<number> {
229+
/**
230+
* Pulls more data from underlying stream to internal buffer.
231+
* Returns a boolean indicating whether the stream is finished.
232+
*/
233+
private async pullMoreDataToBuffer(): Promise<boolean> {
216234
const result = await this.reader.read();
217-
let bytesRead = -1;
218235
if (!result.done) {
219-
bytesRead = result.value.length;
220236
const newBuffer = new Uint8Array(
221237
this.buffer.length + result.value.length
222238
);
223239
newBuffer.set(this.buffer);
224240
newBuffer.set(result.value, this.buffer.length);
225241
this.buffer = newBuffer;
226242
}
227-
return bytesRead;
243+
return result.done;
228244
}
229245
}

‎packages/firestore/test/unit/util/bundle.test.ts

Lines changed: 24 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import {
2121
toReadableStream
2222
} from '../../../src/util/bundle_reader';
2323
import { isNode } from '../../util/test_platform';
24+
import { BundleElement } from '../../../src/protos/firestore_bundle_proto';
2425

2526
function readableStreamFromString(
2627
content: string,
@@ -71,7 +72,7 @@ function lengthPrefixedString(o: {}): string {
7172
function genericBundleReadingTests(bytesPerRead: number): void {
7273
const encoder = new TextEncoder();
7374
// Setting up test data.
74-
const meta = {
75+
const meta: BundleElement = {
7576
metadata: {
7677
id: 'test-bundle',
7778
createTime: { seconds: 1577836805, nanos: 6 },
@@ -82,7 +83,7 @@ function genericBundleReadingTests(bytesPerRead: number): void {
8283
};
8384
const metaString = lengthPrefixedString(meta);
8485

85-
const doc1Meta = {
86+
const doc1Meta: BundleElement = {
8687
documentMetadata: {
8788
name:
8889
'projects/test-project/databases/(default)/documents/collectionId/doc1',
@@ -91,18 +92,18 @@ function genericBundleReadingTests(bytesPerRead: number): void {
9192
}
9293
};
9394
const doc1MetaString = lengthPrefixedString(doc1Meta);
94-
const doc1 = {
95+
const doc1: BundleElement = {
9596
document: {
9697
name:
9798
'projects/test-project/databases/(default)/documents/collectionId/doc1',
98-
createTime: { _seconds: 1, _nanoseconds: 2000000 },
99-
updateTime: { _seconds: 3, _nanoseconds: 4000 },
99+
createTime: { seconds: 1, nanos: 2000000 },
100+
updateTime: { seconds: 3, nanos: 4000 },
100101
fields: { foo: { stringValue: 'value' }, bar: { integerValue: -42 } }
101102
}
102103
};
103104
const doc1String = lengthPrefixedString(doc1);
104105

105-
const doc2Meta = {
106+
const doc2Meta: BundleElement = {
106107
documentMetadata: {
107108
name:
108109
'projects/test-project/databases/(default)/documents/collectionId/doc2',
@@ -111,18 +112,18 @@ function genericBundleReadingTests(bytesPerRead: number): void {
111112
}
112113
};
113114
const doc2MetaString = lengthPrefixedString(doc2Meta);
114-
const doc2 = {
115+
const doc2: BundleElement = {
115116
document: {
116117
name:
117118
'projects/test-project/databases/(default)/documents/collectionId/doc2',
118-
createTime: { _seconds: 1, _nanoseconds: 2000000 },
119-
updateTime: { _seconds: 3, _nanoseconds: 4000 },
119+
createTime: { seconds: 1, nanos: 2000000 },
120+
updateTime: { seconds: 3, nanos: 4000 },
120121
fields: { foo: { stringValue: 'value1' }, bar: { integerValue: 42 } }
121122
}
122123
};
123124
const doc2String = lengthPrefixedString(doc2);
124125

125-
const noDocMeta = {
126+
const noDocMeta: BundleElement = {
126127
documentMetadata: {
127128
name:
128129
'projects/test-project/databases/(default)/documents/collectionId/nodoc',
@@ -132,7 +133,7 @@ function genericBundleReadingTests(bytesPerRead: number): void {
132133
};
133134
const noDocMetaString = lengthPrefixedString(noDocMeta);
134135

135-
const limitQuery = {
136+
const limitQuery: BundleElement = {
136137
namedQuery: {
137138
name: 'limitQuery',
138139
bundledQuery: {
@@ -148,7 +149,7 @@ function genericBundleReadingTests(bytesPerRead: number): void {
148149
}
149150
};
150151
const limitQueryString = lengthPrefixedString(limitQuery);
151-
const limitToLastQuery = {
152+
const limitToLastQuery: BundleElement = {
152153
namedQuery: {
153154
name: 'limitToLastQuery',
154155
bundledQuery: {
@@ -310,23 +311,28 @@ function genericBundleReadingTests(bytesPerRead: number): void {
310311
async () => {
311312
await expect(
312313
parseThroughBundle('metadata: "no length prefix"', bytesPerRead)
313-
).to.be.rejected;
314+
).to.be.rejectedWith(
315+
'Reached the end of bundle when a length string is expected.'
316+
);
314317

315318
await expect(
316319
parseThroughBundle('{metadata: "no length prefix"}', bytesPerRead)
317-
).to.be.rejected;
320+
).to.be.rejectedWith('Unexpected end of JSON input');
318321

319322
await expect(
320323
parseThroughBundle(metaString + 'invalid-string', bytesPerRead, true)
321-
).to.be.rejected;
324+
).to.be.rejectedWith(
325+
'Reached the end of bundle when a length string is expected.'
326+
);
322327

323-
await expect(parseThroughBundle('1' + metaString, bytesPerRead)).to.be
324-
.rejected;
328+
await expect(
329+
parseThroughBundle('1' + metaString, bytesPerRead)
330+
).to.be.rejectedWith('Reached the end of bundle when more is expected.');
325331

326332
// First element is not BundleMetadata.
327333
await expect(
328334
parseThroughBundle(doc1MetaString + doc1String, bytesPerRead)
329-
).to.be.rejected;
335+
).to.be.rejectedWith('The first element of the bundle is not a metadata');
330336
}
331337
);
332338
}

0 commit comments

Comments
 (0)