Skip to content
New issue

Have a question about this project? # for a free GitHub account to open an issue and contact its maintainers and the community.

#

By clicking “#”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? # to your account

Remove unncessary use of RPCProtocol. Fixes #11970 #11972

Merged
merged 1 commit into from
Jan 9, 2023
Merged

Remove unncessary use of RPCProtocol. Fixes #11970 #11972

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 8 additions & 13 deletions packages/terminal/src/browser/terminal-widget-impl.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
import { Terminal, RendererType } from 'xterm';
import { FitAddon } from 'xterm-addon-fit';
import { inject, injectable, named, postConstruct } from '@theia/core/shared/inversify';
import { ContributionProvider, Disposable, Event, Emitter, ILogger, DisposableCollection, RpcProtocol, RequestHandler } from '@theia/core';
import { ContributionProvider, Disposable, Event, Emitter, ILogger, DisposableCollection, Channel } from '@theia/core';
import { Widget, Message, WebSocketConnectionProvider, StatefulWidget, isFirefox, MessageLoop, KeyCode, codicon, ExtractableWidget } from '@theia/core/lib/browser';
import { isOSX } from '@theia/core/lib/common';
import { WorkspaceService } from '@theia/workspace/lib/browser';
Expand Down Expand Up @@ -66,7 +66,7 @@ export class TerminalWidgetImpl extends TerminalWidget implements StatefulWidget
protected searchBox: TerminalSearchWidget;
protected restored = false;
protected closeOnDispose = true;
protected waitForConnection: Deferred<RpcProtocol> | undefined;
protected waitForConnection: Deferred<Channel> | undefined;
protected linkHover: HTMLDivElement;
protected linkHoverButton: HTMLAnchorElement;
protected lastTouchEnd: TouchEvent | undefined;
Expand Down Expand Up @@ -563,23 +563,18 @@ export class TerminalWidgetImpl extends TerminalWidget implements StatefulWidget
}
this.toDisposeOnConnect.dispose();
this.toDispose.push(this.toDisposeOnConnect);
const waitForConnection = this.waitForConnection = new Deferred<RpcProtocol>();
const waitForConnection = this.waitForConnection = new Deferred<Channel>();
this.webSocketConnectionProvider.listen({
path: `${terminalsPath}/${this.terminalId}`,
onConnection: connection => {
const requestHandler: RequestHandler = _method => this.logger.warn('Received an unhandled RPC request from the terminal process');

const rpc = new RpcProtocol(connection, requestHandler);
rpc.onNotification(event => {
if (event.method === 'onData') {
this.write(event.args[0]);
}
connection.onMessage(e => {
this.write(e().readString());
});

// Excludes the device status code emitted by Xterm.js
const sendData = (data?: string) => {
if (data && !this.deviceStatusCodes.has(data) && !this.disableEnterWhenAttachCloseListener()) {
return rpc.sendRequest('write', [data]);
connection.getWriteBuffer().writeString(data).commit();
}
};

Expand All @@ -590,7 +585,7 @@ export class TerminalWidgetImpl extends TerminalWidget implements StatefulWidget
connection.onClose(() => disposable.dispose());

if (waitForConnection) {
waitForConnection.resolve(rpc);
waitForConnection.resolve(connection);
}
}
}, { reconnecting: false });
Expand Down Expand Up @@ -664,7 +659,7 @@ export class TerminalWidgetImpl extends TerminalWidget implements StatefulWidget
sendText(text: string): void {
if (this.waitForConnection) {
this.waitForConnection.promise.then(connection =>
connection.sendRequest('write', [text])
connection.getWriteBuffer().writeString(text).commit()
);
}
}
Expand Down
8 changes: 4 additions & 4 deletions packages/terminal/src/node/buffering-stream.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,12 @@

import { wait } from '@theia/core/lib/common/promise-util';
import { expect } from 'chai';
import { BufferingStream } from './buffering-stream';
import { BufferBufferingStream } from './buffering-stream';

describe('BufferringStream', () => {

it('should emit whatever data was buffered before the timeout', async () => {
const buffer = new BufferingStream({ emitInterval: 1000 });
const buffer = new BufferBufferingStream({ emitInterval: 1000 });
const chunkPromise = waitForChunk(buffer);
buffer.push(Buffer.from([0]));
await wait(100);
Expand All @@ -33,14 +33,14 @@ describe('BufferringStream', () => {
});

it('should not emit chunks bigger than maxChunkSize', async () => {
const buffer = new BufferingStream({ maxChunkSize: 2 });
const buffer = new BufferBufferingStream({ maxChunkSize: 2 });
buffer.push(Buffer.from([0, 1, 2, 3, 4, 5]));
expect(await waitForChunk(buffer)).deep.eq(Buffer.from([0, 1]));
expect(await waitForChunk(buffer)).deep.eq(Buffer.from([2, 3]));
expect(await waitForChunk(buffer)).deep.eq(Buffer.from([4, 5]));
});

function waitForChunk(buffer: BufferingStream): Promise<Buffer> {
function waitForChunk(buffer: BufferBufferingStream): Promise<Buffer> {
return new Promise(resolve => buffer.onData(resolve));
}
});
43 changes: 30 additions & 13 deletions packages/terminal/src/node/buffering-stream.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,27 +33,32 @@ export interface BufferingStreamOptions {
* every {@link BufferingStreamOptions.emitInterval}. It will also ensure that
* the emitted chunks never exceed {@link BufferingStreamOptions.maxChunkSize}.
*/
export class BufferingStream {

protected buffer?: Buffer;
export class BufferingStream<T> {
protected buffer?: T;
protected timeout?: NodeJS.Timeout;
protected maxChunkSize: number;
protected emitInterval: number;

protected onDataEmitter = new Emitter<Buffer>();
protected onDataEmitter = new Emitter<T>();
protected readonly concat: (left: T, right: T) => T;
protected readonly slice: (what: T, start?: number, end?: number) => T;
protected readonly length: (what: T) => number;

constructor(options?: BufferingStreamOptions) {
this.emitInterval = options?.emitInterval ?? 16; // ms
this.maxChunkSize = options?.maxChunkSize ?? (256 * 1024); // bytes
constructor(options: BufferingStreamOptions = {}, concat: (left: T, right: T) => T, slice: (what: T, start?: number, end?: number) => T, length: (what: T) => number) {
this.emitInterval = options.emitInterval ?? 16; // ms
this.maxChunkSize = options.maxChunkSize ?? (256 * 1024); // bytes
this.concat = concat;
this.slice = slice;
this.length = length;
}

get onData(): Event<Buffer> {
get onData(): Event<T> {
return this.onDataEmitter.event;
}

push(chunk: Buffer): void {
push(chunk: T): void {
if (this.buffer) {
this.buffer = Buffer.concat([this.buffer, chunk]);
this.buffer = this.concat(this.buffer, chunk);
} else {
this.buffer = chunk;
this.timeout = setTimeout(() => this.emitBufferedChunk(), this.emitInterval);
Expand All @@ -67,12 +72,24 @@ export class BufferingStream {
}

protected emitBufferedChunk(): void {
this.onDataEmitter.fire(this.buffer!.slice(0, this.maxChunkSize));
if (this.buffer!.byteLength <= this.maxChunkSize) {
this.onDataEmitter.fire(this.slice(this.buffer!, 0, this.maxChunkSize));
if (this.length(this.buffer!) <= this.maxChunkSize) {
this.buffer = undefined;
} else {
this.buffer = this.buffer!.slice(this.maxChunkSize);
this.buffer = this.slice(this.buffer!, this.maxChunkSize);
this.timeout = setTimeout(() => this.emitBufferedChunk(), this.emitInterval);
}
}
}

export class StringBufferingStream extends BufferingStream<string> {
constructor(options: BufferingStreamOptions = {}) {
super(options, (left, right) => left.concat(right), (what, start, end) => what.slice(start, end), what => what.length);
}
}

export class BufferBufferingStream extends BufferingStream<Buffer> {
constructor(options: BufferingStreamOptions = {}) {
super(options, (left, right) => Buffer.concat([left, right]), (what, start, end) => what.slice(start, end), what => what.length);
}
}
27 changes: 13 additions & 14 deletions packages/terminal/src/node/terminal-backend-contribution.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,15 +15,15 @@
// *****************************************************************************

import { injectable, inject, named } from '@theia/core/shared/inversify';
import { ILogger, RequestHandler } from '@theia/core/lib/common';
import { ILogger } from '@theia/core/lib/common';
import { TerminalProcess, ProcessManager } from '@theia/process/lib/node';
import { terminalsPath } from '../common/terminal-protocol';
import { MessagingService } from '@theia/core/lib/node/messaging/messaging-service';
import { RpcProtocol } from '@theia/core/';
import { BufferingStream } from './buffering-stream';
import { StringBufferingStream } from './buffering-stream';

@injectable()
export class TerminalBackendContribution implements MessagingService.Contribution {
protected readonly decoder = new TextDecoder('utf-8');

@inject(ProcessManager)
protected readonly processManager: ProcessManager;
Expand All @@ -39,18 +39,17 @@ export class TerminalBackendContribution implements MessagingService.Contributio
const output = termProcess.createOutputStream();
// Create a RPC connection to the terminal process
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const requestHandler: RequestHandler = async (method: string, args: any[]) => {
if (method === 'write' && args[0]) {
termProcess.write(args[0]);
} else {
this.logger.warn('Terminal process received a request with an unsupported method or argument', { method, args });
}
};
channel.onMessage(e => {
termProcess.write(e().readString());
});

const rpc = new RpcProtocol(channel, requestHandler);
const buffer = new BufferingStream();
buffer.onData(chunk => rpc.sendNotification('onData', [chunk.toString('utf8')]));
output.on('data', chunk => buffer.push(Buffer.from(chunk, 'utf8')));
const buffer = new StringBufferingStream();
buffer.onData(chunk => {
channel.getWriteBuffer().writeString(chunk).commit();
});
output.on('data', chunk => {
buffer.push(chunk);
});
channel.onClose(() => {
buffer.dispose();
output.dispose();
Expand Down