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

{

readonly NpgsqlConnectionStringBuilder _settings;

/// <summary>

/// The physical connection socket to the backend.

/// </summary>

internal Socket Socket { get; set; }

/// <summary>

/// The physical connection stream to the backend.

/// </summary>

internal NpgsqlNetworkStream BaseStream { get; set; }

// The top level stream to the backend.

// This is a BufferedStream.

// With SSL, this stream sits on top of the SSL stream, which sits on top of _baseStream.

// Otherwise, this stream sits directly on top of _baseStream.

internal Stream Stream { get; set; }

/// <summary>

/// Buffer used for reading data.

/// </summary>

internal NpgsqlBuffer Buffer { get; private set; }

/// <summary>

/// The connection mediator.

/// </summary>

internal NpgsqlMediator Mediator { 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>

internal int BackendSecretKey { get; set; }

/// <summary>

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

/// </summary>

internal int BackendProcessId { get; set; }

internal bool Pooled { get; private set; }

internal TypeHandlerRegistry TypeHandlerRegistry { get; set; }

TransactionStatus _txStatus;

NpgsqlTransaction _tx;

/// <summary>

/// The number of messages that were prepended to the current message chain.

/// </summary>

byte _prependedMessages;

/// <summary>

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

/// </summary>

List<FrontendMessage> _messagesToSend;

internal NpgsqlDataReader CurrentReader;

/// <summary>

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

/// </summary>

internal Dictionary<string, string> BackendParams;

/// <summary>

/// Commands to be executed when the reader is done.

/// Current usage is for when a prepared command is disposed while its reader is still open; the

/// actual DEALLOCATE message must be deferred.

/// </summary>

List<string> _deferredCommands;

/// <summary>

/// Whether the backend is an AWS Redshift instance

/// </summary>

bool? _isRedshift;

// For IsValid test

readonly RNGCryptoServiceProvider _rng = new RNGCryptoServiceProvider();

string _initQueries;

internal SSPIHandler SSPI { get; set; }

static readonly ILog _log = LogManager.GetCurrentClassLogger();

SemaphoreSlim _notificationSemaphore;

byte[] _emptyBuffer = new byte[0];

int _notificationBlockRecursionDepth;

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

#endregion

public NpgsqlConnector(NpgsqlConnection connection)

: this(connection.CopyConnectionStringBuilder(), connection.Pooling)

{}

/// <summary>

/// Constructor.

/// </summary>

/// <param name="connectionString">Connection string.</param>

/// <param name="pooled">Pooled</param>

public NpgsqlConnector(NpgsqlConnectionStringBuilder connectionString, bool pooled)

{

State = ConnectorState.Closed;

_txStatus = TransactionStatus.Idle;

_settings = connectionString;

Pooled = pooled;

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

Mediator = new NpgsqlMediator();

_messagesToSend = new List<FrontendMessage>();

_preparedStatementIndex = 0;

_portalIndex = 0;

_deferredCommands = new List<string>();

}

#region Configuration settings

/// <summary>

/// Return Connection String.

/// </summary>

internal static bool UseSslStream = true;

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.ContainsKey(Keywords.Database) ? _settings.Database : _settings.UserName; } }

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

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

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

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

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

internal bool UseMonoSsl { get { return ValidateRemoteCertificateCallback == null; } }

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

internal int DefaultCommandTimeout { get { return _settings.CommandTimeout; } }

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

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

internal Version CompatVersion { get { return _settings.Compatible; } }

#endregion Configuration settings

#region State management

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

switch (value)

{

case ConnectorState.Ready:

ExecuteDeferredCommands();

if (CurrentReader != null) {

CurrentReader.Command.State = CommandState.Idle;

CurrentReader = null;

}

break;

case ConnectorState.Closed:

case ConnectorState.Broken:

if (CurrentReader != null) {

CurrentReader.Command.State = CommandState.Idle;

CurrentReader = null;

}

ClearTransaction();

break;

case ConnectorState.Connecting:

case ConnectorState.Executing:

case ConnectorState.Fetching:

case ConnectorState.CopyIn:

case ConnectorState.CopyOut:

break;

default:

throw new ArgumentOutOfRangeException("value");

}

}

}

/// <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.CopyIn:

case ConnectorState.CopyOut:

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.CopyIn:

case ConnectorState.CopyOut:

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 void CheckReadyState()

{

switch (State)

{

case ConnectorState.Ready:

return;

case ConnectorState.Closed:

case ConnectorState.Broken:

case ConnectorState.Connecting:

throw new InvalidOperationException(L10N.ConnectionNotOpen);

case ConnectorState.Executing:

case ConnectorState.Fetching:

throw new InvalidOperationException("There is already an open DataReader associated with this Connection which must be closed first.");

case ConnectorState.CopyIn:

case ConnectorState.CopyOut:

throw new InvalidOperationException("A COPY operation is in progress and must complete first.");

default:

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

}

}

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

{

if (State != ConnectorState.Closed) {

throw new InvalidOperationException("Can't open, state is " + State);

}

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;

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

RawOpen(connectTimeRemaining);

try

{

var startupMessage = new StartupMessage(Database, UserName);

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

startupMessage.ApplicationName = _settings.ApplicationName;

}

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

startupMessage.SearchPath = _settings.SearchPath;

}

var len = startupMessage.Length;

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

throw new Exception(String.Format("Buffer ({0} bytes) not big enough to contain Startup message ({1} bytes)", Buffer.Size, len));

}

startupMessage.Prepare();

if (startupMessage.Length > Buffer.Size) {

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

}

startupMessage.Write(Buffer);

// TODO: Possible optimization: send settings like ssl_renegotiation in the same packet,

// reduce one roundtrip

Buffer.Flush();

HandleAuthentication();

}

catch

{

if (Stream != null)

{

try {

Stream.Dispose();

}

catch {}

}

throw;

}

// After attachment, the stream will close the connector (this) when the stream gets disposed.

BaseStream.AttachConnector(this);

// Fall back to the old way, SELECT VERSION().

// This should not happen for protocol version 3+.

if (ServerVersion == null)

{

//NpgsqlCommand command = new NpgsqlCommand("set DATESTYLE TO ISO;select version();", this);

//ServerVersion = new Version(PGUtil.ExtractServerVersion((string) command.ExecuteScalar()));

using (var command = new NpgsqlCommand("set DATESTYLE TO ISO;select version();", this))

{

ServerVersion = new Version(PGUtil.ExtractServerVersion((string)command.ExecuteScalar()));

}

}

ProcessServerVersion();

var sbInitQueries = new StringWriter();

// Some connection parameters for protocol 3 had been sent in the startup packet.

// The rest will be setted here.

if (SupportsSslRenegotiationLimit) {

sbInitQueries.WriteLine("SET ssl_renegotiation_limit=0;");

}

_initQueries = sbInitQueries.ToString();

ExecuteBlind(_initQueries);

TypeHandlerRegistry.Setup(this);

// Make a shallow copy of the type mapping that the connector will own.

// It is possible that the connector may add types to its private

// mapping that will not be valid to another connector, even

// if connected to the same backend version.

//NativeToBackendTypeConverterOptions.OidToNameMapping = NpgsqlTypesHelper.CreateAndLoadInitialTypesMapping(this).Clone();

State = ConnectorState.Ready;

if (_settings.SyncNotification)

{

AddNotificationListener();

}

}

public void RawOpen(int timeout)

{

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

}

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

var ips = Dns.EndGetHostAddresses(result);

Socket socket = null;

Exception lastSocketException = null;

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

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

}

socket.EndConnect(result);

// connect was successful, leave the loop

break;

}

catch (Exception e)

{

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

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

lastSocketException = e;

socket.Close();

socket = null;

}

}

if (socket == null)

{

throw lastSocketException;

}

var baseStream = new NpgsqlNetworkStream(socket, true);

Stream sslStream = null;

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

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

{

baseStream

.WriteInt32(8)

.WriteInt32(80877103);

// Receive response

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

if (response != 'S')

{

if (SslMode == SslMode.Require) {

throw new InvalidOperationException(L10N.SslRequestError);

}

}

else

{

//create empty collection

var clientCertificates = new X509CertificateCollection();

//trigger the callback to fetch some certificates

DefaultProvideClientCertificatesCallback(clientCertificates);

//if (context.UseMonoSsl)

if (!UseSslStream)

{

var sslStreamPriv = new SslClientStream(baseStream, Host, true, SecurityProtocolType.Default, clientCertificates)

{

ClientCertSelectionDelegate = DefaultCertificateSelectionCallback,

ServerCertValidationDelegate = DefaultCertificateValidationCallback,

PrivateKeyCertSelectionDelegate = DefaultPrivateKeySelectionCallback

};

sslStream = sslStreamPriv;

IsSecure = true;

}

else

{

var sslStreamPriv = new SslStream(baseStream, true, DefaultValidateRemoteCertificateCallback);

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

sslStream = sslStreamPriv;

IsSecure = true;

}

}

}

Socket = socket;

BaseStream = baseStream;

//Stream = new BufferedStream(sslStream ?? baseStream, 8192);

Stream = BaseStream;

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

_log.DebugFormat("Connected to {0}:{1 }", Host, Port);

}

void HandleAuthentication()

{

_log.Trace("Authenticating...");

while (true)

{

var msg = ReadSingleMessage();

switch (msg.Code)

{

case BackendMessageCode.ReadyForQuery:

State = ConnectorState.Ready;

return;

case BackendMessageCode.AuthenticationRequest:

ProcessAuthenticationMessage((AuthenticationRequestMessage)msg);

continue;

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

}

// For GSSAPI we have to use the supplied hostname

SSPI = new SSPIHandler(Host, "POSTGRES", true);

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

break;

case AuthenticationRequestType.AuthenticationSSPI:

if (!IntegratedSecurity) {

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

}

// For SSPI we have to get the IP-Address (hostname doesn't work)

var ipAddressString = ((IPEndPoint)Socket.RemoteEndPoint).Address.ToString();

SSPI = new SSPIHandler(ipAddressString, "POSTGRES", false);

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

break;

case AuthenticationRequestType.AuthenticationGSSContinue:

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

if (passwdRead.Length != 0)

{

passwordMessage = new PasswordMessage(passwdRead);

break;

}

return;

default:

throw new NotSupportedException(String.Format(L10N.AuthenticationMethodNotSupported, msg.AuthRequestType));

}

passwordMessage.Prepare();

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 PrependMessage(FrontendMessage msg)

{

_prependedMessages++;

_messagesToSend.Add(msg);

}

[GenerateAsync]

internal void SendAllMessages()

{

try

{

foreach (var msg in _messagesToSend)

{

SendMessage(msg);

}

Buffer.Flush();

}

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>

void SendSingleMessage(FrontendMessage msg)

{

AddMessage(msg);

SendAllMessages();

}

[GenerateAsync]

void SendMessage(FrontendMessage msg)

{

try

{

_log.DebugFormat("Sending: {0}", msg);

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, 0, directBuf.Size == 0 ? directBuf.Buffer.Length : directBuf.Size);

directBuf.Buffer = null;

directBuf.Size = 0;

}

}

return;

}

throw PGUtil.ThrowIfReached();

}

catch

{

State = ConnectorState.Broken;

throw;

}

}

#endregion

#region Backend message processing

[GenerateAsync]

internal BackendMessage ReadSingleMessage(DataRowLoadingMode dataRowLoadingMode = DataRowLoadingMode.NonSequential, bool ignoreNotifications = true)

{

try

{

return DoReadSingleMessage(dataRowLoadingMode, ignoreNotifications);

}

catch (NpgsqlException)

{

throw;

}

catch

{

State = ConnectorState.Broken;

throw;

}

}

[GenerateAsync]

BackendMessage DoReadSingleMessage(DataRowLoadingMode dataRowLoadingMode = DataRowLoadingMode.NonSequential, bool ignoreNotifications = true)

{

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

{

if (dataRowLoadingMode == DataRowLoadingMode.Skip)

{

Buffer.Skip(len);

continue;

}

}

else if (len > Buffer.ReadBytesLeft)

{

buf = buf.EnsureOrAllocateTemp(len);

}

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

if (msg != null || !ignoreNotifications && (messageCode == BackendMessageCode.NoticeResponse || messageCode == BackendMessageCode.NotificationResponse))

{

if (error != null)

{

Contract.Assert(messageCode == BackendMessageCode.ReadyForQuery, "Expected ReadyForQuery after ErrorResponse");

throw new NpgsqlException(error);

}

return msg;

}

else if (messageCode == BackendMessageCode.ErrorResponse)

{

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

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

// The exception is during the startup/authentication phase, where the server closes

// the connection after an ErrorResponse

error = new NpgsqlError(buf);

if (State == ConnectorState.Connecting) {

throw new NpgsqlException(error);

}

}

}

}

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

TransactionStatus = 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:

// TODO: Recycle

FireNotice(new NpgsqlError(buf));

return null;

case BackendMessageCode.NotificationResponse:

FireNotification(new NpgsqlNotificationEventArgs(buf));

return null;

case BackendMessageCode.AuthenticationRequest:

var authType = (AuthenticationRequestType)buf.ReadInt32();

_log.Trace("Received AuthenticationRequest of type " + authType);

switch (authType)

{

case AuthenticationRequestType.AuthenticationOk:

return AuthenticationOkMessage.Instance;

case AuthenticationRequestType.AuthenticationCleartextPassword:

return AuthenticationCleartextPasswordMessage.Instance;

case AuthenticationRequestType.AuthenticationMD5Password:

return AuthenticationMD5PasswordMessage.Load(buf);

case AuthenticationRequestType.AuthenticationGSS:

return AuthenticationGSSMessage.Instance;

case AuthenticationRequestType.AuthenticationSSPI:

return AuthenticationSSPIMessage.Instance;

case AuthenticationRequestType.AuthenticationGSSContinue:

return AuthenticationGSSContinueMessage.Load(buf, len);

default:

throw new NotSupportedException(String.Format(L10N.AuthenticationMethodNotSupported, authType));

}

case BackendMessageCode.BackendKeyData:

BackendProcessId = buf.ReadInt32();

BackendSecretKey = buf.ReadInt32();

return null;

case BackendMessageCode.CopyData:

case BackendMessageCode.CopyDone:

case BackendMessageCode.CancelRequest:

case BackendMessageCode.CopyDataRows:

case BackendMessageCode.CopyInResponse:

case BackendMessageCode.CopyOutResponse:

throw new NotImplementedException();

case BackendMessageCode.PortalSuspended:

case BackendMessageCode.IO_ERROR:

Debug.Fail("Unimplemented message: " + code);

throw new NotImplementedException("Unimplemented message: " + code);

case BackendMessageCode.ErrorResponse:

return null;

case BackendMessageCode.FunctionCallResponse:

// We don't use the obsolete function call protocol

throw new Exception("Unexpected backend message: " + code);

default:

throw PGUtil.ThrowIfReached("Unknown backend message code: " + code);

}

}

/// <summary>

/// Reads backend messages and discards them, stopping only after a message of the given types has

/// been seen. Note that when this method is called, the buffer position must be properly set at

/// the start of the next message.

/// </summary>

internal BackendMessage SkipUntil(params BackendMessageCode[] stopAt)

{

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

while (true)

{

var msg = ReadSingleMessage(DataRowLoadingMode.Skip);

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

if (stopAt.Contains(msg.Code)) {

return msg;

}

}

}

/// <summary>

/// Processes the response from any messages that were prepended to the message chain.

/// These messages are currently assumed to be simple queries with no result sets.

/// </summary>

[GenerateAsync]

internal void ReadPrependedMessageResponses()

{

for (; _prependedMessages > 0; _prependedMessages--)

{

SkipUntil(BackendMessageCode.ReadyForQuery);

}

}

#endregion Backend message processing

#region Transactions

/// <summary>

/// The current transaction status for this connector.

/// </summary>

internal TransactionStatus TransactionStatus

{

get { return _txStatus; }

private set

{

if (value == _txStatus) { return; }

switch (value) {

case TransactionStatus.Idle:

ClearTransaction();

break;

case TransactionStatus.InTransactionBlock:

case TransactionStatus.InFailedTransactionBlock:

break;

case TransactionStatus.Pending:

throw new Exception("Invalid TransactionStatus (should be frontend-only)");

default:

throw PGUtil.ThrowIfReached();

}

_txStatus = value;

}

}

/// <summary>

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

/// 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 backend transaction status.

/// </summary>

internal NpgsqlTransaction Transaction

{

get { return _tx; }

set

{

Contract.Requires(TransactionStatus == TransactionStatus.Idle);