diff --git a/src/Persistence/Oracle/OracleTests/oracle_durability_command_translation.cs b/src/Persistence/Oracle/OracleTests/oracle_durability_command_translation.cs new file mode 100644 index 000000000..29fd5858d --- /dev/null +++ b/src/Persistence/Oracle/OracleTests/oracle_durability_command_translation.cs @@ -0,0 +1,46 @@ +using Oracle.ManagedDataAccess.Client; +using Shouldly; +using Wolverine.Oracle; + +namespace OracleTests; + +public class oracle_durability_command_translation +{ + [Fact] + public void rewrites_generic_parameter_markers_for_oracle() + { + OracleMessageStore.normalizeParameterMarkers( + "delete from messages where replayable = @replayable and keep_until <= @p0") + .ShouldBe( + "delete from messages where replayable = :replayable and keep_until <= :p0"); + } + + [Fact] + public void separates_generic_database_batches_into_oracle_statements() + { + OracleMessageStore.splitStatements( + "select destination from outgoing; delete from incoming;") + .ShouldBe( + [ + "select destination from outgoing", + "delete from incoming" + ]); + } + + [Fact] + public void converts_boolean_parameters_to_oracle_number() + { + using var command = new OracleCommand(); + command.Parameters.Add(new OracleParameter + { + ParameterName = "replayable", + Value = true + }); + + OracleMessageStore.normalizeParameters(command); + + var parameter = (OracleParameter)command.Parameters["replayable"]; + parameter.OracleDbType.ShouldBe(OracleDbType.Int16); + parameter.Value.ShouldBe(1); + } +} diff --git a/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.Durability.cs b/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.Durability.cs new file mode 100644 index 000000000..5672e5f67 --- /dev/null +++ b/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.Durability.cs @@ -0,0 +1,133 @@ +using System.Text.RegularExpressions; +using Oracle.ManagedDataAccess.Client; +using Weasel.Core; +using Wolverine.RDBMS.Polling; + +namespace Wolverine.Oracle; + +internal partial class OracleMessageStore +{ + async Task IDatabaseOperationBatchExecutor.ExecuteDatabaseOperationBatchAsync( + IDatabaseOperation[] operations, + CancellationToken cancellationToken) + { + await using var connection = + (OracleConnection)await _dataSource.OpenConnectionAsync(cancellationToken); + await using var transaction = + (OracleTransaction)await connection.BeginTransactionAsync(cancellationToken); + + try + { + foreach (var operation in operations) + { + await executeOperationAsync( + connection, + transaction, + operation, + operations, + cancellationToken); + } + + await transaction.CommitAsync(cancellationToken); + } + catch + { + try + { + await transaction.RollbackAsync(cancellationToken); + } + catch + { + // Preserve the database operation failure that triggered the rollback. + } + + throw; + } + } + + private static async Task executeOperationAsync( + OracleConnection connection, + OracleTransaction transaction, + IDatabaseOperation operation, + IDatabaseOperation[] operations, + CancellationToken cancellationToken) + { + var builder = new DbCommandBuilder(connection); + operation.ConfigureCommand(builder); + + await using var command = (OracleCommand)builder.Compile(); + command.BindByName = true; + command.Transaction = transaction; + + normalizeParameters(command); + + var statements = splitStatements(command.CommandText); + + if (statements.Length == 0) return; + + try + { + if (operation is IDoNotReturnData) + { + foreach (var statement in statements) + { + command.CommandText = normalizeParameterMarkers(statement); + await command.ExecuteNonQueryAsync(cancellationToken); + } + + return; + } + + if (statements.Length != 1) + { + throw new InvalidOperationException( + $"Oracle database operation '{operation}' returns data and must contain exactly one SQL statement."); + } + + command.CommandText = normalizeParameterMarkers(statements[0]); + await using var reader = await command.ExecuteReaderAsync(cancellationToken); + var exceptions = new List(); + await operation.ReadResultsAsync(reader, exceptions, cancellationToken); + await reader.CloseAsync(); + } + catch (ObjectDisposedException) + { + throw; + } + catch (Exception e) + { + throw new DatabaseBatchCommandException(command, operations, e); + } + } + + internal static string normalizeParameterMarkers(string sql) + { + return Regex.Replace(sql, @"@(?=[A-Za-z_])", ":"); + } + + internal static string[] splitStatements(string sql) + { + return sql.Split( + ';', + StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); + } + + internal static void normalizeParameters(OracleCommand command) + { + foreach (OracleParameter parameter in command.Parameters) + { + switch (parameter.Value) + { + case bool boolean: + parameter.OracleDbType = OracleDbType.Int16; + parameter.Value = boolean ? 1 : 0; + break; + + case Guid guid: + parameter.OracleDbType = OracleDbType.Raw; + parameter.Value = guid.ToByteArray(); + break; + } + } + } +} diff --git a/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.cs b/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.cs index 167dbbad0..ebeb0cf23 100644 --- a/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.cs +++ b/src/Persistence/Oracle/Wolverine.Oracle/OracleMessageStore.cs @@ -36,7 +36,7 @@ namespace Wolverine.Oracle; -internal partial class OracleMessageStore : IMessageDatabase, IMessageInbox, IMessageOutbox, IMessageStoreAdmin, IDeadLetters, IScheduledMessages, ISagaSupport +internal partial class OracleMessageStore : IMessageDatabase, IMessageInbox, IMessageOutbox, IMessageStoreAdmin, IDeadLetters, IScheduledMessages, ISagaSupport, IDatabaseOperationBatchExecutor { private readonly OracleDataSource _dataSource; private readonly DatabaseSettings _settings; diff --git a/src/Persistence/Wolverine.RDBMS/Polling/DatabaseOperationBatch.cs b/src/Persistence/Wolverine.RDBMS/Polling/DatabaseOperationBatch.cs index c16595f06..f9ddc3bf6 100644 --- a/src/Persistence/Wolverine.RDBMS/Polling/DatabaseOperationBatch.cs +++ b/src/Persistence/Wolverine.RDBMS/Polling/DatabaseOperationBatch.cs @@ -47,6 +47,21 @@ public async Task ExecuteAsync(IWolverineRuntime runtime, { if (_operations.Length == 0) return AgentCommands.Empty; + if (_database is IDatabaseOperationBatchExecutor executor) + { + try + { + await executor.ExecuteDatabaseOperationBatchAsync(_operations, cancellationToken); + } + catch (ObjectDisposedException) + { + // The system is shutting down, let this go. + return AgentCommands.Empty; + } + + return postProcessingCommands(); + } + var builder = _database.ToCommandBuilder(); foreach (var operation in _operations) operation.ConfigureCommand(builder); @@ -79,12 +94,7 @@ public async Task ExecuteAsync(IWolverineRuntime runtime, try { - var commands = new AgentCommands(); - foreach (var operation in _operations) - { - commands.AddRange(operation.PostProcessingCommands()); - } - return commands; + return postProcessingCommands(); } finally { @@ -99,6 +109,17 @@ public async Task ExecuteAsync(IWolverineRuntime runtime, } } + private AgentCommands postProcessingCommands() + { + var commands = new AgentCommands(); + foreach (var operation in _operations) + { + commands.AddRange(operation.PostProcessingCommands()); + } + + return commands; + } + public static async Task ApplyCallbacksAsync(IReadOnlyList operations, DbDataReader reader, IList exceptions, CancellationToken token) @@ -149,4 +170,4 @@ public static async Task ApplyCallbacksAsync(IReadOnlyList o } } } -} \ No newline at end of file +} diff --git a/src/Persistence/Wolverine.RDBMS/Polling/IDatabaseOperation.cs b/src/Persistence/Wolverine.RDBMS/Polling/IDatabaseOperation.cs index dbcc706ab..16b0ce05a 100644 --- a/src/Persistence/Wolverine.RDBMS/Polling/IDatabaseOperation.cs +++ b/src/Persistence/Wolverine.RDBMS/Polling/IDatabaseOperation.cs @@ -14,4 +14,11 @@ public interface IDatabaseOperation Task ReadResultsAsync(DbDataReader reader, IList exceptions, CancellationToken token); IEnumerable PostProcessingCommands(); -} \ No newline at end of file +} + +internal interface IDatabaseOperationBatchExecutor +{ + Task ExecuteDatabaseOperationBatchAsync( + IDatabaseOperation[] operations, + CancellationToken cancellationToken); +}