{

Command = command;

Debug.Assert(command.Connection != null && command.Connection == Connector.Connection);

_connection = command.Connection!;

_behavior = behavior;

_isSchemaOnly = _behavior.HasFlag(CommandBehavior.SchemaOnly);

_isSequential = _behavior.HasFlag(CommandBehavior.SequentialAccess);

_statements = statements;

StatementIndex = -1;

_sendTask = sendTask;

State = ReaderState.BetweenResults;

_recordsAffected = null;

{

CheckClosed();

var fastRead = TryFastRead();

return fastRead.HasValue

? fastRead.Value

: Read(false).GetAwaiter().GetResult();

{

CheckClosed();

if (cancellationToken.IsCancellationRequested)

return Task.FromCanceled<bool>(cancellationToken);

var fastRead = TryFastRead();

if (fastRead.HasValue)

return fastRead.Value ? PGUtil.TrueTask : PGUtil.FalseTask;

using (NoSynchronizationContextScope.Enter())

return Read(true);

{

// This is an optimized execution path that avoids calling any async methods for the (usual)

// case where the next row (or CommandComplete) is already in memory.

if (_behavior.HasFlag(CommandBehavior.SingleRow))

return null;

switch (State)

{

case ReaderState.BeforeResult:

// First Read() after NextResult. Data row has already been processed.

State = ReaderState.InResult;

return true;

case ReaderState.InResult:

if (_isSequential)

return null;

ConsumeRowNonSequential();

break;

case ReaderState.BetweenResults:

case ReaderState.Consumed:

case ReaderState.Closed:

return false;

}

var readBuf = Connector.ReadBuffer;

if (readBuf.ReadBytesLeft < 5)

return null;

var messageCode = (BackendMessageCode)readBuf.ReadByte();

var len = readBuf.ReadInt32() - 4; // Transmitted length includes itself

if (messageCode != BackendMessageCode.DataRow || readBuf.ReadBytesLeft < len)

{

readBuf.ReadPosition -= 5;

return null;

}

var msg = Connector.ParseServerMessage(readBuf, messageCode, len, false)!;

Debug.Assert(msg.Code == BackendMessageCode.DataRow);

ProcessMessage(msg);

return true;

{

switch (State)

{

case ReaderState.BeforeResult:

// First Read() after NextResult. Data row has already been processed.

State = ReaderState.InResult;

return true;

case ReaderState.InResult:

await ConsumeRow(async);

if (_behavior.HasFlag(CommandBehavior.SingleRow))

{

// TODO: See optimization proposal in #410

await Consume(async);

return false;

}

break;

case ReaderState.BetweenResults:

case ReaderState.Consumed:

case ReaderState.Closed:

return false;

default:

throw new ArgumentOutOfRangeException();

}

try

{

var msg2 = await ReadMessage(async);

ProcessMessage(msg2);

return msg2.Code == BackendMessageCode.DataRow;

}

catch (PostgresException)

{

State = ReaderState.Consumed;

throw;

}

{

return _isSequential ? ReadMessageSequential(async) : Connector.ReadMessage(async);

async ValueTask<IBackendMessage> ReadMessageSequential(bool async2)

{

var msg = await Connector.ReadMessage(async2, DataRowLoadingMode.Sequential);

if (msg.Code == BackendMessageCode.DataRow)

{

// Make sure that the datarow's column count is already buffered

await Connector.ReadBuffer.Ensure(2, async);

return msg;

}

return msg;

}

{

try

{

return (_isSchemaOnly ? NextResultSchemaOnly(false) : NextResult(false))

.GetAwaiter().GetResult();

}

catch (PostgresException e)

{

State = ReaderState.Consumed;

if (StatementIndex >= 0 && StatementIndex < _statements.Count)

e.Statement = _statements[StatementIndex];

throw;

}

{

if (cancellationToken.IsCancellationRequested)

return Task.FromCanceled<bool>(cancellationToken);

try

{

using (NoSynchronizationContextScope.Enter())

return _isSchemaOnly ? NextResultSchemaOnly(true) : NextResult(true);

}

catch (PostgresException e)

{

State = ReaderState.Consumed;

if (StatementIndex >= 0 && StatementIndex < _statements.Count)

e.Statement = _statements[StatementIndex];

throw;

}

{

CheckClosed();

IBackendMessage msg;

Debug.Assert(!_isSchemaOnly);

// If we're in the middle of a resultset, consume it

switch (State)

{

case ReaderState.BeforeResult:

case ReaderState.InResult:

await ConsumeRow(async);

while (true)

{

var completedMsg = await Connector.ReadMessage(async, DataRowLoadingMode.Skip);

switch (completedMsg.Code)

{

case BackendMessageCode.CompletedResponse:

case BackendMessageCode.EmptyQueryResponse:

ProcessMessage(completedMsg);

break;

default:

continue;

}

break;

}

break;

case ReaderState.BetweenResults:

break;

case ReaderState.Consumed:

case ReaderState.Closed:

return false;

default:

throw new ArgumentOutOfRangeException();

}

Debug.Assert(State == ReaderState.BetweenResults);

_hasRows = false;

if (_behavior.HasFlag(CommandBehavior.SingleResult) && StatementIndex == 0 && !isConsuming)

{

await Consume(async);

return false;

}

// We are now at the end of the previous result set. Read up to the next result set, if any.

// Non-prepared statements receive ParseComplete, BindComplete, DescriptionRow/NoData,

// prepared statements receive only BindComplete

for (StatementIndex++; StatementIndex < _statements.Count; StatementIndex++)

{

var statement = _statements[StatementIndex];

if (statement.IsPrepared)

{

Expect<BindCompleteMessage>(await Connector.ReadMessage(async), Connector);

RowDescription = statement.Description;

}

else // Non-prepared flow

{

var pStatement = statement.PreparedStatement;

if (pStatement != null)

{

Debug.Assert(!pStatement.IsPrepared);

Debug.Assert(pStatement.Description == null);

if (pStatement.StatementBeingReplaced != null)

{

Expect<CloseCompletedMessage>(await Connector.ReadMessage(async), Connector);

pStatement.StatementBeingReplaced.CompleteUnprepare();

pStatement.StatementBeingReplaced = null;

}

}

try

{

Expect<ParseCompleteMessage>(await Connector.ReadMessage(async), Connector);

}

catch

{

// An exception occurred. Check if any statements we being prepared and revert our bookkeeping.

pStatement?.CompleteUnprepare();

throw;

}

Expect<BindCompleteMessage>(await Connector.ReadMessage(async), Connector);

msg = await Connector.ReadMessage(async);

RowDescription = statement.Description = msg.Code switch

{

BackendMessageCode.NoData => null,

// RowDescription messages are cached on the connector, but if we're auto-preparing, we need to

// clone our own copy which will last beyond the lifetime of this invocation.

BackendMessageCode.RowDescription => pStatement == null

? (RowDescriptionMessage)msg

: ((RowDescriptionMessage)msg).Clone(),

_ => throw Connector.UnexpectedMessageReceived(msg.Code)

};

pStatement?.CompletePrepare();

}

if (RowDescription == null)

{

// Statement did not generate a resultset (e.g. INSERT)

// Read and process its completion message and move on to the next statement

msg = await ReadMessage(async);

switch (msg.Code)

{

case BackendMessageCode.CompletedResponse:

case BackendMessageCode.EmptyQueryResponse:

break;

default:

throw Connector.UnexpectedMessageReceived(msg.Code);

}

ProcessMessage(msg);

continue;

}

if (StatementIndex == 0 && Command.Parameters.HasOutputParameters)

{

// If output parameters are present and this is the first row of the first resultset,

// we must always read it in non-sequential mode because it will be traversed twice (once

// here for the parameters, then as a regular row).

msg = await Connector.ReadMessage(async);

ProcessMessage(msg);

if (msg.Code == BackendMessageCode.DataRow)

PopulateOutputParameters();

}

else

{

msg = await ReadMessage(async);

ProcessMessage(msg);

}

switch (msg.Code)

{

case BackendMessageCode.DataRow:

case BackendMessageCode.CompletedResponse:

break;

default:

throw Connector.UnexpectedMessageReceived(msg.Code);

}

return true;

}

// There are no more queries, we're done. Read to the RFQ.

ProcessMessage(Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector));

RowDescription = null;

return false;

}

void PopulateOutputParameters()

{

// The first row in a stored procedure command that has output parameters needs to be traversed twice -

// once for populating the output parameters and once for the actual result set traversal. So in this

// case we can't be sequential.

Debug.Assert(Command.Parameters.Any(p => p.IsOutputDirection));

Debug.Assert(StatementIndex == 0);

Debug.Assert(RowDescription != null);

Debug.Assert(State == ReaderState.BeforeResult);

// Temporarily set our state to InResult to allow us to read the values

State = ReaderState.InResult;

var pending = new Queue<object>();

var taken = new List<NpgsqlParameter>();

for (var i = 0; i < FieldCount; i++)

{

if (Command.Parameters.TryGetValue(GetName(i), out var p) && p.IsOutputDirection)

{

p.Value = GetValue(i);

taken.Add(p);

}

else

pending.Enqueue(GetValue(i));

}

// Not sure where this odd behavior comes from: all output parameters which did not get matched by

// name now get populated with column values which weren't matched. Keeping this for backwards compat,

// opened #2252 for investigation.

foreach (var p in Command.Parameters.Where(p => p.IsOutputDirection && !taken.Contains(p)))

{

if (pending.Count == 0)

break;

p.Value = pending.Dequeue();

}

State = ReaderState.BeforeResult; // Set the state back

}

/// <summary>

/// Note that in SchemaOnly mode there are no resultsets, and we read nothing from the backend (all

/// RowDescriptions have already been processed and are available)

/// </summary>

async Task<bool> NextResultSchemaOnly(bool async)

{

Debug.Assert(_isSchemaOnly);

for (StatementIndex++; StatementIndex < _statements.Count; StatementIndex++)

{

var statement = _statements[StatementIndex];

if (statement.IsPrepared)

{

// Row descriptions have already been populated in the statement objects at the

// Prepare phase

RowDescription = _statements[StatementIndex].Description;

}

else

{

Expect<ParseCompleteMessage>(await Connector.ReadMessage(async), Connector);

Expect<ParameterDescriptionMessage>(await Connector.ReadMessage(async), Connector);

var msg = await Connector.ReadMessage(async);

switch (msg.Code)

{

case BackendMessageCode.NoData:

RowDescription = _statements[StatementIndex].Description = null;

break;

case BackendMessageCode.RowDescription:

// We have a resultset

RowDescription = _statements[StatementIndex].Description = (RowDescriptionMessage)msg;

Command.FixupRowDescription(RowDescription, StatementIndex == 0);

break;

default:

throw Connector.UnexpectedMessageReceived(msg.Code);

}

}

// Found a resultset

if (RowDescription != null)

return true;

}

// There are no more queries, we're done. Read to the RFQ.

if (!_statements.All(s => s.IsPrepared))

{

ProcessMessage(Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector));

RowDescription = null;

}

return false;

}

#endregion

#region ProcessMessage

internal void ProcessMessage(IBackendMessage msg)

{

switch (msg.Code)

{

case BackendMessageCode.DataRow:

ProcessDataRowMessage((DataRowMessage)msg);

return;

case BackendMessageCode.CompletedResponse:

var completed = (CommandCompleteMessage)msg;

switch (completed.StatementType)

{

case StatementType.Update:

case StatementType.Insert:

case StatementType.Delete:

case StatementType.Copy:

case StatementType.Move:

if (!_recordsAffected.HasValue)

_recordsAffected = 0;

_recordsAffected += completed.Rows;

break;

}

_statements[StatementIndex].ApplyCommandComplete(completed);

goto case BackendMessageCode.EmptyQueryResponse;

case BackendMessageCode.EmptyQueryResponse:

State = ReaderState.BetweenResults;

return;

case BackendMessageCode.ReadyForQuery:

State = ReaderState.Consumed;

return;

default:

throw new Exception("Received unexpected backend message of type " + msg.Code);

}

{

Connector.State = ConnectorState.Fetching;

// The connector's buffer can actually change between DataRows:

// If a large DataRow exceeding the connector's current read buffer arrives, and we're

// reading in non-sequential mode, a new oversize buffer is allocated. We thus have to

// recapture the connector's buffer on each new DataRow.

// Note that this can happen even in sequential mode, if the row description message is big

// (see #2003)

Buffer = Connector.ReadBuffer;

_hasRows = true;

_column = -1;

ColumnLen = -1;

PosInColumn = 0;

// We assume that the row's number of columns is identical to the description's

_numColumns = Buffer.ReadInt16();

Debug.Assert(_numColumns == RowDescription!.NumFields,

$"Row's number of columns ({_numColumns}) differs from the row description's ({RowDescription.NumFields})");

if (!_isSequential)

{

_dataMsgEnd = Buffer.ReadPosition + msg.Length - 2;

// Initialize our columns array with the offset and length of the first column

_columns.Clear();

var len = Buffer.ReadInt32();

_columns.Add((Buffer.ReadPosition, len));

}

switch (State)

{

case ReaderState.BetweenResults:

State = ReaderState.BeforeResult;

break;

case ReaderState.BeforeResult:

State = ReaderState.InResult;

break;

case ReaderState.InResult:

break;

default:

throw Connector.UnexpectedMessageReceived(BackendMessageCode.DataRow);

}

{

get

{

CheckClosed();

return RowDescription?.NumFields ?? 0;

}