#pragma warning disable

using System;

using System.Collections.Generic;

using System.ComponentModel;

using System.Data;

using System.Data.Common;

using System.Linq;

using System.Runtime.CompilerServices;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

using System.Diagnostics.Contracts;

using System.Globalization;

using System.Net.Sockets;

using JetBrains.Annotations;

using Npgsql.BackendMessages;

using Npgsql.FrontendMessages;

using Npgsql.Logging;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.ComponentModel;

using System.Data;

using System.Data.Common;

using System.Diagnostics.Contracts;

using System.IO;

using System.Net.Security;

using System.Net.Sockets;

using System.Reflection;

using System.Security.Cryptography;

using System.Security.Cryptography.X509Certificates;

using System.Threading;

using System.Threading.Tasks;

using JetBrains.Annotations;

#if NET45 || NET451 || DNX451

using System.Transactions;

#endif

using Npgsql.Logging;

using NpgsqlTypes;

using IsolationLevel = System.Data.IsolationLevel;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Collections.Generic;

using System.Data.Common;

using System.Diagnostics.Contracts;

using System.IO;

using System.Linq;

using System.Net;

using System.Net.Security;

using System.Net.Sockets;

using System.Runtime.CompilerServices;

using System.Security.Authentication;

using System.Security.Cryptography.X509Certificates;

using System.Threading;

using System.Threading.Tasks;

using JetBrains.Annotations;

using Npgsql.BackendMessages;

using Npgsql.FrontendMessages;

using Npgsql.Logging;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Collections;

using System.Collections.Generic;

using System.Data;

using System.Data.Common;

using System.Diagnostics;

using System.Diagnostics.Contracts;

using System.IO;

using System.Linq;

using System.Reflection;

using System.Runtime.CompilerServices;

using System.Threading;

using System.Threading.Tasks;

using JetBrains.Annotations;

using Npgsql.BackendMessages;

using Npgsql.FrontendMessages;

using Npgsql.TypeHandlers;

using Npgsql.TypeHandlers.NumericHandlers;

using NpgsqlTypes;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using Npgsql.FrontendMessages;

using System;

using System.Collections.Generic;

using System.Data;

using System.Linq;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using Npgsql.FrontendMessages;

using System;

using System.Collections.Generic;

using System.Diagnostics.Contracts;

using System.IO;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Data;

using System.Data.Common;

using System.Diagnostics.Contracts;

using System.Threading;

using System.Threading.Tasks;

using JetBrains.Annotations;

using Npgsql.BackendMessages;

using Npgsql.FrontendMessages;

using Npgsql.Logging;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Collections;

using System.Collections.Generic;

using System.Diagnostics.Contracts;

using System.Globalization;

using System.IO;

using System.Net;

using System.Runtime.CompilerServices;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Diagnostics.Contracts;

using System.Linq;

using System.Threading;

using System.Threading.Tasks;

using Npgsql.Logging;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Diagnostics.Contracts;

using System.IO;

using System.Net;

using System.Runtime.InteropServices;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Diagnostics.Contracts;

using System.IO;

using Npgsql.BackendMessages;

using NpgsqlTypes;

using System.Threading;

using System.Threading.Tasks;

using JetBrains.Annotations;

using Npgsql.TypeHandlers;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Collections;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Data;

using System.Linq;

using System.Reflection;

using System.Diagnostics.Contracts;

using JetBrains.Annotations;

using Npgsql.Logging;

using Npgsql.TypeHandlers;

using NpgsqlTypes;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Diagnostics.Contracts;

using System.IO;

using System.Net;

using System.Runtime.InteropServices;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

#pragma warning disable

using System;

using System.Collections.Generic;

using System.Diagnostics.Contracts;

using System.IO;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

using Npgsql;

using Npgsql.TypeHandlers;

using System.Threading;

using System.Threading.Tasks;

namespace Npgsql

{

public sealed partial class NpgsqlCommand

{

async Task<NpgsqlDataReader> ExecuteAsync(CancellationToken cancellationToken, CommandBehavior behavior = CommandBehavior.Default)

{

Validate();

if (!IsPrepared)

ProcessRawQuery();

LogCommand();

State = CommandState.InProgress;

try

{

_connector = Connection.Connector;

// If a cancellation is in progress, wait for it to "complete" before proceeding (#615)

lock (_connector.CancelLock)

{

}

// Send protocol messages for the command

// Unless this is a prepared SchemaOnly command, in which case we already have the RowDescriptions

// from the Prepare phase (no need to send anything).

if (!IsPrepared || (behavior & CommandBehavior.SchemaOnly) == 0)

{

_connector.UserTimeout = CommandTimeout * 1000;

_sendState = SendState.Start;

_writeStatementIndex = 0;

if (IsPrepared)

await SendAsync(PopulateExecutePrepared, cancellationToken);

else if ((behavior & CommandBehavior.SchemaOnly) == 0)

await SendAsync(PopulateExecuteNonPrepared, cancellationToken);

else

await SendAsync(PopulateExecuteSchemaOnly, cancellationToken);

}

var reader = new NpgsqlDataReader(this, behavior, _statements);

await reader.NextResultAsync(cancellationToken);

_connector.CurrentReader = reader;

return reader;

}

catch

{

State = CommandState.Idle;

throw;

}

}

async Task SendAsync(PopulateMethod populateMethod, CancellationToken cancellationToken)

{

while (true)

{

var directBuf = new DirectBuffer();

var completed = populateMethod(ref directBuf);

await _connector.SendBufferAsync(cancellationToken);

if (completed)

break; // Sent all messages

// The following is an optimization hack for writing large byte arrays without passing

// through our buffer

if (directBuf.Buffer != null)

{

await _connector.Stream.WriteAsync(directBuf.Buffer, directBuf.Offset, directBuf.Size == 0 ? directBuf.Buffer.Length : directBuf.Size, cancellationToken);

directBuf.Buffer = null;

directBuf.Size = 0;

}

if (_writeStatementIndex > 0)

{

// We've send all the messages for the first statement in a multistatement command.

// If we continue blocking writes for the rest of the messages, we risk a deadlock where

// PostgreSQL sends large results for the first statement, while we're sending large

// parameter data for the second. To avoid this, switch to async sends.

// See #641

RemainingSendTask = SendRemaining(populateMethod, CancellationToken.None);

return;

}

}

}

async Task<int> ExecuteNonQueryInternalAsync(CancellationToken cancellationToken)

{

Prechecks();

Log.Trace("ExecuteNonQuery", Connection.Connector.Id);

using (Connection.Connector.StartUserAction())

{

NpgsqlDataReader reader;

using (reader = await (ExecuteAsync(cancellationToken)))

{

while (await reader.NextResultAsync(cancellationToken))

{

}

}

return reader.RecordsAffected;

}

}

async Task<object> ExecuteScalarInternalAsync(CancellationToken cancellationToken)

{

Prechecks();

Log.Trace("ExecuteNonScalar", Connection.Connector.Id);

using (Connection.Connector.StartUserAction())

{

var behavior = CommandBehavior.SequentialAccess | CommandBehavior.SingleRow;

using (var reader = Execute(behavior))

{

return await (reader.ReadAsync(cancellationToken)) && reader.FieldCount != 0 ? reader.GetValue(0) : null;

}

}

}

async Task<NpgsqlDataReader> ExecuteDbDataReaderInternalAsync(CommandBehavior behavior, CancellationToken cancellationToken)

{

Prechecks();

Log.Trace("ExecuteReader", Connection.Connector.Id);

Connection.Connector.StartUserAction();

try

{

return await ExecuteAsync(cancellationToken, behavior);

}

catch

{

Connection.Connector?.EndUserAction();

// Close connection if requested even when there is an error.

if ((behavior & CommandBehavior.CloseConnection) == CommandBehavior.CloseConnection)

{

_connection.Close();

}

throw;

}

}

}

public sealed partial class NpgsqlConnection

{

async Task OpenInternalAsync(CancellationToken cancellationToken)

{

if (string.IsNullOrWhiteSpace(Host))

throw new ArgumentException("Host can't be null");

if (string.IsNullOrWhiteSpace(UserName) && !IntegratedSecurity)

throw new ArgumentException("Either Username must be specified or IntegratedSecurity must be on");

Contract.EndContractBlock();

var timeout = new NpgsqlTimeout(TimeSpan.FromSeconds(ConnectionTimeout));

// If we're postponing a close (see doc on this variable), the connection is already

// open and can be silently reused

if (_postponingClose)

return;

CheckConnectionClosed();

Log.Trace("Opening connnection");

// Copy the password aside and remove it from the user-provided connection string

// (unless PersistSecurityInfo has been requested). Note that cloned connections already

// have Password populated and should not be overwritten.

if (Password == null)

{

Password = Settings.Password;

}

if (!Settings.PersistSecurityInfo)

{

Settings.Password = null;

_connectionString = Settings.ToString();

}

_wasBroken = false;

try

{

// Get a Connector, either from the pool or creating one ourselves.

if (Settings.Pooling)

{

Connector = await (PoolManager.GetOrAdd(Settings).AllocateAsync(this, timeout, cancellationToken));

// Since this pooled connector was opened, global enum/composite mappings may have

// changed. Bring this up to date if needed.

Connector.TypeHandlerRegistry.ActivateGlobalMappings();

}

else

{

Connector = new NpgsqlConnector(this);

await Connector.OpenAsync(timeout, cancellationToken);

}

Connector.Notice += _noticeDelegate;

Connector.Notification += _notificationDelegate;

#if NET45 || NET451 || DNX451

if (Settings.Enlist)

{

Promotable.Enlist(Transaction.Current);

}

#endif

}

catch

{

Connector = null;

throw;

}

OpenCounter++;

OnStateChange(new StateChangeEventArgs(ConnectionState.Closed, ConnectionState.Open));

}

}

/// <summary>

/// Represents a connection to a PostgreSQL backend. Unlike NpgsqlConnection objects, which are

/// exposed to users, connectors are internal to Npgsql and are recycled by the connection pool.

/// </summary>

internal partial class NpgsqlConnector

{

internal async Task OpenAsync(NpgsqlTimeout timeout, CancellationToken cancellationToken)

{

Contract.Requires(Connection != null && Connection.Connector == this);

Contract.Requires(State == ConnectorState.Closed);

State = ConnectorState.Connecting;

try

{

await RawOpenAsync(timeout, cancellationToken);

WriteStartupMessage();

await WriteBuffer.FlushAsync(cancellationToken);

timeout.Check();

await HandleAuthenticationAsync(timeout, cancellationToken);

await TypeHandlerRegistry.SetupAsync(this, timeout, cancellationToken);

Log.Debug($"Opened connection to {Host}:{Port}", Id);

}

catch

{

Break();

throw;

}

}

async Task RawOpenAsync(NpgsqlTimeout timeout, CancellationToken cancellationToken)

{

try

{

await ConnectAsync(timeout, cancellationToken);

Contract.Assert(_socket != null);

_baseStream = new NetworkStream(_socket, true);

Stream = _baseStream;

ReadBuffer = new ReadBuffer(Stream, BufferSize, PGUtil.UTF8Encoding);

WriteBuffer = new WriteBuffer(Stream, BufferSize, PGUtil.UTF8Encoding);

if (SslMode == SslMode.Require || SslMode == SslMode.Prefer)

{

Log.Trace("Attempting SSL negotiation");

SSLRequestMessage.Instance.WriteFully(WriteBuffer);

await WriteBuffer.FlushAsync(cancellationToken);

await ReadBuffer.EnsureAsync(1, cancellationToken);

var response = (char)ReadBuffer.ReadByte();

timeout.Check();

switch (response)

{

default:

throw new Exception($"Received unknown response {response} for SSLRequest (expecting S or N)");

case 'N':

if (SslMode == SslMode.Require)

{

throw new InvalidOperationException("SSL connection requested. No SSL enabled connection from this host is configured.");

}

break;

case 'S':

var clientCertificates = new X509CertificateCollection();

Connection.ProvideClientCertificatesCallback?.Invoke(clientCertificates);

RemoteCertificateValidationCallback certificateValidationCallback;

if (_settings.TrustServerCertificate)

{

certificateValidationCallback = (sender, certificate, chain, errors) => true;

}

else if (Connection.UserCertificateValidationCallback != null)

{

certificateValidationCallback = Connection.UserCertificateValidationCallback;

}

else

{

certificateValidationCallback = DefaultUserCertificateValidationCallback;

}

if (!UseSslStream)

{

var sslStream = new TlsClientStream.TlsClientStream(Stream);

sslStream.PerformInitialHandshake(Host, clientCertificates, certificateValidationCallback, false);

Stream = sslStream;

}

else

{

var sslStream = new SslStream(Stream, false, certificateValidationCallback);

#if NETSTANDARD1_3

// CoreCLR removed sync methods from SslStream, see https://github.com/dotnet/corefx/pull/4868.

// Consider exactly what to do here.

sslStream.AuthenticateAsClientAsync(Host, clientCertificates, SslProtocols.Tls | SslProtocols.Tls11 | SslProtocols.Tls12, false).Wait();

#else

sslStream.AuthenticateAsClient(Host, clientCertificates, SslProtocols.Tls | SslProtocols.Tls11 | SslProtocols.Tls12, false);

#endif

Stream = sslStream;

}

timeout.Check();

ReadBuffer.Underlying = Stream;

WriteBuffer.Underlying = Stream;

IsSecure = true;

Log.Trace("SSL negotiation successful");

break;

}

}

Log.Trace($"Socket connected to {Host}:{Port}");

}

catch

{

if (Stream != null)

{

try

{

Stream.Dispose();

}

catch

{

// ignored

}

Stream = null;

}

if (_baseStream != null)

{

try

{

_baseStream.Dispose();

}

catch

{

// ignored

}

_baseStream = null;

}

if (_socket != null)

{

try

{

_socket.Dispose();

}

catch

{

// ignored

}

_socket = null;

}

throw;

}

}

async Task HandleAuthenticationAsync(NpgsqlTimeout timeout, CancellationToken cancellationToken)

{

Log.Trace("Authenticating...", Id);

while (true)

{

var msg = await (ReadSingleMessageAsync(DataRowLoadingMode.NonSequential, cancellationToken));

timeout.Check();

switch (msg.Code)

{

case BackendMessageCode.AuthenticationRequest:

var passwordMessage = ProcessAuthenticationMessage((AuthenticationRequestMessage)msg);

if (passwordMessage != null)

{

passwordMessage.WriteFully(WriteBuffer);

await WriteBuffer.FlushAsync(cancellationToken);

timeout.Check();

}

continue;

case BackendMessageCode.BackendKeyData:

var backendKeyDataMsg = (BackendKeyDataMessage)msg;

BackendProcessId = backendKeyDataMsg.BackendProcessId;

_backendSecretKey = backendKeyDataMsg.BackendSecretKey;

continue;

case BackendMessageCode.ReadyForQuery:

State = ConnectorState.Ready;

return;

default:

throw new Exception("Unexpected message received while authenticating: " + msg.Code);

}

}

}

internal async Task SendSingleMessageAsync(FrontendMessage msg, CancellationToken cancellationToken)

{

Log.Trace($"Sending: {msg}", Id);

while (true)

{

var completed = msg.Write(WriteBuffer);

await SendBufferAsync(cancellationToken);

if (completed)

break; // Sent all messages

}

}

internal async Task SendBufferAsync(CancellationToken cancellationToken)

{

try

{

await WriteBuffer.FlushAsync(cancellationToken);

}

catch

{

Break();

throw;

}

}

async Task<IBackendMessage> ReadSingleMessageWithPrependedAsync(CancellationToken cancellationToken, DataRowLoadingMode dataRowLoadingMode = DataRowLoadingMode.NonSequential)

{

// First read the responses of any prepended messages.

// Exceptions shouldn't happen here, we break the connector if they do

if (_pendingRfqPrependedMessages > 0)

{

try

{

ReceiveTimeout = InternalCommandTimeout;

while (_pendingRfqPrependedMessages > 0)

{

var msg = await (DoReadSingleMessageAsync(cancellationToken, DataRowLoadingMode.Skip, isPrependedMessage: true));

if (msg is ReadyForQueryMessage)

{

_pendingRfqPrependedMessages--;

}

}

}

catch

{

Break();

throw;

}

}

// Now read a non-prepended message

try

{

ReceiveTimeout = UserTimeout;

return await DoReadSingleMessageAsync(cancellationToken, dataRowLoadingMode);

}

catch (NpgsqlException)

{

if (CurrentReader != null)

{

// The reader cleanup will call EndUserAction

CurrentReader.Cleanup();

}

else

{

EndUserAction();

}

throw;

}

catch

{

Break();

throw;

}

}

async Task<IBackendMessage> DoReadSingleMessageAsync(CancellationToken cancellationToken, DataRowLoadingMode dataRowLoadingMode = DataRowLoadingMode.NonSequential, bool returnNullForAsyncMessage = false, bool isPrependedMessage = false)

{

NpgsqlException error = null;

while (true)

{

var buf = ReadBuffer;

await ReadBuffer.EnsureAsync(5, cancellationToken);

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

Contract.Assume(Enum.IsDefined(typeof (BackendMessageCode), messageCode), "Unknown message code: " + messageCode);

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

if ((messageCode == BackendMessageCode.DataRow && dataRowLoadingMode != DataRowLoadingMode.NonSequential) || messageCode == BackendMessageCode.CopyData)

{

if (dataRowLoadingMode == DataRowLoadingMode.Skip)

{

await ReadBuffer.SkipAsync(len, cancellationToken);

continue;

}

}

else if (len > ReadBuffer.ReadBytesLeft)

{

buf = await (buf.EnsureOrAllocateTempAsync(len, cancellationToken));

}

var msg = ParseServerMessage(buf, messageCode, len, dataRowLoadingMode, isPrependedMessage);

switch (messageCode)

{

case BackendMessageCode.ErrorResponse:

Contract.Assert(msg == null);

// An ErrorResponse is (almost) always followed by a ReadyForQuery. Save the error

// and throw it as an exception when the ReadyForQuery is received (next).

error = new NpgsqlException(buf);

if (State == ConnectorState.Connecting)

{

// During the startup/authentication phase, an ErrorResponse isn't followed by

// an RFQ. Instead, the server closes the connection immediately

throw error;

}

continue;

case BackendMessageCode.ReadyForQuery:

if (error != null)

{

throw error;

}

break;

// Asynchronous messages

case BackendMessageCode.NoticeResponse:

case BackendMessageCode.NotificationResponse:

case BackendMessageCode.ParameterStatus:

Contract.Assert(msg == null);

if (!returnNullForAsyncMessage)

{

continue;

}

return null;

}

Contract.Assert(msg != null, "Message is null for code: " + messageCode);

return msg;

}

}

internal async Task<IBackendMessage> SkipUntilAsync(BackendMessageCode stopAt, CancellationToken cancellationToken)

{

Contract.Requires(stopAt != BackendMessageCode.DataRow, "Shouldn't be used for rows, doesn't know about sequential");

while (true)

{

var msg = await (ReadSingleMessageAsync(DataRowLoadingMode.Skip, cancellationToken));

Contract.Assert(!(msg is DataRowMessage));

if (msg.Code == stopAt)

{

return msg;

}

}

}

internal async Task<IBackendMessage> SkipUntilAsync(BackendMessageCode stopAt1, BackendMessageCode stopAt2, CancellationToken cancellationToken)

{

Contract.Requires(stopAt1 != BackendMessageCode.DataRow, "Shouldn't be used for rows, doesn't know about sequential");

Contract.Requires(stopAt2 != BackendMessageCode.DataRow, "Shouldn't be used for rows, doesn't know about sequential");

while (true)

{

var msg = await (ReadSingleMessageAsync(DataRowLoadingMode.Skip, cancellationToken));

Contract.Assert(!(msg is DataRowMessage));

if (msg.Code == stopAt1 || msg.Code == stopAt2)

{

return msg;

}

}

}

internal async Task<T> ReadExpectingAsync<T>(CancellationToken cancellationToken)where T : class, IBackendMessage

{

var msg = await (ReadSingleMessageAsync(DataRowLoadingMode.NonSequential, cancellationToken));

var asExpected = msg as T;

if (asExpected == null)

{

Break();

throw new Exception($"Received backend message {msg.Code} while expecting {typeof (T).Name}. Please file a bug.");

}

return asExpected;

}

internal async Task RollbackAsync(CancellationToken cancellationToken)

{

Log.Debug("Rollback transaction", Id);

await ExecuteInternalCommandAsync(PregeneratedMessage.RollbackTransaction, cancellationToken);

}

internal async Task ExecuteInternalCommandAsync(FrontendMessage message, CancellationToken cancellationToken)

{

Contract.Requires(message is QueryMessage || message is PregeneratedMessage);

using (StartUserAction())

{

await SendSingleMessageAsync(message, cancellationToken);

await ReadExpectingAsync<CommandCompleteMessage>(cancellationToken);

await ReadExpectingAsync<ReadyForQueryMessage>(cancellationToken);

}

}

}

/// <summary>

/// Reads a forward-only stream of rows from a data source.

/// </summary>

public partial class NpgsqlDataReader

{

async Task<bool> ReadInternalAsync(CancellationToken cancellationToken)

{

if (_row != null)

{

await _row.ConsumeAsync(cancellationToken);

_row = null;

}

switch (_state)

{

case ReaderState.InResult:

break;

case ReaderState.BetweenResults:

case ReaderState.Consumed:

case ReaderState.Closed:

return false;

default:

throw new ArgumentOutOfRangeException();

}

try

{

if ((_behavior & CommandBehavior.SingleRow) != 0 && _readOneRow)

{

await ConsumeAsync(cancellationToken);

return false;

}

while (true)

{

var msg = await (ReadMessageAsync(cancellationToken));

switch (ProcessMessage(msg))

{

case ReadResult.RowRead:

return true;

case ReadResult.RowNotRead:

return false;

case ReadResult.ReadAgain:

continue;

default:

throw new ArgumentOutOfRangeException();

}

}

}

catch (NpgsqlException)

{

_state = ReaderState.Consumed;

throw;

}

}

async Task<bool> NextResultInternalAsync(CancellationToken cancellationToken)

{

Contract.Requires(!IsSchemaOnly);

// Contract.Ensures(Command.CommandType != CommandType.StoredProcedure || Contract.Result<bool>() == false);

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

switch (_state)

{

case ReaderState.InResult:

if (_row != null)

{

await _row.ConsumeAsync(cancellationToken);

_row = null;

}

// TODO: Duplication with SingleResult handling above

var completedMsg = await (SkipUntilAsync(BackendMessageCode.CompletedResponse, BackendMessageCode.EmptyQueryResponse, cancellationToken));

ProcessMessage(completedMsg);

break;

case ReaderState.BetweenResults:

break;

case ReaderState.Consumed:

case ReaderState.Closed:

return false;

default:

throw new ArgumentOutOfRangeException();

}

Contract.Assert(_state == ReaderState.BetweenResults);

_hasRows = null;

#if NET45 || NET451 || DNX451

_cachedSchemaTable = null;

#endif

if ((_behavior & CommandBehavior.SingleResult) != 0 && _statementIndex == 0)

{

if (_state == ReaderState.BetweenResults)

await ConsumeAsync(cancellationToken);

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++)

{

if (IsPrepared)

{

await _connector.ReadExpectingAsync<BindCompleteMessage>(cancellationToken);

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

// Prepare phase

_rowDescription = _statements[_statementIndex].Description;

}

else // Non-prepared flow

{

await _connector.ReadExpectingAsync<ParseCompleteMessage>(cancellationToken);

await _connector.ReadExpectingAsync<BindCompleteMessage>(cancellationToken);

var msg = await (_connector.ReadSingleMessageAsync(DataRowLoadingMode.NonSequential, cancellationToken));

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;

break;

default:

throw _connector.UnexpectedMessageReceived(msg.Code);

}

}

if (_rowDescription == null)

{

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

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

var msg = await (_connector.ReadSingleMessageAsync(DataRowLoadingMode.NonSequential, cancellationToken));

if (msg.Code != BackendMessageCode.CompletedResponse && msg.Code != BackendMessageCode.EmptyQueryResponse)

throw _connector.UnexpectedMessageReceived(msg.Code);

ProcessMessage(msg);

continue;

}

// We got a new resultset.

// Read the next message and store it in _pendingRow, this is to make sure that if the

// statement generated an error, it gets thrown here and not on the first call to Read().

if (_statementIndex == 0 && Command.Parameters.Any(p => p.IsOutputDirection))

{

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

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

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

_pendingMessage = await (_connector.ReadSingleMessageAsync(DataRowLoadingMode.NonSequential, cancellationToken));

PopulateOutputParameters();

}

else

_pendingMessage = await (_connector.ReadSingleMessageAsync(IsSequential ? DataRowLoadingMode.Sequential : DataRowLoadingMode.NonSequential, cancellationToken));

_state = ReaderState.InResult;

return true;

}

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

ProcessMessage(_connector.ReadExpecting<ReadyForQueryMessage>());

_rowDescription = null;

return false;

}

async Task<bool> NextResultSchemaOnlyAsync(CancellationToken cancellationToken)

{

Contract.Requires(IsSchemaOnly);

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

{

if (IsPrepared)

{

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

// Prepare phase

_rowDescription = _statements[_statementIndex].Description;

}

else

{

await _connector.ReadExpectingAsync<ParseCompleteMessage>(cancellationToken);

await _connector.ReadExpectingAsync<ParameterDescriptionMessage>(cancellationToken);

var msg = await (_connector.ReadSingleMessageAsync(DataRowLoadingMode.NonSequential, cancellationToken));

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;

break;

default:

throw _connector.UnexpectedMessageReceived(msg.Code);

}

}

if (_rowDescription != null)

{

// Found a resultset

return true;

}

}

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

if (!IsPrepared)

{

ProcessMessage(_connector.ReadExpecting<ReadyForQueryMessage>());

_rowDescription = null;

}

return false;

}

async Task<IBackendMessage> ReadMessageAsync(CancellationToken cancellationToken)

{

if (_pendingMessage != null)

{

var msg = _pendingMessage;