From ec2ca727eb71afe0ad9fc412d319bc85c5c8e265 Mon Sep 17 00:00:00 2001 From: Marcin Hoppe Date: Mon, 20 Jun 2016 15:53:51 +0200 Subject: [PATCH 1/2] Quote table and schema name when assembling queue table name --- .../SqlServerMessageSender.cs | 15 ++++++++++++--- .../SqlServerPollingDequeueStrategy.cs | 18 +++++++++++------- 2 files changed, 23 insertions(+), 10 deletions(-) diff --git a/src/NServiceBus.SqlServer/SqlServerMessageSender.cs b/src/NServiceBus.SqlServer/SqlServerMessageSender.cs index 4a0672feb..b64feca9d 100644 --- a/src/NServiceBus.SqlServer/SqlServerMessageSender.cs +++ b/src/NServiceBus.SqlServer/SqlServerMessageSender.cs @@ -15,7 +15,7 @@ public class SqlServerMessageSender : ISendMessages { const string SqlSend = - @"INSERT INTO [{0}].[{1}] ([Id],[CorrelationId],[ReplyToAddress],[Recoverable],[Expires],[Headers],[Body]) + @"INSERT INTO {0}.{1} ([Id],[CorrelationId],[ReplyToAddress],[Recoverable],[Expires],[Headers],[Body]) VALUES (@Id,@CorrelationId,@ReplyToAddress,@Recoverable,@Expires,@Headers,@Body)"; static JsonMessageSerializer Serializer = new JsonMessageSerializer(null); @@ -71,7 +71,7 @@ public void Send(TransportMessage message, Address address) //if there is an active transaction for the connection, we can use the same native transaction var transaction = UnitOfWork.GetTransaction(queueConnectionString); - using (var command = new SqlCommand(string.Format(SqlSend, schemaName, TableNameUtils.GetTableName(address)), transaction.Connection, transaction) + using (var command = new SqlCommand(SendSqlCommandText(schemaName, address), transaction.Connection, transaction) { CommandType = CommandType.Text }) @@ -85,7 +85,7 @@ public void Send(TransportMessage message, Address address) using (var connection = new SqlConnection(queueConnectionString)) { connection.Open(); - using (var command = new SqlCommand(string.Format(SqlSend, schemaName, TableNameUtils.GetTableName(address)), connection) + using (var command = new SqlCommand(SendSqlCommandText(schemaName, address), connection) { CommandType = CommandType.Text }) @@ -114,6 +114,15 @@ public void Send(TransportMessage message, Address address) } } + static string SendSqlCommandText(string schemaName, Address address) + { + var sanitizer = new SqlCommandBuilder(); + + var quotedSchema = sanitizer.QuoteIdentifier(schemaName); + var quotedTable = sanitizer.QuoteIdentifier(TableNameUtils.GetTableName(address)); + + return string.Format(SqlSend, quotedSchema, quotedTable); + } private static void ThrowFailedToSendException(Address address, Exception ex) { diff --git a/src/NServiceBus.SqlServer/SqlServerPollingDequeueStrategy.cs b/src/NServiceBus.SqlServer/SqlServerPollingDequeueStrategy.cs index 1e44ea629..c72688396 100644 --- a/src/NServiceBus.SqlServer/SqlServerPollingDequeueStrategy.cs +++ b/src/NServiceBus.SqlServer/SqlServerPollingDequeueStrategy.cs @@ -63,9 +63,12 @@ public void Init(Address address, TransactionSettings transactionSettings, Timeout = transactionSettings.TransactionTimeout }; - tableName = TableNameUtils.GetTableName(address); + var sanitizer = new SqlCommandBuilder(); - sql = string.Format(SqlReceive, SchemaName, tableName); + quotedSchemaName = sanitizer.QuoteIdentifier(SchemaName); + quotedTableName = sanitizer.QuoteIdentifier(TableNameUtils.GetTableName(address)); + + sql = string.Format(SqlReceive, quotedSchemaName, quotedTableName); if (PurgeOnStartup) { @@ -110,14 +113,14 @@ void PurgeTable() { connection.Open(); - using (var command = new SqlCommand(string.Format(SqlPurge, SchemaName, tableName), connection) + using (var command = new SqlCommand(string.Format(SqlPurge, quotedSchemaName, quotedTableName), connection) { CommandType = CommandType.Text }) { var numberOfPurgedRows = command.ExecuteNonQuery(); - Logger.InfoFormat("{0} messages was purged from table {1}", numberOfPurgedRows, tableName); + Logger.InfoFormat("{0} messages was purged from table {1}", numberOfPurgedRows, quotedTableName); } } } @@ -405,12 +408,12 @@ IsolationLevel GetSqlIsolationLevel(System.Transactions.IsolationLevel isolation } const string SqlReceive = - @"WITH message AS (SELECT TOP(1) * FROM [{0}].[{1}] WITH (UPDLOCK, READPAST, ROWLOCK) ORDER BY [RowVersion] ASC) + @"WITH message AS (SELECT TOP(1) * FROM {0}.{1} WITH (UPDLOCK, READPAST, ROWLOCK) ORDER BY [RowVersion] ASC) DELETE FROM message OUTPUT deleted.Id, deleted.CorrelationId, deleted.ReplyToAddress, deleted.Recoverable, deleted.Expires, deleted.Headers, deleted.Body;"; - const string SqlPurge = @"DELETE FROM [{0}].[{1}]"; + const string SqlPurge = @"DELETE FROM {0}.{1}"; static readonly JsonMessageSerializer Serializer = new JsonMessageSerializer(null); static readonly ILog Logger = LogManager.GetLogger(typeof(SqlServerPollingDequeueStrategy)); @@ -425,7 +428,8 @@ DELETE FROM message Action endProcessMessage; TransactionSettings settings; string sql; - string tableName; + string quotedSchemaName; + string quotedTableName; CancellationTokenSource tokenSource; TransactionOptions transactionOptions; Func tryProcessMessage; From 8a7b358aa62724346c4e71145e9380534b14f428 Mon Sep 17 00:00:00 2001 From: Marcin Hoppe Date: Mon, 27 Jun 2016 11:09:04 +0200 Subject: [PATCH 2/2] Dispose SqlCommandBuilder --- src/NServiceBus.SqlServer/SqlServerMessageSender.cs | 11 ++++++----- .../SqlServerPollingDequeueStrategy.cs | 9 +++++---- 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/src/NServiceBus.SqlServer/SqlServerMessageSender.cs b/src/NServiceBus.SqlServer/SqlServerMessageSender.cs index b64feca9d..5ea768170 100644 --- a/src/NServiceBus.SqlServer/SqlServerMessageSender.cs +++ b/src/NServiceBus.SqlServer/SqlServerMessageSender.cs @@ -116,12 +116,13 @@ public void Send(TransportMessage message, Address address) static string SendSqlCommandText(string schemaName, Address address) { - var sanitizer = new SqlCommandBuilder(); - - var quotedSchema = sanitizer.QuoteIdentifier(schemaName); - var quotedTable = sanitizer.QuoteIdentifier(TableNameUtils.GetTableName(address)); + using (var sanitizer = new SqlCommandBuilder()) + { + var quotedSchema = sanitizer.QuoteIdentifier(schemaName); + var quotedTable = sanitizer.QuoteIdentifier(TableNameUtils.GetTableName(address)); - return string.Format(SqlSend, quotedSchema, quotedTable); + return string.Format(SqlSend, quotedSchema, quotedTable); + } } private static void ThrowFailedToSendException(Address address, Exception ex) diff --git a/src/NServiceBus.SqlServer/SqlServerPollingDequeueStrategy.cs b/src/NServiceBus.SqlServer/SqlServerPollingDequeueStrategy.cs index c72688396..32654bca9 100644 --- a/src/NServiceBus.SqlServer/SqlServerPollingDequeueStrategy.cs +++ b/src/NServiceBus.SqlServer/SqlServerPollingDequeueStrategy.cs @@ -63,10 +63,11 @@ public void Init(Address address, TransactionSettings transactionSettings, Timeout = transactionSettings.TransactionTimeout }; - var sanitizer = new SqlCommandBuilder(); - - quotedSchemaName = sanitizer.QuoteIdentifier(SchemaName); - quotedTableName = sanitizer.QuoteIdentifier(TableNameUtils.GetTableName(address)); + using (var sanitizer = new SqlCommandBuilder()) + { + quotedSchemaName = sanitizer.QuoteIdentifier(SchemaName); + quotedTableName = sanitizer.QuoteIdentifier(TableNameUtils.GetTableName(address)); + } sql = string.Format(SqlReceive, quotedSchemaName, quotedTableName);