elias/booking-microservices
1
0
Fork 0
mirror of https://github.com/meysamhadeli/booking-microservices.git synced 2026-04-18 09:52:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #137 from meysamhadeli/develop

Browse Source 
Develop
This commit is contained in:
Meysam Hadeli 2023-01-29 23:20:02 +03:30 committed by GitHub
commit 5cb5dad0ac
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 21 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/BuildingBlocks/PersistMessageProcessor/Data/PersistMessageDbContext.cs
View File

@ -58,16 +58,16 @@ public class PersistMessageDbContext : DbContext, IPersistMessageDbContext
{ {
foreach (var entry in ex.Entries) foreach (var entry in ex.Entries)
{ {
var currentValues = entry.CurrentValues;
var databaseValues = await entry.GetDatabaseValuesAsync(cancellationToken); var databaseValues = await entry.GetDatabaseValuesAsync(cancellationToken);
_logger.LogInformation( if (databaseValues == null)
"Entry to entity with database-value: {@databaseValues} and current-value: {@currentValues}", {
databaseValues, _logger.LogError("The record no longer exists in the database, The record has been deleted by another user.");
currentValues); throw;
}
// Refresh the original values with current values // Refresh the original values to bypass next concurrency check
entry.OriginalValues.SetValues(currentValues); entry.OriginalValues.SetValues(databaseValues);
} }
return await base.SaveChangesAsync(cancellationToken); return await base.SaveChangesAsync(cancellationToken);

15
src/BuildingBlocks/PersistMessageProcessor/PersistMessageProcessor.cs
View File

@ -18,6 +18,7 @@ public class PersistMessageProcessor : IPersistMessageProcessor
private readonly IMediator _mediator; private readonly IMediator _mediator;
private readonly IPersistMessageDbContext _persistMessageDbContext; private readonly IPersistMessageDbContext _persistMessageDbContext;
private readonly IPublishEndpoint _publishEndpoint; private readonly IPublishEndpoint _publishEndpoint;
private SemaphoreSlim Semaphore => new SemaphoreSlim(1);
public PersistMessageProcessor( public PersistMessageProcessor(
ILogger<PersistMessageProcessor> logger, ILogger<PersistMessageProcessor> logger,
@ -115,7 +116,19 @@ public class PersistMessageProcessor : IPersistMessageProcessor
.Where(x => x.MessageStatus != MessageStatus.Processed) .Where(x => x.MessageStatus != MessageStatus.Processed)
.ToListAsync(cancellationToken); .ToListAsync(cancellationToken);
foreach (var message in messages) await ProcessAsync(message.Id, message.DeliveryType, cancellationToken); foreach (var message in messages)
{
await Semaphore.WaitAsync(cancellationToken);
try
{
await ProcessAsync(message.Id, message.DeliveryType, cancellationToken);
}
finally
{
Semaphore.Release();
}
}
} }
public async Task ProcessInboxAsync(long messageId, CancellationToken cancellationToken = default) public async Task ProcessInboxAsync(long messageId, CancellationToken cancellationToken = default)