Add RowStreamStateMachine

[fabianfett]

Motivation

To best support back-pressure, we extract the necessary state machine into a dedicated substate machine.

Changes

• Add RowStreamStateMachine

Note

This pr does not include any tests and the new code will not be hit. We will add tests to the ExtendedQueryStateMachine that test the new functionality once we wire the new substate machine up.

[gwynne]

LGTM , one style nit I didn't even handle correctly anyway 😅

[gwynne]

Feel free to "tweak" my "code style" (generous to call it that, I agree) as you find apropos, I mostly just care about the code de-duplication.

Suggested change

	case .waitingForRows(var buffer):
	self.state = .modifying
	buffer.append(newRow)
	self.state = .waitingForRows(buffer)
	// For all the following cases, please note:
	// Normally these code paths should never be hit. However there is one way to trigger
	// this:
	//
	// If the server decides to close a connection, NIO will forward all outstanding
	// `channelRead`s without waiting for a next `context.read` call. For this reason we might
	// receive new rows, when we don't expect them here.
	case .waitingForRead(var buffer):
	self.state = .modifying
	buffer.append(newRow)
	self.state = .waitingForRows(buffer)
	case .waitingForDemand(var buffer):
	self.state = .modifying
	buffer.append(newRow)
	self.state = .waitingForRows(buffer)
	case .waitingForReadOrDemand(var buffer):
	self.state = .modifying
	buffer.append(newRow)
	self.state = .waitingForRows(buffer)
	case .waitingForRows(var buffer),
	// For all the following cases, please note:
	// Normally these code paths should never be hit. However there is one way to trigger
	// this:
	//
	// If the server decides to close a connection, NIO will forward all outstanding
	// `channelRead`s without waiting for a next `context.read` call. For this reason we might
	// receive new rows, when we don't expect them here.
	.waitingForRead(var buffer),
	.waitingForDemand(var buffer),
	.waitingForReadOrDemand(var buffer):
	self.state = .modifying
	buffer.append(newRow)
	self.state = .waitingForRows(buffer

[gwynne]

Yeah, like this! This is better than my suggestion above, do it this way there too!