Skip to content

C
CAP
Commit 6cae200e authored by yangxiaodong's avatar yangxiaodong
Browse files
Options

fix bug.

parent 0c902940
Hide whitespace changes
Inline Side-by-side
Showing with 113 additions and 113 deletions
src/DotNetCore.CAP.SqlServer/SqlServerStorageConnection.cs
View file @ 6cae200e
......@@ -25,12 +25,13 @@ namespace DotNetCore.CAP.SqlServer
return new SqlServerStorageTransaction(this);
}
public Task<CapPublishedMessage> GetPublishedMessageAsync(int id)
public async Task<CapPublishedMessage> GetPublishedMessageAsync(int id)
{
var sql = $@"SELECT * FROM [{_options.Schema}].[Published] WITH (readpast) WHERE Id={id}";
using (var connection = new SqlConnection(_options.ConnectionString))
{
return connection.QueryFirstOrDefaultAsync<CapPublishedMessage>(sql);
return await connection.QueryFirstOrDefaultAsync<CapPublishedMessage>(sql);
}
}
......@@ -56,7 +57,7 @@ OUTPUT DELETED.MessageId,DELETED.[MessageType];";
// CapReceviedMessage
public Task StoreReceivedMessageAsync(CapReceivedMessage message)
public async Task StoreReceivedMessageAsync(CapReceivedMessage message)
{
if (message == null) throw new ArgumentNullException(nameof(message));
......@@ -66,16 +67,16 @@ VALUES(@Name,@Group,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";
using (var connection = new SqlConnection(_options.ConnectionString))
{
return connection.ExecuteAsync(sql, message);
await connection.ExecuteAsync(sql, message);
}
}
public Task<CapReceivedMessage> GetReceivedMessageAsync(int id)
public async Task<CapReceivedMessage> GetReceivedMessageAsync(int id)
{
var sql = $@"SELECT * FROM [{_options.Schema}].[Received] WITH (readpast) WHERE Id={id}";
using (var connection = new SqlConnection(_options.ConnectionString))
{
return connection.QueryFirstOrDefaultAsync<CapReceivedMessage>(sql);
return await connection.QueryFirstOrDefaultAsync<CapReceivedMessage>(sql);
}
}
......@@ -94,26 +95,30 @@ VALUES(@Name,@Group,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";
private async Task<IFetchedMessage> FetchNextMessageCoreAsync(string sql, object args = null)
{
using (var connection = new SqlConnection(_options.ConnectionString))
//here don't use `using` to dispose
var connection = new SqlConnection(_options.ConnectionString);
await connection.OpenAsync();
var transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted);
FetchedMessage fetchedMessage = null;
try
{
fetchedMessage = await connection.QueryFirstOrDefaultAsync<FetchedMessage>(sql, args, transaction);
}
catch (SqlException)
{
using (var transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted))
{
try
{
var fetched = await connection.QueryFirstOrDefaultAsync<FetchedMessage>(sql, args, transaction);
if (fetched == null)
return null;
return new SqlServerFetchedMessage(fetched.MessageId, fetched.MessageType, connection, transaction);
}
catch (Exception)
{
transaction.Rollback();
return null;
}
}
transaction.Dispose();
throw;
}
if (fetchedMessage == null)
{
transaction.Rollback();
transaction.Dispose();
connection.Dispose();
return null;
}
return new SqlServerFetchedMessage(fetchedMessage.MessageId, fetchedMessage.MessageType, connection, transaction);
}
}
}
\ No newline at end of file
src/DotNetCore.CAP.SqlServer/SqlServerStorageTransaction.cs
View file @ 6cae200e
......@@ -20,6 +20,7 @@ namespace DotNetCore.CAP.SqlServer
_schema = options.Schema;
_dbConnection = new SqlConnection(options.ConnectionString);
_dbConnection.Open();
_dbTransaction = _dbConnection.BeginTransaction(IsolationLevel.ReadCommitted);
}
......@@ -27,16 +28,16 @@ namespace DotNetCore.CAP.SqlServer
{
if (message == null) throw new ArgumentNullException(nameof(message));
var sql = $"UPDATE [{_schema}].[Published] SET [ExpiresAt] = @ExpiresAt,[StatusName]=@StatusName WHERE Id=@Id;";
_dbConnection.Execute(sql, message);
var sql = $"UPDATE [{_schema}].[Published] SET [Retries] = @Retries,[ExpiresAt] = @ExpiresAt,[StatusName]=@StatusName WHERE Id=@Id;";
_dbConnection.Execute(sql, message, _dbTransaction);
}
public void UpdateMessage(CapReceivedMessage message)
{
if (message == null) throw new ArgumentNullException(nameof(message));
var sql = $"UPDATE [{_schema}].[Received] SET [ExpiresAt] = @ExpiresAt,[StatusName]=@StatusName WHERE Id=@Id;";
_dbConnection.Execute(sql, message);
var sql = $"UPDATE [{_schema}].[Received] SET [Retries] = @Retries,[ExpiresAt] = @ExpiresAt,[StatusName]=@StatusName WHERE Id=@Id;";
_dbConnection.Execute(sql, message, _dbTransaction);
}
public void EnqueueMessage(CapPublishedMessage message)
......@@ -44,7 +45,7 @@ namespace DotNetCore.CAP.SqlServer
if (message == null) throw new ArgumentNullException(nameof(message));
var sql = $"INSERT INTO [{_schema}].[Queue] values(@MessageId,@MessageType);";
_dbConnection.Execute(sql, new CapQueue { MessageId = message.Id, MessageType = MessageType.Publish });
_dbConnection.Execute(sql, new CapQueue { MessageId = message.Id, MessageType = MessageType.Publish }, _dbTransaction);
}
public void EnqueueMessage(CapReceivedMessage message)
......@@ -52,7 +53,7 @@ namespace DotNetCore.CAP.SqlServer
if (message == null) throw new ArgumentNullException(nameof(message));
var sql = $"INSERT INTO [{_schema}].[Queue] values(@MessageId,@MessageType);";
_dbConnection.Execute(sql, new CapQueue { MessageId = message.Id, MessageType = MessageType.Subscribe });
_dbConnection.Execute(sql, new CapQueue { MessageId = message.Id, MessageType = MessageType.Subscribe }, _dbTransaction);
}
public Task CommitAsync()
......
src/DotNetCore.CAP/IConsumerHandler.Default.cs
View file @ 6cae200e
......@@ -113,7 +113,7 @@ namespace DotNetCore.CAP
var messageStore = provider.GetRequiredService<IStorageConnection>();
var receivedMessage = new CapReceivedMessage(messageContext)
{
StatusName = StatusName.Enqueued,
StatusName = StatusName.Scheduled,
};
messageStore.StoreReceivedMessageAsync(receivedMessage).Wait();
return receivedMessage;
......
src/DotNetCore.CAP/IQueueExecutor.Publish.Base.cs
View file @ 6cae200e
......@@ -24,56 +24,53 @@ namespace DotNetCore.CAP
public async Task<OperateResult> ExecuteAsync(IStorageConnection connection, IFetchedMessage fetched)
{
using (fetched)
var message = await connection.GetPublishedMessageAsync(fetched.MessageId);
try
{
var message = await connection.GetPublishedMessageAsync(fetched.MessageId);
try
{
var sp = Stopwatch.StartNew();
await _stateChanger.ChangeStateAsync(message, new ProcessingState(), connection);
var sp = Stopwatch.StartNew();
await _stateChanger.ChangeStateAsync(message, new ProcessingState(), connection);
if (message.Retries > 0)
{
_logger.JobRetrying(message.Retries);
}
var result = await PublishAsync(message.Name, message.Content);
sp.Stop();
if (message.Retries > 0)
{
_logger.JobRetrying(message.Retries);
}
var result = await PublishAsync(message.Name, message.Content);
sp.Stop();
var newState = default(IState);
if (!result.Succeeded)
var newState = default(IState);
if (!result.Succeeded)
{
var shouldRetry = await UpdateJobForRetryAsync(message, connection);
if (shouldRetry)
{
var shouldRetry = await UpdateJobForRetryAsync(message, connection);
if (shouldRetry)
{
newState = new ScheduledState();
_logger.JobFailedWillRetry(result.Exception);
}
else
{
newState = new FailedState();
_logger.JobFailed(result.Exception);
}
newState = new ScheduledState();
_logger.JobFailedWillRetry(result.Exception);
}
else
{
newState = new SucceededState();
newState = new FailedState();
_logger.JobFailed(result.Exception);
}
await _stateChanger.ChangeStateAsync(message, newState, connection);
fetched.RemoveFromQueue();
}
else
{
newState = new SucceededState();
}
await _stateChanger.ChangeStateAsync(message, newState, connection);
if (result.Succeeded)
{
_logger.JobExecuted(sp.Elapsed.TotalSeconds);
}
fetched.RemoveFromQueue();
return OperateResult.Success;
}
catch (Exception ex)
if (result.Succeeded)
{
_logger.ExceptionOccuredWhileExecutingJob(message?.Name, ex);
return OperateResult.Failed(ex);
_logger.JobExecuted(sp.Elapsed.TotalSeconds);
}
return OperateResult.Success;
}
catch (Exception ex)
{
_logger.ExceptionOccuredWhileExecutingJob(message?.Name, ex);
return OperateResult.Failed(ex);
}
}
......@@ -81,7 +78,7 @@ namespace DotNetCore.CAP
{
var retryBehavior = RetryBehavior.DefaultRetry;
var now = DateTime.UtcNow;
var now = DateTime.Now;
var retries = ++message.Retries;
if (retries >= retryBehavior.RetryCount)
{
......
src/DotNetCore.CAP/IQueueExecutor.Subscibe.cs
View file @ 6cae200e
......@@ -37,61 +37,58 @@ namespace DotNetCore.CAP
public async Task<OperateResult> ExecuteAsync(IStorageConnection connection, IFetchedMessage fetched)
{
using (fetched)
var message = await connection.GetReceivedMessageAsync(fetched.MessageId);
try
{
var message = await connection.GetReceivedMessageAsync(fetched.MessageId);
try
{
var sp = Stopwatch.StartNew();
await _stateChanger.ChangeStateAsync(message, new ProcessingState(), connection);
var sp = Stopwatch.StartNew();
await _stateChanger.ChangeStateAsync(message, new ProcessingState(), connection);
if (message.Retries > 0)
{
_logger.JobRetrying(message.Retries);
}
var result = await ExecuteSubscribeAsync(message);
sp.Stop();
if (message.Retries > 0)
{
_logger.JobRetrying(message.Retries);
}
var result = await ExecuteSubscribeAsync(message);
sp.Stop();
var newState = default(IState);
if (!result.Succeeded)
var newState = default(IState);
if (!result.Succeeded)
{
var shouldRetry = await UpdateJobForRetryAsync(message, connection);
if (shouldRetry)
{
var shouldRetry = await UpdateJobForRetryAsync(message, connection);
if (shouldRetry)
{
newState = new ScheduledState();
_logger.JobFailedWillRetry(result.Exception);
}
else
{
newState = new FailedState();
_logger.JobFailed(result.Exception);
}
newState = new ScheduledState();
_logger.JobFailedWillRetry(result.Exception);
}
else
{
newState = new SucceededState();
}
await _stateChanger.ChangeStateAsync(message, newState, connection);
fetched.RemoveFromQueue();
if (result.Succeeded)
{
_logger.JobExecuted(sp.Elapsed.TotalSeconds);
newState = new FailedState();
_logger.JobFailed(result.Exception);
}
return OperateResult.Success;
}
catch (SubscriberNotFoundException ex)
else
{
_logger.LogError(ex.Message);
return OperateResult.Failed(ex);
newState = new SucceededState();
}
catch (Exception ex)
await _stateChanger.ChangeStateAsync(message, newState, connection);
fetched.RemoveFromQueue();
if (result.Succeeded)
{
_logger.ExceptionOccuredWhileExecutingJob(message?.Name, ex);
return OperateResult.Failed(ex);
_logger.JobExecuted(sp.Elapsed.TotalSeconds);
}
return OperateResult.Success;
}
catch (SubscriberNotFoundException ex)
{
_logger.LogError(ex.Message);
return OperateResult.Failed(ex);
}
catch (Exception ex)
{
_logger.ExceptionOccuredWhileExecutingJob(message?.Name, ex);
return OperateResult.Failed(ex);
}
}
......
test/DotNetCore.CAP.EntityFrameworkCore.Test/DotNetCore.CAP.EntityFrameworkCore.Test.csproj
View file @ 6cae200e
......@@ -18,7 +18,7 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\DotNetCore.CAP.EntityFrameworkCore\DotNetCore.CAP.SqlServer.csproj" />
<ProjectReference Include="..\..\src\DotNetCore.CAP.SqlServer\DotNetCore.CAP.SqlServer.csproj" />
<ProjectReference Include="..\..\src\DotNetCore.CAP\DotNetCore.CAP.csproj" />
</ItemGroup>
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment