test: Make the ReadRows service in tests more modular

[danieljbruce]

Summary

This PR makes the ReadRows service in tests more modular.

Background

This PR is made up of 4 already reviewed PRs merged onto this feature branch that are now being merged onto main.

1. this PR eliminates duplicate code and adds interfaces to simplify data flows
2. this PR eliminated the duplicate copy of the ReadRows service
3. this PR gets rid of a complex block of if statements
4. this PR breaks the ReadRow service down into classes.

[sofisl]

These combined ternary expressions are hard to read. Is this saying if the defaultKey is not undefined, then use default key, else use keyRequestClosed, and if that is undefined use it as opened, or else as closed? I think this might achieve the same thing and be easier to read, or if not maybe you can rewrite it in a way that preserves the logic:

return defaultKey || keyRequestClosed ? parseInt(keyRequestClosed as string) : parseInt(keyRequestOpen as string) + 1;

[danieljbruce]

We can't do it quite like this because when defaultKey is 0 it is falsy so it doesn't get used, but we want it to get used when it is 0. What I did instead was did an equal check instead of a not equal check.

[sofisl]

what's the difference between lastScannedRowKey and canSendMore? Maybe a comment describing the branches would be useful. Also, is it possible to have a lastScannedRowKey and !canSendMore? Otherwise perhaps an else might be useful.

[danieljbruce]

From this, canSendMore is the value returned indicating whether or not we want the stream to drain. lastScannedRowKey is actually part of the response so I think both the if statements are independent.

[danieljbruce]

This code is pretty much an exact copy of what we have on main:

nodejs-bigtable/test/utils/readRowsImpl.ts

Lines 191 to 210 in ed942c5

	const canSendMore = stream.write(response);
	if (response.chunks && response.chunks.length > 0) {
	debugLog(`sent ${response.chunks.length} chunks`);
	}
	if (response.lastScannedRowKey) {
	const binaryKey = Buffer.from(
	response.lastScannedRowKey as string,
	'base64'
	);
	const stringKey = binaryKey.toString();
	debugLog(`sent lastScannedRowKey = ${stringKey}`);
	}
	if (!canSendMore) {
	debugLog('awaiting for back pressure');
	await new Promise<void>(resolve => {
	stopWaiting = resolve;
	stream.once('drain', resolve);
	});
	}
	resolve();

[danieljbruce]

I added comments for the if blocks