src/Coast.PostgreSql/Repository/SagaRepository.cs

这里只显示每个文件的开头 60 行。登录后可以解锁完整代码。

微信扫码关注公众号，代码整理好后第一时间通知你。

namespace Coast.PostgreSql.Repository
{
    using System;
    using System.Data;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;
    using Coast.Core;
    using Dapper;

    public class SagaRepository : ISagaRepository
    {
        private IDbConnection _connection;
        private IDbTransaction _transaction;
        private readonly string _sagaTableName;
        private readonly string _sagaStepTableName;

        /// <summary>
        /// Initializes a new instance of the <see cref="SagaRepository"/> class.
        /// </summary>
        public SagaRepository(string schemaName, IDbConnection connection, IDbTransaction transaction = null)
        {
            _connection = connection;
            _transaction = transaction;
            _sagaTableName = $"\"{schemaName}\".\"Saga\"";
            _sagaStepTableName = $"\"{schemaName}\".\"SagaStep\"";
        }

        /// <inheritdoc/>
        public async Task SaveSagaAsync(Saga saga, CancellationToken cancellationToken = default)
        {
            cancellationToken.ThrowIfCancellationRequested();

            string InsertSagaSql =
$@"INSERT INTO {_sagaTableName} 
(""Id"", ""State"", ""CreationTime"", ""CurrentExecutionSequenceNumber"") 
VALUES (@Id, @State, @CreationTime, @CurrentExecutionSequenceNumber ); ";
            string InsertSagaStepSql =
$@"INSERT INTO {_sagaStepTableName}
(""Id"", ""CorrelationId"", ""EventName"", ""HasCompensation"", ""State"", ""RequestBody"", ""CreationTime"", ""FailedReason"", ""ExecutionSequenceNumber"") 
VALUES (@Id, @CorrelationId, @EventName, @HasCompensation, @State, @RequestBody, @CreationTime,@FailedReason, @ExecutionSequenceNumber); ";

            await _connection.ExecuteAsync(
                    InsertSagaSql,
                    new { Id = saga.Id, State = SagaStateEnum.Started, CreationTime = DateTime.UtcNow, CurrentExecutionSequenceNumber = saga.CurrentExecutionSequenceNumber },
                    transaction: _transaction).ConfigureAwait(false);

            foreach (var step in saga.SagaSteps)
            {
                await _connection.ExecuteAsync(
                    InsertSagaStepSql,
                    new
                    {
                        Id = step.Id,
                        CorrelationId = saga.Id,
                        EventName = step.EventName,
                        State = step.State,
                        RequestBody = step.RequestBody,
                        CreationTime = DateTime.UtcNow,
                        HasCompensation = step.HasCompensation,