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

Stream _stream;

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 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 { 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.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>();

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

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

Buffer.Flush();

timeout.Check();

HandleAuthentication(timeout);

TypeHandlerRegistry.Setup(this, timeout);

State = ConnectorState.Ready;

if (ContinuousProcessing) {

HandleAsyncMessages();

}

}

catch

{

try { Break(); }

catch {

// ignored

}

throw;

}

}

void WriteStartupMessage()

{

var startupMessage = new StartupMessage();

startupMessage["client_encoding"] = "UTF8";

startupMessage["user"] = UserName;

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

}

[RewriteAsync]

public void RawOpen(NpgsqlTimeout timeout)

{

try

{

Connect(timeout);

Contract.Assert(_socket != null);

_baseStream = new NetworkStream(_socket, true);

_stream = _baseStream;

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

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

{

SSLRequestMessage.Instance.Write(Buffer);

Buffer.Flush();

Buffer.Ensure(1);

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

timeout.Check();

switch (response)

{

default:

throw new Exception(string.Format("Received unknown response {0} for SSLRequest (expecting S or N)", response));

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

if (Connection.ProvideClientCertificatesCallback != null)

{

Connection.ProvideClientCertificatesCallback(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)

{

#if DNXCORE50

throw new NotSupportedException("TLS implementation not yet supported with .NET Core, specify UseSslStream=true for now");

#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, SslProtocols.Tls | SslProtocols.Tls11 | SslProtocols.Tls12, false);

_stream = sslStream;

}

timeout.Check();

Buffer.Underlying = _stream;

IsSecure = true;

break;

}

}

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 Connect(NpgsqlTimeout timeout)

{

#if DNXCORE50

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

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

#else

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

#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], Id);

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(string.Format("Timeout after {0} seconds when connecting to {1}",

new TimeSpan(perIpTimeout * 10).TotalSeconds, 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);

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

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(string.Format("Timeout after {0} seconds when connecting to {1}",

perIpTimespan.TotalSeconds, 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.Debug("Authenticating...", Id);

while (true)

{

var msg = ReadSingleMessage();

timeout.Check();

switch (msg.Code)

{

case BackendMessageCode.AuthenticationRequest:

var passwordMessage = ProcessAuthenticationMessage((AuthenticationRequestMessage)msg);

if (passwordMessage != null)

{

passwordMessage.Write(Buffer);

Buffer.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>

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

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

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

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

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

{

return new PasswordMessage(passwdRead);

}

return null;

#endif

default:

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

}

}

#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, bool withTimeout=true)

{

// Set backend timeout if needed.

if (withTimeout) {

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

checked { _pendingRfqPrependedMessages++; }

}

_messagesToSend.Add(msg);

}

internal void PrependBackendTimeoutMessage(int timeout)

{

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

return;

}

_backendTimeout = timeout;

checked { _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)