Skip to content

Commit 6c242bc

Browse files
HaikAsatryanHaikAsatryan
authored
Merge pull request #8 from PandaTechAM/development
Fixed consumer to work without outbox pattern as well
2 parents 99517da + f1629e8 commit 6c242bc

File tree

3 files changed

+70
-61
lines changed

3 files changed

+70
-61
lines changed

‎src/MassTransit.PostgresOutbox/Abstractions/InboxConsumer.cs

+65-56
Original file line numberDiff line numberDiff line change
@@ -9,72 +9,81 @@
99
namespace MassTransit.PostgresOutbox.Abstractions;
1010

1111
public abstract class InboxConsumer<TMessage, TDbContext> : IConsumer<TMessage>
12-
where TMessage : class
13-
where TDbContext : DbContext, IInboxDbContext
12+
where TMessage : class
13+
where TDbContext : DbContext, IInboxDbContext
1414
{
15-
private readonly string _consumerId;
16-
private readonly IServiceScopeFactory _serviceScopeFactory;
15+
private readonly string _consumerId;
16+
private readonly IServiceScopeFactory _serviceScopeFactory;
1717

18-
protected InboxConsumer(IServiceScopeFactory serviceScopeFactory)
19-
{
20-
_consumerId = GetType().ToString();
21-
_serviceScopeFactory = serviceScopeFactory;
22-
}
18+
protected InboxConsumer(IServiceScopeFactory serviceScopeFactory)
19+
{
20+
_consumerId = GetType().ToString();
21+
_serviceScopeFactory = serviceScopeFactory;
22+
}
2323

24-
public async Task Consume(ConsumeContext<TMessage> context)
25-
{
26-
using var scope = _serviceScopeFactory.CreateScope();
27-
var messageId = context.Headers.Get<Guid>(Constants.OutboxMessageId);
24+
public async Task Consume(ConsumeContext<TMessage> context)
25+
{
26+
var messageId = context.Headers.Get<Guid>(Constants.OutboxMessageId);
2827

29-
var dbContext = scope.ServiceProvider.GetRequiredService<TDbContext>();
30-
var logger = scope.ServiceProvider.GetRequiredService<ILogger<InboxConsumer<TMessage, TDbContext>>>();
28+
if (messageId is null)
29+
{
30+
await Consume(context.Message);
31+
return;
32+
}
3133

32-
var exists = await dbContext.InboxMessages.AnyAsync(x => x.MessageId == messageId && x.ConsumerId == _consumerId);
34+
using var scope = _serviceScopeFactory.CreateScope();
3335

34-
if (!exists)
35-
{
36-
dbContext.InboxMessages.Add(new InboxMessage
37-
{
38-
MessageId = messageId!.Value,
39-
CreatedAt = DateTime.UtcNow,
40-
State = MessageState.New,
41-
ConsumerId = _consumerId,
42-
});
36+
var dbContext = scope.ServiceProvider.GetRequiredService<TDbContext>();
37+
var logger = scope.ServiceProvider.GetRequiredService<ILogger<InboxConsumer<TMessage, TDbContext>>>();
4338

44-
await dbContext.SaveChangesAsync();
45-
}
39+
var exists =
40+
await dbContext.InboxMessages.AnyAsync(x => x.MessageId == messageId && x.ConsumerId == _consumerId);
4641

47-
using var transactionScope = await dbContext.Database.BeginTransactionAsync(System.Data.IsolationLevel.ReadCommitted);
42+
if (!exists)
43+
{
44+
dbContext.InboxMessages.Add(new InboxMessage
45+
{
46+
MessageId = messageId.Value,
47+
CreatedAt = DateTime.UtcNow,
48+
State = MessageState.New,
49+
ConsumerId = _consumerId,
50+
});
4851

49-
var inboxMessage = await dbContext.InboxMessages
50-
.Where(x => x.MessageId == messageId)
51-
.Where(x => x.ConsumerId == _consumerId)
52-
.Where(x => x.State == MessageState.New)
53-
.ForUpdate(LockBehavior.SkipLocked)
54-
.FirstOrDefaultAsync();
52+
await dbContext.SaveChangesAsync();
53+
}
5554

56-
if (inboxMessage == null)
57-
{
58-
return;
59-
}
55+
await using var transactionScope =
56+
await dbContext.Database.BeginTransactionAsync(System.Data.IsolationLevel.ReadCommitted);
6057

61-
try
62-
{
63-
await Consume(context.Message);
64-
inboxMessage.State = MessageState.Done;
65-
}
66-
catch (Exception ex)
67-
{
68-
logger.LogError(ex, "Exception thrown while consuming message");
69-
throw;
70-
}
71-
finally
72-
{
73-
inboxMessage!.UpdatedAt = DateTime.UtcNow;
74-
await dbContext.SaveChangesAsync();
75-
await transactionScope.CommitAsync();
76-
}
77-
}
58+
var inboxMessage = await dbContext.InboxMessages
59+
.Where(x => x.MessageId == messageId)
60+
.Where(x => x.ConsumerId == _consumerId)
61+
.Where(x => x.State == MessageState.New)
62+
.ForUpdate(LockBehavior.SkipLocked)
63+
.FirstOrDefaultAsync();
7864

79-
public abstract Task Consume(TMessage message);
65+
if (inboxMessage == null)
66+
{
67+
return;
68+
}
69+
70+
try
71+
{
72+
await Consume(context.Message);
73+
inboxMessage.State = MessageState.Done;
74+
}
75+
catch (Exception ex)
76+
{
77+
logger.LogError(ex, "Exception thrown while consuming message");
78+
throw;
79+
}
80+
finally
81+
{
82+
inboxMessage.UpdatedAt = DateTime.UtcNow;
83+
await dbContext.SaveChangesAsync();
84+
await transactionScope.CommitAsync();
85+
}
86+
}
87+
88+
protected abstract Task Consume(TMessage message);
8089
}

‎src/MassTransit.PostgresOutbox/MassTransit.PostgresOutbox.csproj

+4-4
Original file line numberDiff line numberDiff line change
@@ -8,13 +8,13 @@
88
<PackageReadmeFile>Readme.md</PackageReadmeFile>
99
<Authors>Pandatech</Authors>
1010
<Copyright>MIT</Copyright>
11-
<Version>1.0.5</Version>
11+
<Version>1.0.6</Version>
1212
<PackageId>Pandatech.MassTransit.PostgresOutbox</PackageId>
1313
<Title>Pandatech MassTransit PostgreSQL Outbox Extension</Title>
1414
<PackageTags>Pandatech, library, postgres, distributed systems, microservices, modular monolith, messaging, efcore, mass transit, outbox pattern, inbox pattern</PackageTags>
1515
<Description>Pandatech.MassTransit.PostgresOutbox extends MassTransit to offer advanced message handling capabilities for distributed systems. With first-class support for multiple DbContexts, this library integrates seamlessly with Entity Framework Core and PostgreSQL, providing reliable Outbox and Inbox patterns. It ensures consistent message delivery and processing in complex microservices architectures, leveraging PostgreSQL's ForUpdate feature to handle concurrency with ease.</Description>
1616
<RepositoryUrl>https://github.com/PandaTechAM/be-lib-pandatech-masstransit-postgres-outbox</RepositoryUrl>
17-
<PackageReleaseNotes>Readme update</PackageReleaseNotes>
17+
<PackageReleaseNotes>Fixed consumer to work without outbox pattern as well</PackageReleaseNotes>
1818
</PropertyGroup>
1919

2020
<ItemGroup>
@@ -24,8 +24,8 @@
2424

2525
<ItemGroup>
2626
<PackageReference Include="MassTransit" Version="8.2.2" />
27-
<PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="8.0.5" />
28-
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="8.0.5" />
27+
<PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="8.0.6" />
28+
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="8.0.6" />
2929
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
3030
<PackageReference Include="Pandatech.EFCore.PostgresExtensions" Version="2.0.1" />
3131
</ItemGroup>

‎test/PandaNuGet.Demo/MassTransit.PostgresOutbox.Demo.csproj

+1-1
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
</PropertyGroup>
99

1010
<ItemGroup>
11-
<PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="8.0.5" />
11+
<PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="8.0.6" />
1212
<PackageReference Include="Swashbuckle.AspNetCore" Version="6.6.2" />
1313
</ItemGroup>
1414

0 commit comments

Comments
 (0)