ReadableStreamMessageReader.listen should register data handler once per readeable

[karthiknadig]

The issue is that if the listen() is called multiple times. The callback gets overwritten but , we now have two onData events registered.

The proposal to change is this, so that onData is registered on the reader when the instance is created, and not on listen. Also, the dispose removes callback, and not the onData handler.

For efficiency, we don't need to process data unless there is a listener on. So, adding a isEmpty to the Emitter, and using emitter to manage callbacks and cleanup.

[dbaeumer]

I actually prefer to not allowing this and throw like coded here: dbaeumer/probable-muskox-peach

Can you explain why you want to call listen n times.

[karthiknadig]

I don’t want to call listen multiple times. My worry is since we allowed it. Throwing now may cause crashes. If that is not a risk then, exception option is better.

[karthiknadig]

If there is no concern with potentially crashing consumers of that API, then I like direction this implementation is going:

vscode-languageserver-node/jsonrpc/src/common/messageReader.ts

Lines 202 to 217 in a5a067b

	public listen(callback: DataCallback): Disposable {
	if (this.callback !== undefined) {
	throw new Error('Reader can only listen once.');
	}
	this.nextMessageLength = -1;
	this.messageToken = 0;
	this.partialMessageTimer = undefined;
	this.callback = callback;
	const disposables = new DisposableStore();
	disposables.add(this.readable.onData((data: Uint8Array) => {
	this.onData(data);
	}));
	disposables.add(this.readable.onError((error: any) => this.fireError(error)));
	disposables.add(this.readable.onClose(() => this.fireClose()));
	return disposables;
	}

I do want to suggest that, if the reader is only going to be bound to one readable, then all the events should be hooked up in the constructor. The only thing that the listen needs to do is this:

	public listen(callback: DataCallback): Disposable {
		if (this.callback !== undefined) {
			throw new Error('Reader can only listen once.');
		}
		this.callback = callback;
		return new Disposable(()=>({this.callback = undefined}));
	}

the constructor should be doing this:

        private disposables = new DisposableStore();
	public constructor(readable: RAL.ReadableStream, options?: RAL.MessageBufferEncoding | MessageReaderOptions) {
		super();
		this.readable = readable;
		this.options = ResolvedMessageReaderOptions.fromOptions(options);
		this.buffer = RAL().messageBuffer.create(this.options.charset);
		this._partialMessageTimeout = 10000;
                this.partialMessageTimer = undefined;
		this.nextMessageLength = -1;
		this.messageToken = 0;
		this.readSemaphore = new Semaphore(1);

                this.disposables.add(this.readable.onData((data: Uint8Array) => {
			if(this.callback){this.onData(data);}
		}));
		this.disposables.add(this.readable.onError((error: any) => {if(this.callback){this.fireError(error);}}));
		this.disposables.add(this.readable.onClose(() => {if(this.callback){this.fireClose();}}));
	}
        public dispose(): void {
		super.dispose();
		this.disposables.dispose();
                this.callback = undefined;
	}

[dbaeumer]

I agree with the approach of hooking things up in the constructor since we pass in the readable there.