/// <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

{

#region Fields and Properties

readonly NpgsqlConnectionStringBuilder _settings;

/// <summary>

/// The physical connection socket to the backend.

/// </summary>

Socket _socket;

/// <summary>

/// The physical connection stream to the backend, without anything on top.

/// </summary>

NetworkStream _baseStream;

/// <summary>

/// The physical connection stream to the backend, layered with an SSL/TLS stream if in secure mode.

/// </summary>

Stream _stream;

/// <summary>

/// Buffer used for reading data.

/// </summary>

internal NpgsqlBuffer Buffer { get; private set; }

/// <summary>

/// Version of backend server this connector is connected to.

/// </summary>

internal Version ServerVersion { get; set; }

/// <summary>

/// The secret key of the backend for this connector, used for query cancellation.

/// </summary>

int _backendSecretKey;

/// <summary>

/// The process ID of the backend for this connector.

/// </summary>

internal int BackendProcessId { get; private set; }

/// <summary>

/// A unique ID identifying this connector, used for logging. Currently mapped to BackendProcessId

/// </summary>

internal int Id { get { return BackendProcessId; } }

internal TypeHandlerRegistry TypeHandlerRegistry { get; set; }

/// <summary>

/// The current transaction status for this connector.

/// </summary>

internal TransactionStatus TransactionStatus { get; set; }

/// <summary>

/// The transaction currently in progress, if any.

/// </summary>

/// <remarks>

/// <para>

/// Note that this doesn't mean a transaction request has actually been sent to the backend - for

/// efficiency we defer sending the request to the first query after BeginTransaction is called.

/// See <see cref="TransactionStatus"/> for the actual transaction status.

/// </para>

/// <para>

/// Also, the user can initiate a transaction in SQL (i.e. BEGIN), in which case there will be no

/// NpgsqlTransaction instance. As a result, never check <see cref="Transaction"/> to know whether

/// a transaction is in progress, check <see cref="TransactionStatus"/> instead.

/// </para>

/// </remarks>

internal NpgsqlTransaction Transaction { private get; set; }

/// <summary>

/// The NpgsqlConnection that (currently) owns this connector. Null if the connector isn't

/// owned (i.e. idle in the pool)

/// </summary>

internal NpgsqlConnection Connection { get; set; }

/// <summary>

/// The number of messages that were prepended to the current message chain, but not yet sent.

/// Note that this only tracks messages which produce a ReadyForQuery message

/// </summary>

byte _pendingRfqPrependedMessages;

/// <summary>

/// The number of messages that were prepended and sent to the last message chain.

/// Note that this only tracks messages which produce a ReadyForQuery message

/// </summary>

byte _sentRfqPrependedMessages;

/// <summary>

/// A chain of messages to be sent to the backend.

/// </summary>

readonly List<FrontendMessage> _messagesToSend;

internal NpgsqlDataReader CurrentReader;

/// <summary>

/// If the connector is currently in COPY mode, holds a reference to the importer/exporter object.

/// Otherwise null.

/// </summary>

internal ICancelable CurrentCopyOperation;

/// <summary>

/// Holds all run-time parameters received from the backend (via ParameterStatus messages)

/// </summary>

internal Dictionary<string, string> BackendParams;

#if !DNXCORE50

internal SSPIHandler SSPI { get; set; }

#endif

/// <summary>

/// The frontend timeout for reading messages that are part of the user's command

/// (i.e. which aren't internal prepended commands).

/// </summary>

internal int UserCommandFrontendTimeout { get; set; }

/// <summary>

/// Contains the current value of the statement_timeout parameter at the backend,

/// used to determine whether we need to change it when commands are sent.

/// </summary>

/// <remarks>

/// 0 means means no timeout.

/// -1 means that the current value is unknown.

/// </remarks>

int _backendTimeout;

/// <summary>

/// Contains the current value of the socket's ReceiveTimeout, used to determine whether

/// we need to change it when commands are received.

/// </summary>

int _frontendTimeout;

/// <summary>

/// A lock that's taken while a user action is in progress, e.g. a command being executed.

/// </summary>

SemaphoreSlim _userLock;

/// <summary>

/// A lock that's taken while a non-user-triggered async action is in progress, e.g. handling of an

/// asynchronous notification or a connection keepalive. Does <b>not</b> get taken for a user async

/// action such as <see cref="DbCommand.ExecuteReaderAsync()"/>.

/// </summary>

SemaphoreSlim _asyncLock;

/// <summary>

/// A lock that's taken while a cancellation is being delivered; new queries are blocked until the

/// cancellation is delivered. This reduces the chance that a cancellation meant for a previous

/// command will accidentally cancel a later one, see #615.

/// </summary>

readonly object _cancelLock;

readonly UserAction _userAction;

readonly Timer _keepAliveTimer;

static readonly byte[] EmptyBuffer = new byte[0];

static readonly NpgsqlLogger Log = NpgsqlLogManager.GetCurrentClassLogger();

#endregion

#region Constants

/// <summary>

/// Number of seconds added as a margin to the backend timeout to yield the frontend timeout.

/// We prefer the backend to timeout - it's a clean error which doesn't break the connector.

/// </summary>

const int FrontendTimeoutMargin = 3;

/// <summary>

/// The minimum timeout that can be set on internal commands such as COMMIT, ROLLBACK.

/// </summary>

internal const int MinimumInternalCommandTimeout = 3;

#endregion

#region Reusable Message Objects

// Frontend. Note that these are only used for single-query commands.

internal readonly ParseMessage ParseMessage = new ParseMessage();

internal readonly BindMessage BindMessage = new BindMessage();

internal readonly DescribeMessage DescribeMessage = new DescribeMessage();

internal readonly ExecuteMessage ExecuteMessage = new ExecuteMessage();

// Backend

readonly CommandCompleteMessage _commandCompleteMessage = new CommandCompleteMessage();

readonly ReadyForQueryMessage _readyForQueryMessage = new ReadyForQueryMessage();

readonly ParameterDescriptionMessage _parameterDescriptionMessage = new ParameterDescriptionMessage();

readonly DataRowSequentialMessage _dataRowSequentialMessage = new DataRowSequentialMessage();

readonly DataRowNonSequentialMessage _dataRowNonSequentialMessage = new DataRowNonSequentialMessage();

// Since COPY is rarely used, allocate these lazily

CopyInResponseMessage _copyInResponseMessage;

CopyOutResponseMessage _copyOutResponseMessage;

CopyDataMessage _copyDataMessage;

#endregion

#region Constructors

internal NpgsqlConnector(NpgsqlConnection connection)

: this(connection.Settings)

{

Connection = connection;

Connection.Connector = this;

}

/// <summary>

/// Creates a new connector with the given connection string.

/// </summary>

/// <param name="connectionString">The connection string.</param>

NpgsqlConnector(NpgsqlConnectionStringBuilder connectionString)

{

State = ConnectorState.Closed;

TransactionStatus = TransactionStatus.Idle;

_settings = connectionString;

BackendParams = new Dictionary<string, string>();

_messagesToSend = new List<FrontendMessage>();

_preparedStatementIndex = 0;

_portalIndex = 0;

_userLock = new SemaphoreSlim(1, 1);

_asyncLock = new SemaphoreSlim(1, 1);

_userAction = new UserAction(this);

_cancelLock = new object();

if (KeepAlive > 0) {

_keepAliveTimer = new Timer(PerformKeepAlive, null, Timeout.Infinite, Timeout.Infinite);

}

}

#endregion

#region Configuration settings

internal string ConnectionString { get { return _settings.ConnectionString; } }

internal string Host { get { return _settings.Host; } }

internal int Port { get { return _settings.Port; } }

internal string Database { get { return _settings.Database; } }

internal string UserName { get { return _settings.Username; } }

internal string Password { get { return _settings.Password; } }

internal string KerberosServiceName { get { return _settings.KerberosServiceName; } }

internal SslMode SslMode { get { return _settings.SslMode; } }

internal bool UseSslStream { get { return _settings.UseSslStream; } }

internal int BufferSize { get { return _settings.BufferSize; } }

internal int ConnectionTimeout { get { return _settings.Timeout; } }

internal bool BackendTimeouts { get { return _settings.BackendTimeouts; } }

internal int KeepAlive { get { return _settings.KeepAlive; } }

internal bool Enlist { get { return _settings.Enlist; } }

internal bool IntegratedSecurity { get { return _settings.IntegratedSecurity; } }

internal bool ConvertInfinityDateTime { get { return _settings.ConvertInfinityDateTime; } }

internal bool ContinuousProcessing { get { return _settings.ContinuousProcessing; } }

internal int ActualInternalCommandTimeout

{

get

{

Contract.Ensures(Contract.Result<int>() == 0 || Contract.Result<int>() >= MinimumInternalCommandTimeout);

var internalTimeout = _settings.InternalCommandTimeout;

if (internalTimeout == -1) {

return Math.Max(_settings.CommandTimeout, MinimumInternalCommandTimeout);

}

Contract.Assert(internalTimeout == 0 || internalTimeout >= MinimumInternalCommandTimeout);

return internalTimeout;

}

}

#endregion Configuration settings

#region State management

int _state;

/// <summary>

/// Gets the current state of the connector

/// </summary>

internal ConnectorState State

{

get { return (ConnectorState)_state; }

set

{

var newState = (int)value;

if (newState == _state)

return;

Interlocked.Exchange(ref _state, newState);

}

}

/// <summary>

/// Returns whether the connector is open, regardless of any task it is currently performing

/// </summary>

internal bool IsConnected

{

get

{

switch (State)

{

case ConnectorState.Ready:

case ConnectorState.Executing:

case ConnectorState.Fetching:

case ConnectorState.Copy:

return true;

case ConnectorState.Closed:

case ConnectorState.Connecting:

case ConnectorState.Broken:

return false;

default:

throw new ArgumentOutOfRangeException("State", "Unknown state: " + State);

}

}

}

/// <summary>

/// Returns whether the connector is open and performing a task, i.e. not ready for a query

/// </summary>

internal bool IsBusy

{

get

{

switch (State)

{

case ConnectorState.Executing:

case ConnectorState.Fetching:

case ConnectorState.Copy:

return true;

case ConnectorState.Ready:

case ConnectorState.Closed:

case ConnectorState.Connecting:

case ConnectorState.Broken:

return false;

default:

throw new ArgumentOutOfRangeException("State", "Unknown state: " + State);

}

}

}

internal bool IsReady { get { return State == ConnectorState.Ready; } }

internal bool IsClosed { get { return State == ConnectorState.Closed; } }

internal bool IsBroken { get { return State == ConnectorState.Broken; } }

#endregion

#region Open

/// <summary>

/// Opens the physical connection to the server.

/// </summary>

/// <remarks>Usually called by the RequestConnector

/// Method of the connection pool manager.</remarks>

internal void Open()

{

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

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

Contract.Ensures(State == ConnectorState.Ready);

State = ConnectorState.Connecting;

ServerVersion = null;

// Keep track of time remaining; Even though there may be multiple timeout-able calls,

// this allows us to still respect the caller's timeout expectation.

var connectTimeRemaining = ConnectionTimeout * 1000;

try {

// Get a raw connection, possibly SSL...

RawOpen(connectTimeRemaining);

var startupMessage = new StartupMessage(Database, UserName);

if (!string.IsNullOrEmpty(_settings.ApplicationName)) {

startupMessage["application_name"] = _settings.ApplicationName;

}

if (!string.IsNullOrEmpty(_settings.SearchPath)) {

startupMessage["search_path"] = _settings.SearchPath;

}

if (_settings.BackendTimeouts && _settings.CommandTimeout != 0) {

startupMessage["statement_timeout"] = (_settings.CommandTimeout * 1000).ToString();

_backendTimeout = _settings.CommandTimeout;

}

if (IsSecure && !IsRedshift) {

startupMessage["ssl_renegotiation_limit"] = "0";

}

if (startupMessage.Length > Buffer.Size) { // Should really never happen, just in case

throw new Exception("Startup message bigger than buffer");

}

startupMessage.Write(Buffer);

Buffer.Flush();

HandleAuthentication();

ProcessServerVersion();

TypeHandlerRegistry.Setup(this);

State = ConnectorState.Ready;

if (ContinuousProcessing) {

HandleAsyncMessages();

}

}

catch

{

try { Break(); }

catch {

// ignored

}

throw;

}

}

public void RawOpen(int timeout)

{

try

{

// Keep track of time remaining; Even though there may be multiple timeout-able calls,

// this allows us to still respect the caller's timeout expectation.

var attemptStart = DateTime.Now;

var result = Dns.BeginGetHostAddresses(Host, null, null);

if (!result.AsyncWaitHandle.WaitOne(timeout, true))

{

// Timeout was used up attempting the Dns lookup

throw new TimeoutException("Dns hostname lookup timeout. Increase Timeout value in ConnectionString.");

}

timeout -= Convert.ToInt32((DateTime.Now - attemptStart).TotalMilliseconds);

var ips = Dns.EndGetHostAddresses(result);

// try every ip address of the given hostname, use the first reachable one

// make sure not to exceed the caller's timeout expectation by splitting the

// time we have left between all the remaining ip's in the list.

for (var i = 0; i < ips.Length; i++)

{

Log.Trace("Attempting to connect to " + ips[i], Id);

var ep = new IPEndPoint(ips[i], Port);

_socket = new Socket(ep.AddressFamily, SocketType.Stream, ProtocolType.Tcp);

attemptStart = DateTime.Now;

try

{

result = _socket.BeginConnect(ep, null, null);

if (!result.AsyncWaitHandle.WaitOne(timeout / (ips.Length - i), true)) {

throw new TimeoutException("Connection establishment timeout. Increase Timeout value in ConnectionString.");

}

_socket.EndConnect(result);

// connect was successful, leave the loop

break;

}

catch (TimeoutException)

{

throw;

}

catch

{

Log.Warn("Failed to connect to " + ips[i]);

timeout -= Convert.ToInt32((DateTime.Now - attemptStart).TotalMilliseconds);

if (i == ips.Length - 1) {

throw;

}

}

}

Contract.Assert(_socket != null);

_baseStream = new NetworkStream(_socket, true);

_stream = _baseStream;

// If the PostgreSQL server has SSL connectors enabled Open SslClientStream if (response == 'S') {

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

{

_stream

.WriteInt32(8)

.WriteInt32(80877103);

// Receive response

var response = (Char)_stream.ReadByte();

if (response != 'S')

{

if (SslMode == SslMode.Require) {

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

}

}

else

{

var clientCertificates = new X509CertificateCollection();

if (ProvideClientCertificatesCallback != null) {

ProvideClientCertificatesCallback(clientCertificates);

}

RemoteCertificateValidationCallback certificateValidationCallback;

if (_settings.TrustServerCertificate) {

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

} else if (UserCertificateValidationCallback != null) {

certificateValidationCallback = UserCertificateValidationCallback;

} else {

certificateValidationCallback = DefaultUserCertificateValidationCallback;

}

if (!UseSslStream)

{

#if DNXCORE50

throw new NotSupportedException("TLS implementation not yet supported with .NET Core");

#else

var sslStream = new TlsClientStream.TlsClientStream(_stream);

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

_stream = sslStream;

#endif

}

else

{

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

sslStream.AuthenticateAsClient(Host, clientCertificates, System.Security.Authentication.SslProtocols.Default, false);

_stream = sslStream;

}

IsSecure = true;

}

}

Buffer = new NpgsqlBuffer(_stream, BufferSize, PGUtil.UTF8Encoding);

Log.Debug(String.Format("Connected to {0}:{1}", 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;

}

}

void HandleAuthentication()

{

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

while (true)

{

var msg = ReadSingleMessage();

switch (msg.Code)

{

case BackendMessageCode.AuthenticationRequest:

ProcessAuthenticationMessage((AuthenticationRequestMessage)msg);

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

}

}

}

void ProcessAuthenticationMessage(AuthenticationRequestMessage msg)

{

PasswordMessage passwordMessage;

switch (msg.AuthRequestType)

{

case AuthenticationRequestType.AuthenticationOk:

return;

case AuthenticationRequestType.AuthenticationCleartextPassword:

passwordMessage = PasswordMessage.CreateClearText(Password);

break;

case AuthenticationRequestType.AuthenticationMD5Password:

passwordMessage = PasswordMessage.CreateMD5(Password, UserName, ((AuthenticationMD5PasswordMessage)msg).Salt);

break;

case AuthenticationRequestType.AuthenticationGSS:

if (!IntegratedSecurity) {

throw new Exception("GSS authentication but IntegratedSecurity not enabled");

}

#if DNXCORE50

throw new NotSupportedException("SSPI not yet supported in .NET Core");

#else

// For GSSAPI we have to use the supplied hostname

SSPI = new SSPIHandler(Host, KerberosServiceName, true);

passwordMessage = new PasswordMessage(SSPI.Continue(null));

break;

#endif

case AuthenticationRequestType.AuthenticationSSPI:

if (!IntegratedSecurity) {

throw new Exception("SSPI authentication but IntegratedSecurity not enabled");

}

#if DNXCORE50

throw new NotSupportedException("SSPI not yet supported in .NET Core");

#else

SSPI = new SSPIHandler(Host, KerberosServiceName, false);

passwordMessage = new PasswordMessage(SSPI.Continue(null));

break;

#endif

case AuthenticationRequestType.AuthenticationGSSContinue:

#if DNXCORE50

throw new NotSupportedException("SSPI not yet supported in .NET Core");

#else

var passwdRead = SSPI.Continue(((AuthenticationGSSContinueMessage)msg).AuthenticationData);

if (passwdRead.Length != 0)

{

passwordMessage = new PasswordMessage(passwdRead);

break;

}

return;

#endif

default:

throw new NotSupportedException(String.Format("Authentication method not supported (Received: {0})", msg.AuthRequestType));

}

passwordMessage.Write(Buffer);

Buffer.Flush();

}

#endregion

#region Frontend message processing

internal void AddMessage(FrontendMessage msg)

{

_messagesToSend.Add(msg);

}

/// <summary>

/// Prepends a message to be sent at the beginning of the next message chain.

/// </summary>

internal void PrependInternalMessage(FrontendMessage msg)

{

// Set backend timeout if needed.

PrependBackendTimeoutMessage(ActualInternalCommandTimeout);

if (msg is QueryMessage || msg is PregeneratedMessage || msg is SyncMessage)

{

// These messages produce a ReadyForQuery response, which we will be looking for when

// processing the message chain results

_pendingRfqPrependedMessages++;

}

_messagesToSend.Add(msg);

}

internal void PrependBackendTimeoutMessage(int timeout)

{

if (_backendTimeout == timeout || !BackendTimeouts) {

return;

}

_backendTimeout = timeout;

_pendingRfqPrependedMessages++;

switch (timeout) {

case 10:

_messagesToSend.Add(PregeneratedMessage.SetStmtTimeout10Sec);

return;

case 20:

_messagesToSend.Add(PregeneratedMessage.SetStmtTimeout20Sec);

return;

case 30:

_messagesToSend.Add(PregeneratedMessage.SetStmtTimeout30Sec);

return;

case 60:

_messagesToSend.Add(PregeneratedMessage.SetStmtTimeout60Sec);

return;

case 90:

_messagesToSend.Add(PregeneratedMessage.SetStmtTimeout90Sec);

return;

case 120:

_messagesToSend.Add(PregeneratedMessage.SetStmtTimeout120Sec);

return;

default:

_messagesToSend.Add(new QueryMessage(string.Format("SET statement_timeout = {0}", timeout * 1000)));

return;

}

}

[RewriteAsync]

internal void SendAllMessages()

{

if (!_messagesToSend.Any()) {

return;

}

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

lock (_cancelLock) { }

_sentRfqPrependedMessages = _pendingRfqPrependedMessages;

_pendingRfqPrependedMessages = 0;

try

{

foreach (var msg in _messagesToSend)

{

SendMessage(msg);

}

Buffer.Flush();

}

catch

{

Break();

throw;

}

finally

{

_messagesToSend.Clear();

}

}

/// <summary>

/// Sends a single frontend message, used for simple messages such as rollback, etc.

/// Note that additional prepend messages may be previously enqueued, and will be sent along

/// with this message.

/// </summary>

/// <param name="msg"></param>

internal void SendSingleMessage(FrontendMessage msg)

{

AddMessage(msg);

SendAllMessages();

}

[RewriteAsync]

void SendMessage(FrontendMessage msg)

{

Log.Trace(String.Format("Sending: {0}", msg), Id);

var asSimple = msg as SimpleFrontendMessage;

if (asSimple != null)

{

if (asSimple.Length > Buffer.WriteSpaceLeft)

{

Buffer.Flush();

}

Contract.Assume(Buffer.WriteSpaceLeft >= asSimple.Length);

asSimple.Write(Buffer);

return;

}

var asComplex = msg as ChunkingFrontendMessage;

if (asComplex != null)

{

var directBuf = new DirectBuffer();

while (!asComplex.Write(Buffer, ref directBuf))

{

Buffer.Flush();

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

// through our buffer

if (directBuf.Buffer != null)

{

Buffer.Underlying.Write(directBuf.Buffer, directBuf.Offset, directBuf.Size == 0 ? directBuf.Buffer.Length : directBuf.Size);

directBuf.Buffer = null;

directBuf.Size = 0;

}

}

return;

}

throw PGUtil.ThrowIfReached();

}

#endregion

#region Backend message processing

[RewriteAsync]

internal IBackendMessage ReadSingleMessage(DataRowLoadingMode dataRowLoadingMode = DataRowLoadingMode.NonSequential, bool returnNullForAsyncMessage = false)

{

// First read the responses of any prepended messages.

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

if (_sentRfqPrependedMessages > 0)

{

try

{

SetFrontendTimeout(ActualInternalCommandTimeout);

while (_sentRfqPrependedMessages > 0)

{

var msg = DoReadSingleMessage(DataRowLoadingMode.Skip);

if (msg is ReadyForQueryMessage)

{

_sentRfqPrependedMessages--;

}

}

}

catch

{

Break();

throw;

}

}

// Now read a non-prepended message

try

{

SetFrontendTimeout(UserCommandFrontendTimeout);

return DoReadSingleMessage(dataRowLoadingMode, returnNullForAsyncMessage);

}

catch (NpgsqlException)

{

if (CurrentReader != null)

{

// The reader cleanup will call EndUserAction

CurrentReader.Cleanup();

}

else

{

EndUserAction();

}

throw;

}

catch

{

Break();

throw;

}

}

[RewriteAsync]

IBackendMessage DoReadSingleMessage(DataRowLoadingMode dataRowLoadingMode = DataRowLoadingMode.NonSequential, bool returnNullForAsyncMessage = false)

{

Contract.Ensures(returnNullForAsyncMessage || Contract.Result<IBackendMessage>() != null);

NpgsqlException error = null;

while (true)

{

var buf = Buffer;

Buffer.Ensure(5);

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

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

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

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

messageCode == BackendMessageCode.CopyData)

{

if (dataRowLoadingMode == DataRowLoadingMode.Skip)

{

Buffer.Skip(len);

continue;

}

}

else if (len > Buffer.ReadBytesLeft)

{

buf = buf.EnsureOrAllocateTemp(len);

}

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

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;

}

}

IBackendMessage ParseServerMessage(NpgsqlBuffer buf, BackendMessageCode code, int len, DataRowLoadingMode dataRowLoadingMode)

{

switch (code)

{

case BackendMessageCode.RowDescription:

// TODO: Recycle

var rowDescriptionMessage = new RowDescriptionMessage();

return rowDescriptionMessage.Load(buf, TypeHandlerRegistry);

case BackendMessageCode.DataRow:

Contract.Assert(dataRowLoadingMode == DataRowLoadingMode.NonSequential || dataRowLoadingMode == DataRowLoadingMode.Sequential);

return dataRowLoadingMode == DataRowLoadingMode.Sequential

? _dataRowSequentialMessage.Load(buf)

: _dataRowNonSequentialMessage.Load(buf);

case BackendMessageCode.CompletedResponse:

return _commandCompleteMessage.Load(buf, len);

case BackendMessageCode.ReadyForQuery:

var rfq = _readyForQueryMessage.Load(buf);

ProcessNewTransactionStatus(rfq.TransactionStatusIndicator);

return rfq;

case BackendMessageCode.EmptyQueryResponse:

return EmptyQueryMessage.Instance;

case BackendMessageCode.ParseComplete:

return ParseCompleteMessage.Instance;

case BackendMessageCode.ParameterDescription:

return _parameterDescriptionMessage.Load(buf);

case BackendMessageCode.BindComplete:

return BindCompleteMessage.Instance;

case BackendMessageCode.NoData:

return NoDataMessage.Instance;

case BackendMessageCode.CloseComplete:

return CloseCompletedMessage.Instance;

case BackendMessageCode.ParameterStatus:

HandleParameterStatus(buf.ReadNullTerminatedString(), buf.ReadNullTerminatedString());

return null;

case BackendMessageCode.NoticeResponse:

FireNotice(new NpgsqlNotice(buf));

return null;

case BackendMessageCode.NotificationResponse:

FireNotification(new NpgsqlNotificationEventArgs(buf));