using System;

using System.Collections.Generic;

using System.Diagnostics;

using System.IO;

using System.Linq;

using System.Net;

using System.Net.Security;

using System.Net.Sockets;

using System.Runtime.CompilerServices;

using System.Runtime.InteropServices;

using System.Runtime.ExceptionServices;

using System.Security.Authentication;

using System.Security.Cryptography.X509Certificates;

using System.Text;

using System.Threading;

using System.Threading.Tasks;

using Npgsql.BackendMessages;

using Npgsql.Logging;

using Npgsql.TypeMapping;

using Npgsql.Util;

using static Npgsql.Util.Statics;

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>

sealed partial class NpgsqlConnector : IDisposable

{

#region Fields and Properties

/// <summary>

/// The physical connection socket to the backend.

/// </summary>

Socket _socket = default!;

/// <summary>

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

/// </summary>

NetworkStream _baseStream = default!;

/// <summary>

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

/// </summary>

Stream _stream = default!;

internal NpgsqlConnectionStringBuilder Settings { get; }

internal string ConnectionString { get; }

ProvideClientCertificatesCallback? ProvideClientCertificatesCallback { get; }

RemoteCertificateValidationCallback? UserCertificateValidationCallback { get; }

ProvidePasswordCallback? ProvidePasswordCallback { get; }

internal Encoding TextEncoding { get; private set; } = default!;

/// <summary>

/// Same as <see cref="TextEncoding"/>, except that it does not throw an exception if an invalid char is

/// encountered (exception fallback), but rather replaces it with a question mark character (replacement

/// fallback).

/// </summary>

internal Encoding RelaxedTextEncoding { get; private set; } = default!;

/// <summary>

/// Buffer used for reading data.

/// </summary>

internal NpgsqlReadBuffer ReadBuffer { get; private set; } = default!;

/// <summary>

/// If we read a data row that's bigger than <see cref="ReadBuffer"/>, we allocate an oversize buffer.

/// The original (smaller) buffer is stored here, and restored when the connection is reset.

/// </summary>

NpgsqlReadBuffer? _origReadBuffer;

/// <summary>

/// Buffer used for writing data.

/// </summary>

internal NpgsqlWriteBuffer WriteBuffer { get; private set; } = default!;

/// <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 NpgsqlDatabaseInfo DatabaseInfo { get; private set; } = default!;

internal ConnectorTypeMapper TypeMapper { get; set; } = default!;

/// <summary>

/// The current transaction status for this connector.

/// </summary>

internal TransactionStatus TransactionStatus { get; set; }

/// <summary>

/// A transaction object for this connector. Since only one transaction can be in progress at any given time,

/// this instance is recycled. To check whether a transaction is currently in progress on this connector,

/// see <see cref="TransactionStatus"/>.

/// </summary>

internal NpgsqlTransaction Transaction { get; }

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

int _pendingPrependedResponses;

internal NpgsqlDataReader? CurrentReader;

internal PreparedStatementManager PreparedStatementManager;

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

/// <summary>

/// Holds all run-time parameters in raw, binary format for efficient handling without allocations.

/// </summary>

readonly List<(byte[] Name, byte[] Value)> _rawParameters = new List<(byte[], byte[])>();

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

int ReceiveTimeout

{

set

{

// TODO: Socket.ReceiveTimeout doesn't work for async.

if (value != _currentTimeout)

_socket.ReceiveTimeout = _currentTimeout = value;

}

}

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

// This is used by NpgsqlCommand, but we place it on the connector because only one instance is needed

// at any one time (per connection).

internal SqlQueryParser SqlParser { get; }

/// <summary>

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

/// Only used when keepalive is enabled, otherwise null.

/// </summary>

SemaphoreSlim? _userLock;

/// <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 bool _isKeepAliveEnabled;

readonly Timer? _keepAliveTimer;

/// <summary>

/// The command currently being executed by the connector, null otherwise.

/// Used only for concurrent use error reporting purposes.

/// </summary>

NpgsqlCommand? _currentCommand;

/// <summary>

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

/// </summary>

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

/// <summary>

/// If pooled, the pool index on which this connector will be returned to the pool.

/// </summary>

internal int PoolIndex { get; set; } = int.MaxValue;

internal int ClearCounter { get; set; }

static readonly NpgsqlLogger Log = NpgsqlLogManager.CreateLogger(nameof(NpgsqlConnector));

#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

byte[]? _resetWithoutDeallocateMessage;

int _resetWithoutDeallocateResponseCount;

// Backend

readonly CommandCompleteMessage _commandCompleteMessage = new CommandCompleteMessage();

readonly ReadyForQueryMessage _readyForQueryMessage = new ReadyForQueryMessage();

readonly ParameterDescriptionMessage _parameterDescriptionMessage = new ParameterDescriptionMessage();

readonly DataRowMessage _dataRowMessage = new DataRowMessage();

readonly RowDescriptionMessage _rowDescriptionMessage = new RowDescriptionMessage();

// Since COPY is rarely used, allocate these lazily

CopyInResponseMessage? _copyInResponseMessage;

CopyOutResponseMessage? _copyOutResponseMessage;

CopyDataMessage? _copyDataMessage;

#endregion

internal NpgsqlDataReader DataReader { get; }

#region Constructors

internal NpgsqlConnector(NpgsqlConnection connection)

: this(connection.Settings, connection.OriginalConnectionString)

{

Connection = connection;

Connection.Connector = this;

ProvideClientCertificatesCallback = Connection.ProvideClientCertificatesCallback;

UserCertificateValidationCallback = Connection.UserCertificateValidationCallback;

ProvidePasswordCallback = Connection.ProvidePasswordCallback;

}

NpgsqlConnector(NpgsqlConnector connector)

: this(connector.Settings, connector.ConnectionString)

{

ProvideClientCertificatesCallback = connector.ProvideClientCertificatesCallback;

UserCertificateValidationCallback = connector.UserCertificateValidationCallback;

ProvidePasswordCallback = connector.ProvidePasswordCallback;

}

/// <summary>

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

/// </summary>

/// <param name="settings">The parsed connection string.</param>

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

NpgsqlConnector(NpgsqlConnectionStringBuilder settings, string connectionString)

{

State = ConnectorState.Closed;

TransactionStatus = TransactionStatus.Idle;

Settings = settings;

ConnectionString = connectionString;

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

SqlParser = new SqlQueryParser(UseConformantStrings);

Transaction = new NpgsqlTransaction(this);

CancelLock = new object();

_isKeepAliveEnabled = Settings.KeepAlive > 0;

if (_isKeepAliveEnabled)

{

_userLock = new SemaphoreSlim(1, 1);

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

}

DataReader = new NpgsqlDataReader(this);

// TODO: Not just for automatic preparation anymore...

PreparedStatementManager = new PreparedStatementManager(this);

}

#endregion

#region Configuration settings

string Host => Settings.Host!;

int Port => Settings.Port;

string KerberosServiceName => Settings.KerberosServiceName;

SslMode SslMode => Settings.SslMode;

int ConnectionTimeout => Settings.Timeout;

bool IntegratedSecurity => Settings.IntegratedSecurity;

internal bool ConvertInfinityDateTime => Settings.ConvertInfinityDateTime;

int InternalCommandTimeout

{

get

{

var internalTimeout = Settings.InternalCommandTimeout;

if (internalTimeout == -1)

return Math.Max(Settings.CommandTimeout, MinimumInternalCommandTimeout) * 1000;

Debug.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 => (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

=> State switch

{

ConnectorState.Ready => true,

ConnectorState.Executing => true,

ConnectorState.Fetching => true,

ConnectorState.Waiting => true,

ConnectorState.Copy => true,

ConnectorState.Closed => false,

ConnectorState.Connecting => false,

ConnectorState.Broken => false,

_ => throw new ArgumentOutOfRangeException("Unknown state: " + State)

};

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

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

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

bool _isConnecting;

#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 async Task Open(NpgsqlTimeout timeout, bool async, CancellationToken cancellationToken)

{

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

Debug.Assert(State == ConnectorState.Closed);

if (string.IsNullOrWhiteSpace(Host))

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

_isConnecting = true;

State = ConnectorState.Connecting;

try {

await RawOpen(timeout, async, cancellationToken);

var username = GetUsername();

if (Settings.Database == null)

Settings.Database = username;

WriteStartupMessage(username);

await Flush(async);

timeout.Check();

await Authenticate(username, timeout, async);

// We treat BackendKeyData as optional because some PostgreSQL-like database

// don't send it (CockroachDB, CrateDB)

var msg = await ReadMessage(async);

if (msg.Code == BackendMessageCode.BackendKeyData)

{

var keyDataMsg = (BackendKeyDataMessage)msg;

BackendProcessId = keyDataMsg.BackendProcessId;

_backendSecretKey = keyDataMsg.BackendSecretKey;

msg = await ReadMessage(async);

}

if (msg.Code != BackendMessageCode.ReadyForQuery)

throw new NpgsqlException($"Received backend message {msg.Code} while expecting ReadyForQuery. Please file a bug.");

State = ConnectorState.Ready;

await LoadDatabaseInfo(timeout, async);

if (Settings.Pooling && DatabaseInfo.SupportsDiscard)

GenerateResetMessage();

Counters.NumberOfNonPooledConnections.Increment();

Counters.HardConnectsPerSecond.Increment();

Log.Trace($"Opened connection to {Host}:{Port}");

// If an exception occurs during open, Break() below shouldn't close the connection, which would also

// update pool state. Instead we let the exception propagate and get handled by the calling pool code.

// We use an extra state flag because the connector's State varies during the type loading query

// above (Executing, Fetching...). Note also that Break() gets called from ReadMessageLong().

_isConnecting = false;

}

catch

{

Break();

throw;

}

}

internal async Task LoadDatabaseInfo(NpgsqlTimeout timeout, bool async)

{

// The type loading below will need to send queries to the database, and that depends on a type mapper

// being set up (even if its empty)

TypeMapper = new ConnectorTypeMapper(this);

if (!NpgsqlDatabaseInfo.Cache.TryGetValue(ConnectionString, out var database))

NpgsqlDatabaseInfo.Cache[ConnectionString] = database = await NpgsqlDatabaseInfo.Load(Connection!, timeout, async);

DatabaseInfo = database;

TypeMapper.Bind(DatabaseInfo);

}

void WriteStartupMessage(string username)

{

var startupParams = new Dictionary<string, string>

{

["user"] = username,

["client_encoding"] =

Settings.ClientEncoding ??

Environment.GetEnvironmentVariable("PGCLIENTENCODING") ??

"UTF8"

};

startupParams["database"] = Settings.Database!;

if (Settings.ApplicationName?.Length > 0)

startupParams["application_name"] = Settings.ApplicationName;

if (Settings.SearchPath?.Length > 0)

startupParams["search_path"] = Settings.SearchPath;

var timezone = Settings.Timezone ?? Environment.GetEnvironmentVariable("PGTZ");

if (timezone != null)

startupParams["TimeZone"] = timezone;

WriteStartup(startupParams);

}

string GetUsername()

{

var username = Settings.Username;

if (username?.Length > 0)

return username;

#if NET461

if (PGUtil.IsWindows && Type.GetType("Mono.Runtime") == null)

{

username = WindowsUsernameProvider.GetUsername(Settings.IncludeRealm);

if (username?.Length > 0)

return username;

}

#endif

if (!PGUtil.IsWindows)

{

username = KerberosUsernameProvider.GetUsername(Settings.IncludeRealm);

if (username?.Length > 0)

return username;

}

username = Environment.UserName;

if (username?.Length > 0)

return username;

username = Environment.GetEnvironmentVariable("USERNAME") ?? Environment.GetEnvironmentVariable("USER");

if (username?.Length > 0)

return username;

if (username is null)

throw new NpgsqlException("No username could be found, please specify one explicitly");

return username;

}

async Task RawOpen(NpgsqlTimeout timeout, bool async, CancellationToken cancellationToken)

{

try

{

if (async)

await ConnectAsync(timeout, cancellationToken);

else

Connect(timeout);

_baseStream = new NetworkStream(_socket, true);

_stream = _baseStream;

if (Settings.Encoding == "UTF8")

{

TextEncoding = PGUtil.UTF8Encoding;

RelaxedTextEncoding = PGUtil.RelaxedUTF8Encoding;

}

else

{

TextEncoding = Encoding.GetEncoding(Settings.Encoding, EncoderFallback.ExceptionFallback, DecoderFallback.ExceptionFallback);

RelaxedTextEncoding = Encoding.GetEncoding(Settings.Encoding, EncoderFallback.ReplacementFallback, DecoderFallback.ReplacementFallback);

}

ReadBuffer = new NpgsqlReadBuffer(this, _stream, Settings.ReadBufferSize, TextEncoding, RelaxedTextEncoding);

WriteBuffer = new NpgsqlWriteBuffer(this, _stream, Settings.WriteBufferSize, TextEncoding);

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

{

WriteSslRequest();

await Flush(async);

await ReadBuffer.Ensure(1, async);

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

timeout.Check();

switch (response)

{

default:

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

case 'N':

if (SslMode == SslMode.Require)

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

break;

case 'S':

var clientCertificates = new X509CertificateCollection();

if (Settings.ClientCertificate != null)

clientCertificates.Add(new X509Certificate(Settings.ClientCertificate));

else if (Environment.GetEnvironmentVariable("PGSSLCERT") is string envCertPath)

clientCertificates.Add(new X509Certificate(envCertPath));

else if (Environment.GetEnvironmentVariable(PGUtil.IsWindows ? "APPDATA" : "HOME") is string homeDir)

{

var certPath = Path.Combine(homeDir, "postgresql", "postgresql.crt");

if (File.Exists(certPath))

clientCertificates.Add(new X509Certificate(certPath));

}

ProvideClientCertificatesCallback?.Invoke(clientCertificates);

RemoteCertificateValidationCallback certificateValidationCallback;

if (Settings.TrustServerCertificate)

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

else if (UserCertificateValidationCallback != null)

certificateValidationCallback = UserCertificateValidationCallback;

else

certificateValidationCallback = DefaultUserCertificateValidationCallback;

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

if (async)

await sslStream.AuthenticateAsClientAsync(Host, clientCertificates, SslProtocols.Tls | SslProtocols.Tls11 | SslProtocols.Tls12, Settings.CheckCertificateRevocation);

else

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

_stream = sslStream;

timeout.Check();

ReadBuffer.Clear(); // Reset to empty after reading single SSL char

ReadBuffer.Underlying = _stream;

WriteBuffer.Underlying = _stream;

IsSecure = true;

Log.Trace("SSL negotiation successful");

break;

}

}

if (!IsSecure)

{

WriteBuffer.AwaitableSocket = new AwaitableSocket(new SocketAsyncEventArgs(), _socket);

ReadBuffer.AwaitableSocket = new AwaitableSocket(new SocketAsyncEventArgs(), _socket);

}

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

}

catch

{

#pragma warning disable CS8625

try { _stream?.Dispose(); } catch {

// ignored

}

_stream = null;

try { _baseStream?.Dispose(); }

catch

{

// ignored

}

_baseStream = null;

try { _socket?.Dispose(); }

catch

{

// ignored

}

_socket = null;

throw;

#pragma warning restore CS8625

}

}

void Connect(NpgsqlTimeout timeout)

{

EndPoint[] endpoints;

if (!string.IsNullOrEmpty(Host) && Host[0] == '/')

{

endpoints = new EndPoint[] { new UnixEndPoint(Path.Combine(Host, $".s.PGSQL.{Port}")) };

}

else

{

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

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

endpoints = Dns.GetHostAddresses(Host).Select(a => new IPEndPoint(a, Port)).ToArray();

timeout.Check();

}

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

var perEndpointTimeout = -1; // Default to infinity

if (timeout.IsSet)

{

var timeoutTicks = timeout.TimeLeft.Ticks;

if (timeoutTicks <= 0)

throw new TimeoutException();

perEndpointTimeout = (int)(timeoutTicks / endpoints.Length / 10);

}

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

{

var endpoint = endpoints[i];

Log.Trace($"Attempting to connect to {endpoint}");

var protocolType = endpoint.AddressFamily == AddressFamily.InterNetwork ? ProtocolType.Tcp : ProtocolType.IP;

var socket = new Socket(endpoint.AddressFamily, SocketType.Stream, protocolType)

{

Blocking = false

};

try

{

try

{

socket.Connect(endpoint);

}

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

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

if (errorCode != 0)

throw new SocketException(errorCode);

if (!write.Any())

throw new TimeoutException("Timeout during connection attempt");

socket.Blocking = true;

SetSocketOptions(socket);

_socket = socket;

return;

}

catch (Exception e)

{

try { socket.Dispose(); }

catch

{

// ignored

}

Log.Trace($"Failed to connect to {endpoint}", e);

if (i == endpoints.Length - 1)

throw new NpgsqlException("Exception while connecting", e);

}

}

}

async Task ConnectAsync(NpgsqlTimeout timeout, CancellationToken cancellationToken)

{

EndPoint[] endpoints;

if (!string.IsNullOrEmpty(Host) && Host[0] == '/')

{

endpoints = new EndPoint[] { new UnixEndPoint(Path.Combine(Host, $".s.PGSQL.{Port}")) };

}

else

{

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

endpoints = (await Dns.GetHostAddressesAsync(Host).WithCancellation(cancellationToken))

.Select(a => new IPEndPoint(a, Port)).ToArray();

}

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

var perIpTimespan = default(TimeSpan);

var perIpTimeout = timeout;

if (timeout.IsSet)

{

var timeoutTicks = timeout.TimeLeft.Ticks;

if (timeoutTicks <= 0)

throw new TimeoutException();

perIpTimespan = new TimeSpan(timeoutTicks / endpoints.Length);

perIpTimeout = new NpgsqlTimeout(perIpTimespan);

}

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

{

var endpoint = endpoints[i];

Log.Trace($"Attempting to connect to {endpoint}");

var protocolType = endpoint.AddressFamily == AddressFamily.InterNetwork ? ProtocolType.Tcp : ProtocolType.IP;

var socket = new Socket(endpoint.AddressFamily, SocketType.Stream, protocolType);

try

{

await socket.ConnectAsync(endpoint)

.WithCancellationAndTimeout(perIpTimeout, cancellationToken);

SetSocketOptions(socket);

_socket = socket;

return;

}

catch (Exception e)

{

try { socket.Dispose(); }

catch

{

// ignored

}

cancellationToken.ThrowIfCancellationRequested();

if (e is OperationCanceledException)

e = new TimeoutException("Timeout during connection attempt");

Log.Trace($"Failed to connect to {endpoint}", e);

if (i == endpoints.Length - 1)

{

throw new NpgsqlException("Exception while connecting", e);

}

}

}

}

void SetSocketOptions(Socket socket)

{

if (socket.AddressFamily == AddressFamily.InterNetwork)

socket.NoDelay = true;

if (Settings.SocketReceiveBufferSize > 0)

socket.ReceiveBufferSize = Settings.SocketReceiveBufferSize;

if (Settings.SocketSendBufferSize > 0)

socket.SendBufferSize = Settings.SocketSendBufferSize;

if (Settings.TcpKeepAlive)

socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true);

if (Settings.TcpKeepAliveInterval > 0 && Settings.TcpKeepAliveTime == 0)

throw new ArgumentException("If TcpKeepAliveInterval is defined, TcpKeepAliveTime must be defined as well");

if (Settings.TcpKeepAliveTime > 0)

{

if (!PGUtil.IsWindows)

throw new PlatformNotSupportedException(

"Npgsql management of TCP keepalive is supported only on Windows. " +

"TCP keepalives can still be used on other systems but are enabled via the TcpKeepAlive option or configured globally for the machine, see the relevant docs.");

var time = Settings.TcpKeepAliveTime;

var interval = Settings.TcpKeepAliveInterval > 0

? Settings.TcpKeepAliveInterval

: Settings.TcpKeepAliveTime;

// For the following see https://msdn.microsoft.com/en-us/library/dd877220.aspx

var uintSize = Marshal.SizeOf(typeof(uint));

var inOptionValues = new byte[uintSize * 3];

BitConverter.GetBytes((uint)1).CopyTo(inOptionValues, 0);

BitConverter.GetBytes((uint)time).CopyTo(inOptionValues, uintSize);

BitConverter.GetBytes((uint)interval).CopyTo(inOptionValues, uintSize * 2);

var result = socket.IOControl(IOControlCode.KeepAliveValues, inOptionValues, null);

if (result != 0)

throw new NpgsqlException($"Got non-zero value when trying to set TCP keepalive: {result}");

}

}

#endregion

#region Frontend message processing

/// <summary>

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

/// </summary>

internal void PrependInternalMessage(byte[] rawMessage, int responseMessageCount)

{

_pendingPrependedResponses += responseMessageCount;

var t = WritePregenerated(rawMessage);

Debug.Assert(t.IsCompleted, "Could not fully write pregenerated message into the buffer");

}

#endregion

#region Backend message processing

internal IBackendMessage ReadMessage(DataRowLoadingMode dataRowLoadingMode=DataRowLoadingMode.NonSequential)

=> ReadMessage(false, dataRowLoadingMode).GetAwaiter().GetResult()!;

internal ValueTask<IBackendMessage> ReadMessage(bool async, DataRowLoadingMode dataRowLoadingMode = DataRowLoadingMode.NonSequential)

=> DoReadMessage(async, dataRowLoadingMode)!;

internal ValueTask<IBackendMessage?> ReadMessageWithNotifications(bool async)

=> DoReadMessage(async, DataRowLoadingMode.NonSequential, true);

ValueTask<IBackendMessage?> DoReadMessage(

bool async,

DataRowLoadingMode dataRowLoadingMode = DataRowLoadingMode.NonSequential,

bool readingNotifications = false)

{

if (_pendingPrependedResponses > 0 ||

dataRowLoadingMode != DataRowLoadingMode.NonSequential ||

readingNotifications ||

ReadBuffer.ReadBytesLeft < 5)

{

return ReadMessageLong(dataRowLoadingMode, readingNotifications);

}

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

switch (messageCode)

{

case BackendMessageCode.NoticeResponse:

case BackendMessageCode.NotificationResponse:

case BackendMessageCode.ParameterStatus:

case BackendMessageCode.ErrorResponse:

ReadBuffer.ReadPosition--;

return ReadMessageLong(dataRowLoadingMode, readingNotifications2: false);

}

PGUtil.ValidateBackendMessageCode(messageCode);

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

if (len > ReadBuffer.ReadBytesLeft)

{

ReadBuffer.ReadPosition -= 5;

return ReadMessageLong(dataRowLoadingMode, readingNotifications2: false);

}

return new ValueTask<IBackendMessage?>(ParseServerMessage(ReadBuffer, messageCode, len, false));

async ValueTask<IBackendMessage?> ReadMessageLong(

DataRowLoadingMode dataRowLoadingMode2,

bool readingNotifications2,

bool isReadingPrependedMessage = false)

{

// First read the responses of any prepended messages.

if (_pendingPrependedResponses > 0 && !isReadingPrependedMessage)

{

try

{

// TODO: There could be room for optimization here, rather than the async call(s)

ReceiveTimeout = InternalCommandTimeout;

for (; _pendingPrependedResponses > 0; _pendingPrependedResponses--)

await ReadMessageLong(DataRowLoadingMode.Skip, false, true);

}

catch (PostgresException)

{

Break();

throw;

}

}

PostgresException? error = null;

try

{

ReceiveTimeout = UserTimeout;

while (true)

{

await ReadBuffer.Ensure(5, async, readingNotifications2);

messageCode = (BackendMessageCode)ReadBuffer.ReadByte();

PGUtil.ValidateBackendMessageCode(messageCode);

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

if ((messageCode == BackendMessageCode.DataRow &&

dataRowLoadingMode2 != DataRowLoadingMode.NonSequential) ||

messageCode == BackendMessageCode.CopyData)

{

if (dataRowLoadingMode2 == DataRowLoadingMode.Skip)

{

await ReadBuffer.Skip(len, async);

continue;

}

}

else if (len > ReadBuffer.ReadBytesLeft)

{

if (len > ReadBuffer.Size)

{

if (_origReadBuffer == null)

_origReadBuffer = ReadBuffer;

ReadBuffer = ReadBuffer.AllocateOversize(len);

}

await ReadBuffer.Ensure(len, async);

}

var msg = ParseServerMessage(ReadBuffer, messageCode, len, isReadingPrependedMessage);

switch (messageCode)

{

case BackendMessageCode.ErrorResponse:

Debug.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 = PostgresException.Load(ReadBuffer);

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)

{

NpgsqlEventSource.Log.CommandFailed();

throw error;

}

break;

// Asynchronous messages which can come anytime, they have already been handled

// in ParseServerMessage. Read the next message.

case BackendMessageCode.NoticeResponse:

case BackendMessageCode.NotificationResponse:

case BackendMessageCode.ParameterStatus:

Debug.Assert(msg == null);

if (!readingNotifications2)

continue;

return null;

}

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

return msg;

}

}

catch (PostgresException)

{

if (CurrentReader != null)

{

// The reader cleanup will call EndUserAction

await CurrentReader.Cleanup(async);

}

else

{

EndUserAction();

}

throw;

}

catch (NpgsqlException)

{

// An ErrorResponse isn't followed by ReadyForQuery

if (error != null)

ExceptionDispatchInfo.Capture(error).Throw();

throw;

}

}

}

internal IBackendMessage? ParseServerMessage(NpgsqlReadBuffer buf, BackendMessageCode code, int len, bool isPrependedMessage)

{

switch (code)

{

case BackendMessageCode.RowDescription:

return _rowDescriptionMessage.Load(buf, TypeMapper);

case BackendMessageCode.DataRow:

return _dataRowMessage.Load(len);

case BackendMessageCode.CompletedResponse:

return _commandCompleteMessage.Load(buf, len);

case BackendMessageCode.ReadyForQuery:

var rfq = _readyForQueryMessage.Load(buf);

if (!isPrependedMessage) {

// Transaction status on prepended messages shouldn't be processed, because there may be prepended messages

// before the begin transaction message. In this case, they will contain transaction status Idle, which will

// clear our Pending transaction status. Only process transaction status on RFQ's from user-provided, non

// prepended messages.

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: