#region License

// The PostgreSQL License

//

// Copyright (C) 2016 The Npgsql Development Team

//

// Permission to use, copy, modify, and distribute this software and its

// documentation for any purpose, without fee, and without a written

// agreement is hereby granted, provided that the above copyright notice

// and this paragraph and the following two paragraphs appear in all copies.

//

// IN NO EVENT SHALL THE NPGSQL DEVELOPMENT TEAM BE LIABLE TO ANY PARTY

// FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES,

// INCLUDING LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS

// DOCUMENTATION, EVEN IF THE NPGSQL DEVELOPMENT TEAM HAS BEEN ADVISED OF

// THE POSSIBILITY OF SUCH DAMAGE.

//

// THE NPGSQL DEVELOPMENT TEAM SPECIFICALLY DISCLAIMS ANY WARRANTIES,

// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY

// AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS

// ON AN "AS IS" BASIS, AND THE NPGSQL DEVELOPMENT TEAM HAS NO OBLIGATIONS

// TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.

#endregion

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

using JetBrains.Annotations;

using Npgsql.BackendMessages;

using Npgsql.FrontendMessages;

using Npgsql.Logging;

namespace Npgsql

{

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

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

internal Stream Stream { get; private set; }

readonly NpgsqlConnectionStringBuilder _settings;

/// <summary>

/// Contains the clear text password which was extracted from the user-provided connection string.

/// If non-cleartext authentication is requested from the server, this is set to null.

/// </summary>

readonly string _password;

/// <summary>

/// Buffer used for reading data.

/// </summary>

internal ReadBuffer ReadBuffer { get; private set; }

/// <summary>

/// Buffer used for writing data.

/// </summary>

internal WriteBuffer WriteBuffer { get; private set; }

/// <summary>

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

/// </summary>

internal Version ServerVersion { get; private 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 => 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 { 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;

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 readonly Dictionary<string, string> BackendParams;

#if NET45 || NET451 || DNX451

SSPIHandler _sspi;

#endif

/// <summary>

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

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

/// </summary>

internal int UserTimeout { private get; set; }

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

/// <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 connection keepalive is in progress. Used to make sure

/// keepalives and user actions don't interfere with one another.

/// </summary>

SemaphoreSlim _keepAliveLock;

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

internal object CancelLock { get; }

readonly UserAction _userAction;

readonly Timer _keepAliveTimer;

/// <summary>

/// If pooled, the timestamp when this connector was returned to the pool.

/// </summary>

internal DateTime ReleaseTimestamp { get; set; } = DateTime.MaxValue;

internal int ClearCounter { get; set; }

static readonly NpgsqlLogger Log = NpgsqlLogManager.GetCurrentClassLogger();

#endregion

#region Constants

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

internal readonly ParseMessage ParseMessage = new ParseMessage();

internal readonly BindMessage BindMessage = new BindMessage();

internal readonly DescribeMessage DescribeMessage = new DescribeMessage();

internal readonly ExecuteMessage ExecuteMessage = new ExecuteMessage();

internal readonly QueryMessage QueryMessage = new QueryMessage();

// 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.Password)

{

Connection = connection;

Connection.Connector = this;

}

/// <summary>

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

/// </summary>

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

/// <param name="password">The clear-text password or null if not using a password.</param>

NpgsqlConnector(NpgsqlConnectionStringBuilder connectionString, string password)

{

State = ConnectorState.Closed;

TransactionStatus = TransactionStatus.Idle;

_settings = connectionString;

_password = password;

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

_preparedStatementIndex = 0;

_userLock = new SemaphoreSlim(1, 1);

_userAction = new UserAction(this);

CancelLock = new object();

if (IsKeepAliveEnabled) {

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

_keepAliveLock = new SemaphoreSlim(1, 1);

}

}

#endregion

#region Configuration settings

internal string ConnectionString => _settings.ConnectionString;

string Host => _settings.Host;

int Port => _settings.Port;

string Database => _settings.Database;

string Username => _settings.Username;

string KerberosServiceName => _settings.KerberosServiceName;

SslMode SslMode => _settings.SslMode;

bool UseSslStream => _settings.UseSslStream;

int BufferSize => _settings.BufferSize;

int ConnectionTimeout => _settings.Timeout;

int KeepAlive => _settings.KeepAlive;

bool IsKeepAliveEnabled => KeepAlive > 0;

bool IntegratedSecurity => _settings.IntegratedSecurity;

internal bool ConvertInfinityDateTime => _settings.ConvertInfinityDateTime;

int InternalCommandTimeout

{

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 * 1000;

}

}

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

bool IsConnected

{

get

{

switch (State)

{

case ConnectorState.Ready:

case ConnectorState.Executing:

case ConnectorState.Fetching:

case ConnectorState.Waiting:

case ConnectorState.Copy:

return true;

case ConnectorState.Closed:

case ConnectorState.Connecting:

case ConnectorState.Broken:

return false;

default:

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

}

}

}

internal bool IsReady => State == ConnectorState.Ready;

internal bool IsClosed => State == ConnectorState.Closed;

internal bool IsBroken => State == ConnectorState.Broken;

#endregion

#region Open

/// <summary>

/// Totally temporary until the connection pool is rewritten with timeout support

/// </summary>

internal void Open()

{

Open(new NpgsqlTimeout(TimeSpan.Zero));

}

/// <summary>

/// Opens the physical connection to the server.

/// </summary>

/// <remarks>Usually called by the RequestConnector

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

[RewriteAsync]

internal void Open(NpgsqlTimeout timeout)

{

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

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

State = ConnectorState.Connecting;

try {

RawOpen(timeout);

WriteStartupMessage();

WriteBuffer.Flush();

timeout.Check();

HandleAuthentication(timeout);

TypeHandlerRegistry.Setup(this, timeout);

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

}

catch

{

Break();

throw;

}

}

void WriteStartupMessage()

{

var startupMessage = new StartupMessage

{

["client_encoding"] =

_settings.ClientEncoding ??

Environment.GetEnvironmentVariable("PGCLIENTENCODING") ??

"UTF8"

};

if (!string.IsNullOrEmpty(Username))

startupMessage["user"] = Username;

#if NET45 || NET451 || DNX451

else if (IntegratedSecurity)

startupMessage["user"] = UsernameProvider.GetIntegratedUserName(_settings.IncludeRealm);

#endif

if (!string.IsNullOrEmpty(Database))

startupMessage["database"] = Database;

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

startupMessage["application_name"] = _settings.ApplicationName;

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

startupMessage["search_path"] = _settings.SearchPath;

if (IsSecure && !IsRedshift)

startupMessage["ssl_renegotiation_limit"] = "0";

// Should really never happen, just in case

if (startupMessage.Length > WriteBuffer.Size)

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

startupMessage.WriteFully(WriteBuffer);

}

[RewriteAsync]

void RawOpen(NpgsqlTimeout timeout)

{

try

{

Connect(timeout);

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

WriteBuffer.Flush();

ReadBuffer.Ensure(1);

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;

}

}

void Connect(NpgsqlTimeout timeout)

{

#if NET45 || NET451 || DNX451

// Note that there aren't any timeoutable DNS methods, and we want to use sync-only

// methods (not to rely on any TP threads etc.)

var ips = Dns.GetHostAddresses(Host);

#else

// .NET Core doesn't appear to have sync DNS methods (yet?)

var ips = Dns.GetHostAddressesAsync(Host).Result;

#endif

timeout.Check();

// Give each IP an equal share of the remaining time

var perIpTimeout = timeout.IsSet ? (int)((timeout.TimeLeft.Ticks / ips.Length) / 10) : -1;

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

{

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

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

var socket = new Socket(ep.AddressFamily, SocketType.Stream, ProtocolType.Tcp)

{

Blocking = false

};

try

{

try

{

socket.Connect(ep);

}

catch (SocketException e)

{

if (e.SocketErrorCode != SocketError.WouldBlock)

{

throw;

}

}

var write = new List<Socket> { socket };

var error = new List<Socket> { socket };

Socket.Select(null, write, error, perIpTimeout);

var errorCode = (int) socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Error);

if (errorCode != 0) {

throw new SocketException((int)socket.GetSocketOption(SocketOptionLevel.Socket, SocketOptionName.Error));

}

if (!write.Any())

{

Log.Warn(

$"Timeout after {new TimeSpan(perIpTimeout*10).TotalSeconds} seconds when connecting to {ips[i]}");

try { socket.Dispose(); }

catch

{

// ignored

}

if (i == ips.Length - 1)

{

throw new TimeoutException();

}

continue;

}

socket.Blocking = true;

_socket = socket;

return;

}

catch (TimeoutException) { throw; }

catch

{

try { socket.Dispose(); }

catch

{

// ignored

}

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

if (i == ips.Length - 1)

{

throw;

}

}

}

}

async Task ConnectAsync(NpgsqlTimeout timeout, CancellationToken cancellationToken)

{

// Note that there aren't any timeoutable or cancellable DNS methods

var ips = await Dns.GetHostAddressesAsync(Host).WithCancellation(cancellationToken);

// Give each IP an equal share of the remaining time

var perIpTimespan = timeout.IsSet ? new TimeSpan(timeout.TimeLeft.Ticks / ips.Length) : TimeSpan.Zero;

var perIpTimeout = timeout.IsSet ? new NpgsqlTimeout(perIpTimespan) : timeout;

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

{

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

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

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

#if NETSTANDARD1_3

var connectTask = socket.ConnectAsync(ep);

#else

var connectTask = Task.Factory.FromAsync(socket.BeginConnect, socket.EndConnect, ep, null);

#endif

try

{

try

{

await connectTask.WithCancellationAndTimeout(perIpTimeout, cancellationToken);

}

catch (OperationCanceledException)

{

#pragma warning disable 4014

// ReSharper disable once MethodSupportsCancellation

connectTask.ContinueWith(t => socket.Dispose());

#pragma warning restore 4014

if (timeout.HasExpired)

{

Log.Warn($"Timeout after {perIpTimespan.TotalSeconds} seconds when connecting to {ips[i]}");

if (i == ips.Length - 1)

{

throw new TimeoutException();

}

continue;

}

// We're here if an actual cancellation was requested (not a timeout)

throw;

}

_socket = socket;

return;

}

catch (TimeoutException) { throw; }

catch (OperationCanceledException) { throw; }

catch

{

try { socket.Dispose(); }

catch

{

// ignored

}

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

if (i == ips.Length - 1)

{

throw;

}

}

}

}

[RewriteAsync]

void HandleAuthentication(NpgsqlTimeout timeout)

{

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

while (true)

{

var msg = ReadSingleMessage(DataRowLoadingMode.NonSequential);

timeout.Check();

switch (msg.Code)

{

case BackendMessageCode.AuthenticationRequest:

var passwordMessage = ProcessAuthenticationMessage((AuthenticationRequestMessage)msg);

if (passwordMessage != null)

{

passwordMessage.WriteFully(WriteBuffer);

WriteBuffer.Flush();

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

}

}

}

/// <summary>

/// Performs a step in the PostgreSQL authentication protocol

/// </summary>

/// <param name="msg">A message read from the server, instructing us on the required response</param>

/// <returns>a PasswordMessage to be sent, or null if authentication has completed successfully</returns>

[CanBeNull]

PasswordMessage ProcessAuthenticationMessage(AuthenticationRequestMessage msg)

{

switch (msg.AuthRequestType)

{

case AuthenticationRequestType.AuthenticationOk:

return null;

case AuthenticationRequestType.AuthenticationCleartextPassword:

if (_password == null) {

throw new Exception("No password has been provided but the backend requires one (in cleartext)");

}

return PasswordMessage.CreateClearText(_password);

case AuthenticationRequestType.AuthenticationMD5Password:

if (_password == null) {

throw new Exception("No password has been provided but the backend requires one (in MD5)");

}

return PasswordMessage.CreateMD5(_password, Username, ((AuthenticationMD5PasswordMessage)msg).Salt);

case AuthenticationRequestType.AuthenticationGSS:

if (!IntegratedSecurity) {

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

}

#if NET45 || NET451 || DNX451

// For GSSAPI we have to use the supplied hostname

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

return new PasswordMessage(_sspi.Continue(null));

#else

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

#endif

case AuthenticationRequestType.AuthenticationSSPI:

if (!IntegratedSecurity) {

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

}

#if NET45 || NET451 || DNX451

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

return new PasswordMessage(_sspi.Continue(null));

#else

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

#endif

case AuthenticationRequestType.AuthenticationGSSContinue:

#if NET45 || NET451 || DNX451

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

if (passwdRead.Length != 0)

{

return new PasswordMessage(passwdRead);

}

return null;

#else

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

#endif

default:

throw new NotSupportedException($"Authentication method not supported (Received: {msg.AuthRequestType})");

}

}

#endregion

#region Frontend message processing

/// <summary>

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

/// </summary>

internal void PrependInternalMessage(FrontendMessage msg)

{

Contract.Requires(msg is PregeneratedMessage);

// Prepended messages are simple queries (pregenerated, which produce a ReadyForQuery response,

// which we will be looking for as we're reading the results

_pendingRfqPrependedMessages++;

if (!msg.Write(WriteBuffer))

throw new Exception($"Could not fully write message of type {msg.GetType().Name} into the buffer");

}

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

[RewriteAsync]

internal void SendSingleMessage(FrontendMessage msg)

{

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

while (true)

{

var completed = msg.Write(WriteBuffer);

SendBuffer();

if (completed)

break; // Sent all messages

}

}

internal void SendSingleQuery(string query) => SendSingleMessage(QueryMessage.Populate(query));

[RewriteAsync]

internal void SendBuffer()

{

try

{

WriteBuffer.Flush();

}

catch

{

Break();

throw;

}

}

#endregion

#region Backend message processing

internal IBackendMessage ReadSingleMessage(DataRowLoadingMode dataRowLoadingMode)

{

var msg = ReadSingleMessageWithPrepended(dataRowLoadingMode);

Contract.Assert(msg != null);

return msg;

}

internal Task<IBackendMessage> ReadSingleMessageAsync(DataRowLoadingMode dataRowLoadingMode, CancellationToken cancellationToken)

=> ReadSingleMessageWithPrependedAsync(cancellationToken, dataRowLoadingMode);

internal void ReadAsyncMessage(int timeout)

{

ReceiveTimeout = timeout;

var msg = DoReadSingleMessage(DataRowLoadingMode.NonSequential, true);

if (msg != null)

throw new Exception($"While waiting for an asynchronous message, received an unexpected message of type {msg.Code}");

}

internal async Task ReadAsyncMessageAsync()

{

var msg = await DoReadSingleMessageAsync(CancellationToken.None, DataRowLoadingMode.NonSequential, true);

if (msg != null)

throw new Exception($"While waiting for an asynchronous message, received an unexpected message of type {msg.Code}");

}

[RewriteAsync]

[CanBeNull]

[MethodImpl(MethodImplOptions.AggressiveInlining)]

IBackendMessage ReadSingleMessageWithPrepended(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 = DoReadSingleMessage(DataRowLoadingMode.Skip, isPrependedMessage: true);

if (msg is ReadyForQueryMessage)

{

_pendingRfqPrependedMessages--;

}

}

}

catch

{

Break();

throw;

}

}

// Now read a non-prepended message

try

{

ReceiveTimeout = UserTimeout;

return DoReadSingleMessage(dataRowLoadingMode);

}

catch (NpgsqlException)

{

if (CurrentReader != null)

{

// The reader cleanup will call EndUserAction

CurrentReader.Cleanup();

}

else

{

EndUserAction();

}

throw;

}

catch

{

Break();

throw;

}

}

[RewriteAsync]

[CanBeNull]

IBackendMessage DoReadSingleMessage(DataRowLoadingMode dataRowLoadingMode = DataRowLoadingMode.NonSequential,

bool returnNullForAsyncMessage = false,

bool isPrependedMessage = false)

{

NpgsqlException error = null;

while (true)

{

var buf = ReadBuffer;

ReadBuffer.Ensure(5);

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)

{

ReadBuffer.Skip(len);

continue;

}

}

else if (len > ReadBuffer.ReadBytesLeft)

{

buf = buf.EnsureOrAllocateTemp(len);

}

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;

}

}

[CanBeNull]

IBackendMessage ParseServerMessage(ReadBuffer buf, BackendMessageCode code, int len, DataRowLoadingMode dataRowLoadingMode, bool isPrependedMessage)

{

switch (code)

{

case BackendMessageCode.RowDescription:

// TODO: Recycle