Source code
Revision control
Copy as Markdown
Other Tools
/**
* @license
* Copyright 2024 Google Inc.
* SPDX-License-Identifier: Apache-2.0
*/
import {Readable, Writable} from 'node:stream';
import {describe, it, beforeEach, afterEach} from 'node:test';
import expect from 'expect';
import {PipeTransport} from './PipeTransport.js';
describe('PipeTransport', () => {
let transport: PipeTransport;
let myReadable: Readable;
async function waitForNextMessage() {
return await new Promise<string>(res => {
transport.onmessage = message => {
res(message);
};
});
}
function waitForNumberOfMessages(count: number): Promise<string[]> {
const messages: string[] = [];
return new Promise<string[]>(resolve => {
transport.onmessage = (message: string) => {
messages.push(message);
if (messages.length === count) {
resolve(messages);
}
};
});
}
beforeEach(() => {
const myWritable = new Writable({
write(_chunk: string, _encoding: string, callback) {
callback();
},
writev(_chunks: any[], callback) {
callback();
},
});
myReadable = new Readable({
read() {
// No-op as we will push data manually.
},
});
transport = new PipeTransport(myWritable, myReadable);
});
afterEach(() => {
transport.close();
});
it('should dispatch messages in order handling microtasks for each message first', async () => {
const log: string[] = [];
const result = new Promise<void>(resolve => {
transport.onmessage = message => {
log.push('message received ' + message);
return Promise.resolve().then(() => {
log.push('microtask1 ' + message);
return Promise.resolve().then(() => {
log.push('microtask2 ' + message);
if (log.length === 6) {
resolve();
}
});
});
};
});
myReadable.push('m1\0');
myReadable.push('m2\0');
await result;
expect(log).toEqual([
'message received m1',
'microtask1 m1',
'microtask2 m1',
'message received m2',
'microtask1 m2',
'microtask2 m2',
]);
});
describe('message handling', () => {
it('should work with message with ending', async () => {
let message = waitForNextMessage();
myReadable.push('m1\0');
expect(await message).toBe('m1');
message = waitForNextMessage();
myReadable.push('m2\0');
expect(await message).toBe('m2');
});
it('should work for messages ending in multiple lines', async () => {
const message = waitForNextMessage();
myReadable.push('Hello wor');
myReadable.push('ld!\0');
expect(await message).toBe('Hello world!');
});
it('should work with messages continuing from previous one', async () => {
let message = waitForNextMessage();
myReadable.push('Hello wor');
myReadable.push('ld!\0I started in ');
expect(await message).toBe('Hello world!');
message = waitForNextMessage();
myReadable.push('the previous message\0');
expect(await message).toBe('I started in the previous message');
});
it('should work with multiple messages in a single line', async () => {
const messagesPromise = waitForNumberOfMessages(3);
myReadable.push('First\0Second\0Third\0');
const messages = await messagesPromise;
expect(messages).toHaveLength(3);
expect(messages[0]).toBe('First');
expect(messages[1]).toBe('Second');
expect(messages[2]).toBe('Third');
});
});
});